GatewayWorkerの概要と特徴

GatewayWorkerは、Workermanをベースに構築された分散型リアルタイム通信フレームワークです。TCP長接続を前提として設計されており、メッセージのユニキャスト（単体送信）、マルチキャスト（グループ送信）、ブロードキャスト（全体送信）といった機能を標準で備えています。また、MySQLクライアントを内蔵しており、GatewayプロセスとWorkerプロセスを分離することで、大規模な同時接続数にも耐えうるスケーラビリティを実現しています。

システムアーキテクチャの動作原理

1. プロセスの起動: Register（登録）、Gateway（ゲートウェイ）、BusinessWorker（ビジネスロジック）の各サービスを起動します。
2. サービス登録: GatewayとBusinessWorkerは、起動時にRegisterサービスへ自身の通信アドレスを通知します。
3. アドレスの同期: Registerは、現在アクティブなGatewayのアドレス一覧をBusinessWorkerへ配信します。
4. 内部接続の確立: BusinessWorkerは取得したアドレスを基に、Gatewayとの間に内部通信用コネクションを確立します。
5. クライアント処理: クライアントからの接続やデータ送信はGatewayが受け取り、確立された内部通信路を通じてBusinessWorkerに転送され、実際の業務ロジックが実行されます。

環境構築とライブラリの導入

プロジェクトのルートディレクトリにて、Composerを使用して必要なパッケージをインストールします。

# ThinkPHP向けWorkerman拡張の導入
composer require topthink/think-worker

# GatewayWorker本体の導入
composer require workerman/gateway-worker

サーバーサイドの実装

通常、Applications/Service などのディレクトリを作成し、役割ごとの起動スクリプトを用意します。

1. ゲートウェイプロセス (start_gateway.php)

クライアントとの直接的な接続を管理します。

<?php
use Workerman\Worker;
use GatewayWorker\Gateway;

require_once __DIR__ . '/../../vendor/autoload.php';

// WebSocketプロトコルを指定（ポート: 9501）
$gwProc = new Gateway("websocket://0.0.0.0:9501");

$gwProc->name = 'AppGateway';
$gwProc->count = 4;
$gwProc->lanIp = '127.0.0.1';
$gwProc->startPort = 2000;
$gwProc->registerAddress = '127.0.0.1:1236';

if(!defined('GLOBAL_START')) {
    Worker::runAll();
}

2. ビジネスワーカープロセス (start_businessworker.php)

実際のビジネスロジックを実行するプロセスです。

<?php
use Workerman\Worker;
use GatewayWorker\BusinessWorker;

require_once __DIR__ . '/../../vendor/autoload.php';

$bizProc = new BusinessWorker();
$bizProc->name = 'AppBusinessWorker';
$bizProc->count = 4;
$bizProc->registerAddress = '127.0.0.1:1236';

if(!defined('GLOBAL_START')) {
    Worker::runAll();
}

3. レジスタープロセス (start_register.php)

各プロセス間のコーディネートを行います。

<?php
use Workerman\Worker;
use GatewayWorker\Register;

require_once __DIR__ . '/../../vendor/autoload.php';

$regProc = new Register('text://0.0.0.0:1236');

if(!defined('GLOBAL_START')) {
    Worker::runAll();
}

4. イベントハンドラ (Events.php)

接続時やメッセージ受信時の具体的な挙動を定義します。

<?php
use GatewayWorker\Lib\Gateway;

class Events {
    public static function onConnect($clientId) {
        $initMsg = json_encode([
            'type' => 'init',
            'client_id' => $clientId,
            'content' => 'Connected successfully.'
        ]);
        Gateway::sendToClient($clientId, $initMsg);
    }

    public static function onMessage($clientId, $rawMessage) {
        $payload = json_decode($rawMessage, true);
        $broadcastData = json_encode([
            'from' => $clientId,
            'msg' => $payload['text'] ?? '',
            'timestamp' => time()
        ]);
        Gateway::sendToAll($broadcastData);
    }

    public static function onClose($clientId) {
        // 切断時の処理
    }
}

SSL環境（WSS）への対応

HTTPS環境でWebSocketを使用する場合、ブラウザのセキュリティ制限によりwss://プロトコルが必須となります。Nginxをリバースプロキシとして構成するのが一般的です。

location /wss {
    proxy_pass http://127.0.0.1:9501;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "Upgrade";
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
}

uni-appによるクライアント実装

フロントエンドでは、接続管理、心跳（ハートビート）、自動再接続のロジックを組み込みます。

export default {
    data() {
        return {
            socketInstance: null,
            reconnectTimer: null,
            heartbeatTimer: null
        }
    },
    methods: {
        connectService() {
            this.socketInstance = uni.connectSocket({
                url: 'wss://your-domain.com/wss',
                success: () => console.log('Socket initialized')
            });

            this.socketInstance.onOpen(() => {
                console.log('WebSocket opened');
                this.startHeartbeat();
            });

            this.socketInstance.onMessage((res) => {
                const data = JSON.parse(res.data);
                console.log('Received:', data);
            });

            this.socketInstance.onClose(() => {
                console.warn('Socket closed');
                this.attemptReconnect();
            });
        },
        startHeartbeat() {
            clearInterval(this.heartbeatTimer);
            this.heartbeatTimer = setInterval(() => {
                this.socketInstance.send({
                    data: JSON.stringify({ type: 'ping' })
                });
            }, 30000);
        },
        attemptReconnect() {
            if (this.reconnectTimer) return;
            this.reconnectTimer = setTimeout(() => {
                console.log('Reconnecting...');
                this.connectService();
                this.reconnectTimer = null;
            }, 5000);
        }
    }
}