弄個(gè)類，存儲(chǔ)在靜態(tài)變量里就行了

class MyConnections {
    public static connections = [];
}

存的時(shí)候

MyConnections::connection[標(biāo)記] = $connection;

起始按照你截圖的用法，存$connection沒什么意義。

class Worker extends Server
{
    protected $processes = 1;

    protected function init()
    {
        $ws = new \Workerman\Worker('websocket://0.0.0.0:2000');
        // 啟動(dòng)4個(gè)進(jìn)程對(duì)外提供服務(wù)
        $ws->count = 1;
        $ws->onConnect = function (TcpConnection $connection){
        };

        $ws->onMessage = function (TcpConnection $connection,$data) {
            $connection->send('aaaa');
        };
        $ws->onClose = function (TcpConnection $connection) {
            var_dump('ws斷開');
        };

        parent::init(); // TODO: Change the autogenerated stub
    }

    public static $i = 0;

    /**
     * http服務(wù)
     * @param $connection
     * @param $data
     */
    public function onMessage($connection, $data)
    {
        //pc管理后臺(tái)新建用戶會(huì)請(qǐng)求這個(gè)http服務(wù)，收到新注冊(cè)用戶消息，為用戶創(chuàng)建ws鏈接
//        var_dump($data['get']['ws_url']);
        $con = new AsyncTcpConnection('ws://127.0.0.1:2000');

        //開始鏈接
        $con->onConnect = function ($con) {

        };

        //鏈接成功
        $con->onWebSocketConnect = function (AsyncTcpConnection $connection) {
            $connection->uid = ++ self::$i;
            //1更新用戶在線狀態(tài)

            //2啟動(dòng)心跳定時(shí)器
            $connection->timer_id = Timer::add(10, function () use($connection) {
                $ping = [
                    'msgId' => 122333,
                    'type'  => 'common.topic.clientheartbeat',
                    'msg'   => '',
                ];
                $connection->send(json_encode($ping, JSON_UNESCAPED_UNICODE));
            });
        };

        //收到消息
        $con->onMessage = function (AsyncTcpConnection $connection, $data) {
            var_dump($connection->uid.':收到消息');
//            $data = json_decode($data, true);
//            //服務(wù)端心跳
//            if ($data['heartbeat']) {
//                return;
//            }
//
//            $msgId = $data['msgId'];
//            switch ($data['type']) {
//                case 'authorize.topic.dismissnotify':
//                    echo '解除門店授權(quán)';
//
//                    break;
//                case 'product.topic.flowstatuschange':
//                    echo '商品狀態(tài)變更';
//
//                    break;
//                default:
//                    break;
//            }
//
//            //回復(fù)
//            $ack = [
//                'msgId' => $msgId,
//                'type'  => '',
//                'msg'   => 'ack'
//            ];
//            $connection->send(json_encode($ack, JSON_UNESCAPED_UNICODE));
        };

        //鏈接關(guān)閉
        $con->onClose = function (AsyncTcpConnection $connection) {
            echo 'onClose' . PHP_EOL;
            //1刪除心跳定時(shí)器
            if (isset($connection->timer_id)) {
                Timer::del($connection->timer_id);
            }
            //2更新用戶下線狀態(tài)
        };

        //鏈接失敗
        $con->onError = function (AsyncTcpConnection $connection, $code, $msg) {
        };

        //鏈接
        $con->connect();
    }

    /**
     * 當(dāng)連接建立時(shí)觸發(fā)的回調(diào)函數(shù)
     * @param $connection
     */
    public function onConnect($connection)
    {

    }

    /**
     * 當(dāng)連接斷開時(shí)觸發(fā)的回調(diào)函數(shù)
     * @param $connection
     */
    public function onClose($connection)
    {

    }

    /**
     * 當(dāng)客戶端的連接上發(fā)生錯(cuò)誤時(shí)觸發(fā)
     * @param $connection
     * @param $code
     * @param $msg
     */
    public function onError($connection, $code, $msg)
    {
        echo "error $code $msg\n";
    }

    /**
     * 每個(gè)進(jìn)程啟動(dòng)
     * @param $worker
     */
    public function onWorkerStart($worker)
    {

    }
}

大佬我把代碼修改成了這樣 我發(fā)現(xiàn)客戶同雖然每次都new 了 但是并沒有影響到之前的鏈接，我開了ws服務(wù)根http服務(wù)，在http沒請(qǐng)求一次就new AsyncTcpConnection，模擬了下我的需求，貌似這么寫沒啥問題，服務(wù)端能收到消息，動(dòng)態(tài)新增的客戶端也能收到消息。