workerman 队列 内存泄漏

问题描述

如果我正常运行订阅 没有转发 到队列 就不会出现内存泄漏 如果我转发到队列就会出现内存泄漏为啥呢

程序代码

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数据是长期稳定
            // 但是如果我将消息数据组装成新的数组 newData 转发到队列就会出现内存泄漏
            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();
    }
}

报错信息

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"