問題描述

請問下各位神仙哥哥姐姐，這段代碼中的定時(shí)器為什么沒有正確的讀取到客戶端鏈接信息？
幫忙檢閱看看，跪拜
日志：
連接ID: 2
收到客戶端消息: {"type":"sub","params":"market_ti"}
訂閱主題: market_ti
定時(shí)任務(wù)：廣播最新數(shù)據(jù)
當(dāng)前活躍客戶端數(shù): 2
檢查客戶端 ID: 1
客戶端 1 未在連接列表中

// Redis 配置
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);
// 創(chuàng)建 WebSocket Worker
$ws_worker = new Worker("websocket://0.0.0.0:2346", $context);
$ws_worker->transport = 'ssl';
$ws_worker->count = 1;

// 客戶端連接邏輯
$ws_worker->onConnect = function ($connection) use ($redis, $httpClient, $ws_worker) {
    echo "新連接加入\n";
    echo "new connection from ip " . $connection->getRemoteIp() . "\n";
    echo "連接ID: {$connection->id}\n";

    // 在連接時(shí)記錄連接ID
    $redis->sAdd("client:{$connection->id}:connected", true);  // 確保客戶端連接的標(biāo)識被保存
    $ws_worker->connections[$connection->id] = $connection;

    // 消息處理
    $connection->onMessage = function ($connection, $message) use ($redis, $httpClient) {
        echo "收到客戶端消息: $message\n";
        $msg = json_decode($message, true);
        if ($msg && isset($msg['type'], $msg['params']) && $msg['type'] === 'sub') {
            $theme = $msg['params'];
            echo "訂閱主題: $theme\n";

            // 將客戶端連接和主題存入 Redis
            $redis->sAdd("themes:$theme:clients", $connection->id);
            $redis->sAdd("client:$connection->id:themes", $theme);

            // 返回歷史數(shù)據(jù)
            $historyData = ($theme === 'market_ti') 
                ? fetchMergedallData($httpClient)
                : fetchKlineData($theme, $httpClient, 2);

            $connection->send(json_encode([
                'type' => 'history',
                'data' => $historyData,
            ]));
        }
    };

    // 斷開連接處理
    $ws_worker->onClose = function ($connection) use ($redis, $ws_worker) {
        echo "連接斷開\n";

        // 從 Redis 清除相關(guān)數(shù)據(jù)
        $themes = $redis->sMembers("client:$connection->id:themes");
        foreach ($themes as $theme) {
            $redis->sRem("themes:$theme:clients", $connection->id);
        }
        $redis->del("client:$connection->id:themes");
        $redis->del("client:$connection->id:connected");

        // 從 WebSocket worker 的連接列表中刪除該連接
        unset($ws_worker->connections[$connection->id]);
    };
};

// 定時(shí)任務(wù)：每 10 秒廣播數(shù)據(jù)
Timer::add(10, function () use ($redis, $httpClient, $ws_worker) {
    echo "定時(shí)任務(wù)：廣播最新數(shù)據(jù)\n";

    // 獲取所有主題的客戶端連接
    $themes = $redis->keys("themes:*:clients");

    foreach ($themes as $themeKey) {
        $theme = str_replace(['themes:', ':clients'], '', $themeKey);
        if ($theme === 'market_ti') {
            $data = fetchMergedallData($httpClient);
        } else {
            $data = fetchKlineData($theme, $httpClient, 1);
            $depthData = fetchdepthData($theme, $httpClient);
        }

        // 獲取主題下的所有客戶端
        $clients = $redis->sMembers($themeKey);
        echo "當(dāng)前活躍客戶端數(shù): " . count($clients) . "\n";  // 輸出當(dāng)前主題下的客戶端數(shù)

        foreach ($clients as $clientId) {
            echo "檢查客戶端 ID: $clientId\n";

            // 從 WebSocket 連接列表中獲取客戶端連接
            if (isset($ws_worker->connections[$clientId])) {
                $connection = $ws_worker->connections[$clientId];
                if ($connection->isOpened()) {
                    echo "找到客戶端 $clientId\n";
                    $connection->send(json_encode([
                        'type' => 'update',
                        'kline' => $data,
                        'depth' => $depthData,
                    ]));
                } else {
                    echo "客戶端 $clientId 不在線，無法發(fā)送數(shù)據(jù)\n";  // 客戶端不在線
                }
            } else {
                echo "客戶端 $clientId 未在連接列表中\(zhòng)n";  // 客戶端未找到
            }
        }
    }
});