先梳理一下代碼整體的邏輯吧

代碼有bug吧。
1、你代碼里用的self::$ws_connection，就是所有前端用戶共用一個(gè) self::$ws_connection，但是你代碼里又是在onMessage里不斷重置 self::$ws_connection。應(yīng)該是在onWorkerStart里設(shè)置self::$ws_connection才對(duì)。

2、沒(méi)有人保證連接永遠(yuǎn)不斷開(kāi)，所以要在onClose里加一個(gè)reconnect()邏輯，斷開(kāi)自動(dòng)重連。前端代碼也是一樣。

其他都就是連接數(shù)超過(guò)1024要裝event擴(kuò)展，優(yōu)化linux內(nèi)核

onWorkerStart干的事情

/**
 * 進(jìn)程啟動(dòng)時(shí)
 * @param Worker $worker
 * @return void
 */
public function onWorkerStart(Worker $worker): void
{
    try {
        $this->startChannel();
        $this->startCrontab();

        $this->startPublic();
        $this->startPrivate();
        //實(shí)時(shí)交易
        TradeOrderHelper::start($this->websocket_api->getConfig());
    } catch (Throwable $throwable) {
        print_r(['【onWorkerStart系統(tǒng)異?！? . date('Y-m-d H:i:s') . ' 文件：' . __FILE__ . ' 行號(hào)：' . __LINE__, $throwable->getMessage()]);
        sleep(2);
        Worker::stopAll();
    }
}

啟動(dòng)公共連接：

/**
 * @return void
 * @throws Exception
 */
private function startPublic(): void
{
    [$http, $public, $private] = $this->service_url;
    $connection = new AsyncTcpConnection($public, $this->getContextOption());
    $connection->setLabel(AsyncTcpConnection::LABEL_PUBLIC);
    $this->bindConnection($connection);
    $connection->connect();
}

設(shè)置異步連接：

/**
 * 設(shè)置異步連接
 * @param AsyncTcpConnection $con
 * @return void
 */
private function bindConnection(AsyncTcpConnection $con): void
{
    $con->setStartTime(time());
    $con->account_name = $this->websocket_api->getConfig()->title;
    $con->api_key = $this->websocket_api->getConfig()->api_key;
    // 設(shè)置以ssl加密方式訪問(wèn)
    $con->transport = 'ssl';
    $con->onError = function (AsyncTcpConnection $connection, $err_code, $err_msg) {
        //echo "$err_code, $err_msg\n";
    };
    $con->onClose = function (AsyncTcpConnection $connection) {
    };
    $con->onConnect = function (AsyncTcpConnection $con) {
        $con->send(json_encode($this->websocket_api->login()));
        $con->send('ping');
    };
    $con->onMessage = function (AsyncTcpConnection $con, $data) {
        if (!$data) {
            return;
        }
        try {
            switch ($data) {
                case 'pong':    //心跳回應(yīng)
                    $filename = 'pong_' . $con->getLabel() . '_' . $con->api_key . '.log';
                    $account_name = $this->websocket_api->getConfig()->title;
                    file_put_contents($this->logsPath() . $filename, $account_name . ' 最后收到pong：' . date('Y-m-d H:i:s'));
                    $con->setLastPongTime(time());
                    break;
                default:
                    $event = json_decode($data, true);
                    if (isset($event['event'])) {
                        //事件
                        //WssMessageService::listenEvent($event, $data, $this->websocket_api, $con);
                        $handler = EventFactory::make($event);
                        $handler->handle($event, $this->websocket_api, $con);
                    } else if (isset($event['arg']['channel'])) {
                        //頻道
                        //WssMessageService::listenChannel($event, $data, $this->websocket_api, $con);
                        $handler = ChannelFactory::make($event);
                        $handler->handle($event, $this->websocket_api, $con);
                    } else if (isset($event['id']) && isset($event['op'])) {
                        //交易
                        //WssMessageService::listenOp($event, $data, $this->websocket_api, $con);
                        $handler = OperationFactory::make($event);
                        $handler->handle($event, $this->websocket_api, $con);
                    }
                    break;
            }
        } catch (Throwable $throwable) {
            fileDebug([($con->isPublicChannel() ? '【公共頻道】' : '【私有頻道】') . date('Y-m-d H:i:s'), '【onMessage業(yè)務(wù)異?！?文件：' . __FILE__ . ' 行號(hào)：' . __LINE__, $throwable->getMessage(), $data], 'logs', 'onMessage_Throwable');
        }
    };
}

斷線重連邏輯：

/**
 * 檢查ping狀態(tài)，超時(shí)斷線重連
 * @param AsyncTcpConnection $con
 * @return void
 * @throws Exception
 */
private function checkStatus(AsyncTcpConnection $con): void
{
    try {
        $ttl = config('okx.ping_interval', 20);
        $last_pong_time = $con->getLastPongTime();
        $start_time = $con->getStartTime();
        if (empty($last_pong_time) && ($start_time + $ttl > time())) {
            //剛啟動(dòng)，還沒(méi)收到pong
            return;
        }

        //超時(shí)判斷
        if (($last_pong_time + $ttl * 1.3) < time()) {
            RobotNotify::send('ping狀態(tài)異常正在重連', 'checkStatus：' . $con->getLabel() . PHP_EOL . ($con->account_name ?? ''));
            //重連
            $con->close();
            if ($con->isPrivateChannel()) {
                $this->startPrivate();
            } else {
                $this->startPublic();
            }
        }
    } catch (Throwable $throwable) {
        //通知開(kāi)發(fā)者
        RobotNotify::send('重連異常', 'reconnect：' . $con->getLabel() . PHP_EOL . ($con->account_name ?? '') . PHP_EOL . $throwable->getMessage());
        Worker::stopAll();
    }
}

你看看這個(gè)http://www.wtbis.cn/q/11278