Workerman\Mqtt\Client 客户端

问题描述

当服务端响应的数据过长 ,就会报错
buffer:2c2262757a7a6572223a226f6e222c22636f6c6f72223a2235222c227068617365223a302c22736e223a225a51323636 lenth:11298 not enough for unpackString

程序代码

<?php

namespace app\process;

ini_set('memory_limit', '512M');

use Workerman\Timer;
use Workerman\Mqtt\Client;
use support\Redis;
use app\common\model\RequestLogs;

class MqttSubscribeProcess
{
    protected ?Client $mqtt = null;

    public function onWorkerStart($worker): void
    {
        try {
            $config = config('lightstrip.dongji');
            $this->mqtt = new Client($config['baseUrl'], [
                'client_id' => 'workerman_' . getmypid(),
                'username' => $config['username'],
                'password' => $config['password'],
                'max_packet_size' => 10 * 1024 * 1024,
                'protocol_level' => 4,
                'keepalive' => 60,
                'reconnect_period' => 5,
                'ssl' => [
                    'verify_peer' => false,
                    'verify_peer_name' => false,
                ],
            ]);

            // 连接成功
            $this->mqtt->onConnect = function ($mqtt) {
                try {
                    $this->logInfo("✅ MQTT 连接成功");
                    // $mqtt->subscribe('/CLMGW/TFRESTRSP', ['qos' => 1]); //v2用这个
                    $mqtt->subscribe('/CLMGW/ZQRSP', ['qos' => 1]); //v3用这个有响应数据{"code":0,"id":39,"sn":"CLMGW-DFDA2601"}
                    $this->logInfo("✅ 已订阅主题: /CLMGW/TFRESTRSP");
                } catch (\Throwable $e) {
                    $this->logError('onConnect异常', $e);
                }
            };

            // 收到消息
            $this->mqtt->onMessage = function ($topic, $content) {
                try {
                    file_put_contents(
                        runtime_path() . '/mqtt_raw_message.log',
                        "【" . date('Y-m-d H:i:s') . "】\n"
                        . "topic: {$topic}\n"
                        . "len: " . strlen($content) . "\n"
                        . $content . "\n"
                        . str_repeat('-', 120) . "\n",
                        FILE_APPEND | LOCK_EX
                    );

                    $this->logInfo("📩 收到响应 topic: {$topic} len=" . strlen($content));

                    $data = json_decode($content, true);

                    if (!is_array($data)) {
                        $this->logMessageError('onMessage JSON解析失败', $topic, $content, null);
                        return;
                    }

                    $id = $data['id'] ?? null;
                    if (!$id) {
                        $this->logMessageError('onMessage 响应中没有 id', $topic, $content, $data);
                        return;
                    }

                    $responseQueue = "mqtt:response:{$id}";
                    $waitingKey = "mqtt:waiting:{$id}";

                    $waiting = false;
                    try {
                        $waiting = Redis::exists($waitingKey);
                    } catch (\Throwable $e) {
                        $this->logError('Redis exists异常', $e);
                    }

                    if ($waiting) {
                        Redis::lPush($responseQueue, json_encode($data, JSON_UNESCAPED_UNICODE));
                        Redis::expire($responseQueue, 15);
                        $this->logInfo("✅ 响应写入Redis: {$responseQueue}");
                    } else {
                        $this->logInfo("⚠️ waitingKey不存在，忽略迟到响应 id={$id}");
                    }

                    RequestLogs::where(['type' => 'dongji', 'number' => $id])->update([
                        'result' => json_encode($data, JSON_UNESCAPED_UNICODE),
                        'update_time' => time()
                    ]);

                } catch (\Throwable $e) {
                    $this->logMessageError('onMessage处理异常', $topic, $content, null, $e);
                }
            };

            // MQTT错误
            $this->mqtt->onError = function ($error) {
                $message = is_object($error) && method_exists($error, 'getMessage')
                    ? $error->getMessage()
                    : print_r($error, true);

                $file = is_object($error) && method_exists($error, 'getFile')
                    ? $error->getFile()
                    : 'unknown';

                $line = is_object($error) && method_exists($error, 'getLine')
                    ? $error->getLine()
                    : 'unknown';

                $trace = is_object($error) && method_exists($error, 'getTraceAsString')
                    ? $error->getTraceAsString()
                    : print_r($error, true);

                $log = "【MQTT onError异常】" . date('Y-m-d H:i:s') . PHP_EOL
                    . "message: {$message}" . PHP_EOL
                    . "file: {$file}" . PHP_EOL
                    . "line: {$line}" . PHP_EOL
                    . "trace: {$trace}" . PHP_EOL
                    . str_repeat('-', 120) . PHP_EOL;

                file_put_contents(runtime_path() . '/mqtt_error.log', $log, FILE_APPEND | LOCK_EX);
            };

            // 发起连接
            $this->mqtt->connect();

            // 同步命令队列
            Timer::add(0.1, function () {
                try {
                    $task = Redis::rPop('mqtt:sync:queue');
                    if (!$task) {
                        return;
                    }

                    $data = json_decode($task, true);
                    if (!$data) {
                        file_put_contents(
                            runtime_path() . '/mqtt_error.log',
                            "【sync队列JSON异常】" . date('Y-m-d H:i:s') . PHP_EOL
                            . "raw: {$task}" . PHP_EOL
                            . str_repeat('-', 120) . PHP_EOL,
                            FILE_APPEND | LOCK_EX
                        );
                        return;
                    }

                    if (empty($data['gateway_sn']) || empty($data['id'])) {
                        file_put_contents(
                            runtime_path() . '/mqtt_error.log',
                            "【sync队列缺少关键字段】" . date('Y-m-d H:i:s') . PHP_EOL
                            . "data: " . json_encode($data, JSON_UNESCAPED_UNICODE) . PHP_EOL
                            . str_repeat('-', 120) . PHP_EOL,
                            FILE_APPEND | LOCK_EX
                        );
                        return;
                    }

                    $topic = "/CLMGW/{$data['gateway_sn']}";
                    $payload = 'API:' . json_encode($data, JSON_UNESCAPED_UNICODE);
                    $this->mqtt->publish($topic, $payload, ['qos' => 1]);
                    dump($topic,$payload);
                    $this->logInfo("🚀 同步指令已发送 id:{$data['id']} command:{$data['command']}");

                    // file_put_contents(
                    //     runtime_path() . '/mqtt_publish.log',
                    //     "【SYNC】" . date('Y-m-d H:i:s') . PHP_EOL
                    //     . "topic: {$topic}" . PHP_EOL
                    //     . "payload: {$payload}" . PHP_EOL
                    //     . str_repeat('-', 120) . PHP_EOL,
                    //     FILE_APPEND | LOCK_EX
                    // );
                } catch (\Throwable $e) {
                    $this->logError('sync timer异常', $e);
                }
            });

            // 异步命令队列
            Timer::add(0.5, function () {
                try {
                    $task = Redis::rPop('mqtt:async:queue');
                    if (!$task) {
                        return;
                    }

                    $data = json_decode($task, true);
                    if (!$data) {
                        file_put_contents(
                            runtime_path() . '/mqtt_error.log',
                            "【async队列JSON异常】" . date('Y-m-d H:i:s') . PHP_EOL
                            . "raw: {$task}" . PHP_EOL
                            . str_repeat('-', 120) . PHP_EOL,
                            FILE_APPEND | LOCK_EX
                        );
                        return;
                    }

                    if (empty($data['gateway_sn']) || empty($data['id'])) {
                        file_put_contents(
                            runtime_path() . '/mqtt_error.log',
                            "【async队列缺少关键字段】" . date('Y-m-d H:i:s') . PHP_EOL
                            . "data: " . json_encode($data, JSON_UNESCAPED_UNICODE) . PHP_EOL
                            . str_repeat('-', 120) . PHP_EOL,
                            FILE_APPEND | LOCK_EX
                        );
                        return;
                    }

                    $topic = "/CLMGW/{$data['gateway_sn']}";
                    $payload = 'API:' . json_encode($data, JSON_UNESCAPED_UNICODE);

                    $this->mqtt->publish($topic, $payload, ['qos' => 0]);

                    $this->logInfo("📤 异步指令已发送 id:{$data['id']} command:{$data['command']}");

                    file_put_contents(
                        runtime_path() . '/mqtt_publish.log',
                        "【ASYNC】" . date('Y-m-d H:i:s') . PHP_EOL
                        . "topic: {$topic}" . PHP_EOL
                        . "payload: {$payload}" . PHP_EOL
                        . str_repeat('-', 120) . PHP_EOL,
                        FILE_APPEND | LOCK_EX
                    );
                } catch (\Throwable $e) {
                    $this->logError('async timer异常', $e);
                }
            });

            // Redis 心跳检测（可选）
            Timer::add(30, function () {
                try {
                    Redis::ping();
                    $this->logInfo("✅ Redis ping 正常");
                } catch (\Throwable $e) {
                    $this->logError('Redis心跳异常', $e);
                }
            });

        } catch (\Throwable $e) {
            $this->logError('onWorkerStart异常', $e);
        }
    }

    protected function logInfo(string $message): void
    {
        file_put_contents(
            runtime_path() . '/mqtt_info.log',
            '【' . date('Y-m-d H:i:s') . "】 {$message}" . PHP_EOL,
            FILE_APPEND | LOCK_EX
        );
    }

    protected function logError(string $title, \Throwable $e): void
    {
        $log = "【{$title}】" . date('Y-m-d H:i:s') . PHP_EOL
            . "message: " . $e->getMessage() . PHP_EOL
            . "file: " . $e->getFile() . PHP_EOL
            . "line: " . $e->getLine() . PHP_EOL
            . "trace: " . $e->getTraceAsString() . PHP_EOL
            . str_repeat('-', 120) . PHP_EOL;

        file_put_contents(runtime_path() . '/mqtt_error.log', $log, FILE_APPEND | LOCK_EX);
    }

    protected function logMessageError(string $title, string $topic, string $content, $data = null, \Throwable $e = null): void
    {
        $log = "【{$title}】" . date('Y-m-d H:i:s') . PHP_EOL
            . "topic: {$topic}" . PHP_EOL
            . "content_len: " . strlen($content) . PHP_EOL
            . "content_head: " . substr($content, 0, 500) . PHP_EOL
            . "content_tail: " . substr($content, -500) . PHP_EOL;

        if ($data !== null) {
            $log .= "decoded: " . json_encode($data, JSON_UNESCAPED_UNICODE) . PHP_EOL;
        }

        if ($e) {
            $log .= "message: " . $e->getMessage() . PHP_EOL
                . "file: " . $e->getFile() . PHP_EOL
                . "line: " . $e->getLine() . PHP_EOL
                . "trace: " . $e->getTraceAsString() . PHP_EOL;
        }

        $log .= str_repeat('-', 120) . PHP_EOL;

        file_put_contents(runtime_path() . '/mqtt_error.log', $log, FILE_APPEND | LOCK_EX);
    }
}

截图报错信息里报错文件相关代码

这里粘贴截图

操作系统及workerman/webman等框架组件具体版本

php8.2 webman2.2 "workerman/mqtt": "^2.2"
这里写具体的系统环境相关信息