問題描述

如果我正常運(yùn)行訂閱 沒有轉(zhuǎn)發(fā) 到隊(duì)列 就不會出現(xiàn)內(nèi)存泄漏 如果我轉(zhuǎn)發(fā)到隊(duì)列就會出現(xiàn)內(nèi)存泄漏為啥呢

程序代碼

class Subscribe extends Worker
{
    static $taskName = 'subscribe';
    static $apiConfig;
    static $queue;
    public function __construct($socket_name = '', array $context_option = array())
    {
        parent::__construct($socket_name, $context_option);
    }
    public function on_start($worker)
    {
        global $db_config;
        Db::setConfig($db_config);

        global $redis_config;
        self::$queue = new Client($redis_config['queue']['host'],$redis_config['queue']['options']);

        Utils::echoLogs(self::$taskName.' 啟動成功');

        // 作為客戶端鏈接WS
        self::subscribe($list);

    }
    public function on_stop($worker)
    {
        Utils::echoLogs(self::$taskName.' 已停止');
    }
    public function subscribe($data)
    {
        $con = new AsyncTcpConnection(self::$apiConfig['spot_ws']);
        $con->transport = 'ssl';
        $con->websocketPingInterval = 55;
        $con->onMessage = function ($con, $data) {
            // 正常接收Data數(shù)據(jù)是長期穩(wěn)定
            // 但是如果我將消息數(shù)據(jù)組裝成新的數(shù)組 newData 轉(zhuǎn)發(fā)到隊(duì)列就會出現(xiàn)內(nèi)存泄漏
            self::$queue->send('data_queue', $newData);
        };
        $con->onError = function ($con, $err_code, $err_msg) {
        };
        $con->onClose = function ($con) {
            $con->reConnect(1);
        };
        $con->connect();
    }
    public function run()
    {
        $this->onWorkerStart = array($this, 'on_start');
        $this->onWorkerStop = array($this, 'on_stop');
        parent::run();
    }
}

報(bào)錯信息

Worker[2584] process terminated with ERROR: E_ERROR "Allowed memory size of 536870912 bytes exhausted (tried to allocate 69632 bytes) in /mnt/wss/vendor/workerman/workerman/Connection/TcpConnection.php on line 582"