Closed yuanxin32323 closed 3 years ago
使用场景是homeassistant同时开启10个开关状态,相当于并发10条消息到10个topic, 但用simps/mqtt客户端只能收到其中3个左右的topic消息。 同时也测试了其它的mqtt client包,bluerhinos/phpMQTT使用正常,能收到全部消息。 服务端应该没有问题,服务端使用的EMQXBorker。
代码如下
\Swoole\Runtime::enableCoroutine(); // 此行代码后,文件操作,sleep,Mysqli,PDO,streams等都变成异步IO,见'一键协程化'章节 \Co\run(function ()use ($output) { go(function ()use ($output) { $config = [ 'userName' => 'xxxxxx', // 用户名 'password' => 'xxxxxx', // 密码 'clientId' => 'xxxxxxx' . uniqid(), // 客户端id 'keepAlive' => 0, // 默认0秒,设置成0代表禁用 'protocolName' => 'MQTT', // 协议名,默认为MQTT(3.1.1版本),也可为MQIsdp(3.1版本) 'protocolLevel' => 4, // 协议等级,MQTT3.1.1版本为4,5.0版本为5,MQIsdp为3 'properties' => [], // MQTT5 中所需要的属性 'delay' => 3000, // 重连时的延迟时间 (毫秒) 'maxAttempts' => 5, // 最大重连次数。默认-1,表示不限制 'swooleConfig' => [] ]; $configObj = new \Simps\MQTT\Config\ClientConfig($config); $client = new \Simps\MQTT\Client('127.0.0.1', 1883, $configObj); while (!$client->connect()) { \Swoole\Coroutine::sleep(3); $output->writeln('mqtt connect fail...wait 3s reconnect'); } $output->writeln('mqtt connect success'); //订阅 $topics = [ 'homeassistant/switch/8266/+/set' => 0, ]; if ($client->subscribe($topics)) { $output->writeln('subscribe success!!'); } else { $output->writeln('subscribe fail!!'); } //获取消息 while ($buffer = $client->recv()) { var_dump($buffer); if ($buffer && $buffer !== true) { $timeSincePing = time(); // 收到的数据包 $state_topic = str_replace('set', 'state', $buffer['topic']); //原样返回状态 $output->writeln("返回topic:{$state_topic} {$buffer['message']}"); $client->publish($state_topic, $buffer['message'], 0); } if (isset($config['keep_alive']) && $timeSincePing < (time() - $config['keep_alive'])) { $buffer = $client->ping(); if ($buffer) { $timeSincePing = time(); } else { $client->close(); break; } } } }); });
使用tcpdump抓包,把mqtt.pcap文件发上来
tcpdump
mqtt.pcap
tcpdump -i en0 port 1883 -w mqtt.pcap
未开启Swoole配置项:
'swooleConfig' => [ 'open_mqtt_protocol' => true, ]
使用场景是homeassistant同时开启10个开关状态,相当于并发10条消息到10个topic, 但用simps/mqtt客户端只能收到其中3个左右的topic消息。 同时也测试了其它的mqtt client包,bluerhinos/phpMQTT使用正常,能收到全部消息。 服务端应该没有问题,服务端使用的EMQXBorker。
代码如下