无法在 Python 中接收所有 MQTT 消息答案

【问题标题】：Unable to receive all MQTT messages in Python无法在 Python 中接收所有 MQTT 消息
【发布时间】：2017-12-09 08:15:03
【问题描述】：

我正在尝试使用 MQTT 将消息从一个 python 脚本发送到另一个。一个脚本是出版商。第二个脚本是订阅者。我每 0.1 秒发送一次消息。

出版商：

client = mqtt.Client('DataReaderPub')
client.connect('127.0.0.1', 1883, 60)
print("MQTT parameters set.")

# Read from all files
count = 0
for i in range(1,51):
    payload = "Hello world" + str(count)
    client.publish(testtopic, payload, int(publisherqos))
    client.loop()
    count = count+1
    print(count, ' msg sent: ', payload)
    sleep(0.1)

订阅者：

subclient = mqtt.Client("DynamicDetectorSub")
subclient.on_message = on_message
subclient.connect('127.0.0.1')

subclient.subscribe(testtopic, int(subscriberqos))

subclient.loop_forever()

mosquitto 代理版本 - 3.1

mosquitto.conf 将最大飞行消息设置为 0，持久性为真。

发布者 QOS = 2

订阅者 QOS = 2

两个脚本中的topic = 'test'

当我在同一个脚本中运行订阅者和发布者时，消息会按预期发送和接收。但是当它们在单独的脚本中时，我没有收到所有消息，有时甚至没有消息。我先运行订阅者，然后再运行发布者。我已经尝试使用 loop.start() 和 loop.stop() 订阅用户等待几分钟。

我无法调试此问题。任何指针都会很棒！

编辑：

我在发布后包含了 client.loop()。 -> 和以前一样的输出
当我在 'on_connect' 和 'on_disconnect' 中打印出语句时，我注意到客户端 mqtt 连接已建立并几乎立即断开连接。这种情况每秒都会发生。我什至曾经收到过这条消息-

[WinError 10053] 已建立的连接被主机中的软件中止

保持活力 = 60

还有其他参数我应该看吗？

【问题讨论】：

编辑问题以包含发布者的完整代码。此外，如果您有来自代理的日志，请包括那些
另外，您同时运行多少个发布者实例？
只有一个发布者和一个订阅者。

标签： python mqtt

【解决方案1】：

您还需要在发布者中调用网络循环函数，以便客户端实际上有一些时间来执行 IO（以及 QOS2 的双握手）。

在客户端调用client.publish()后添加client.loop()：

import paho.mqtt.client as mqtt
import time

client = mqtt.Client('DataReaderPub')
client.connect('127.0.0.1', 1883, 60)
print("MQTT parameters set.")

# Read from all files
count = 0
for i in range(1,51):
    payload = "Hello world" + str(count)
    client.publish("test", payload, 2)
    client.loop()
    count = count+1
    print(count, ' msg sent: ', payload)
    time.sleep(0.1)

订阅者代码：

import paho.mqtt.client as mqtt

def on_message(client, userdata, msg):
  print(msg.topic + " " + str(msg.payload))

subclient = mqtt.Client("DynamicDetectorSub")
subclient.on_message = on_message
subclient.connect('127.0.0.1')

subclient.subscribe("test", 2)

subclient.loop_forever()

【讨论】：

不。这没有帮助。
是的，确实如此。编辑您的问题以包含发布者的完整代码以及此更改
我已编辑答案以显示我编写的测试的完整代码，以证明client.loop() 是必需的
根据您的建议编辑代码。还是行不通。如上述编辑中所述，我收到错误，即已建立的连接被主机系统中的软件中止。我不知道那个是什么意思。我不知道还有哪些软件会干扰订阅者 mqtt 连接。

【解决方案2】：

当我运行您的代码时，订阅者经常丢失最后一个数据包。否则我无法重现您描述的问题。

如果我改写这样的发布者...

from time import sleep
import paho.mqtt.client as mqtt

client = mqtt.Client('DataReaderPub')
client.connect('127.0.0.1', 1883, 60)
print("MQTT parameters set.")

client.loop_start()

# Read from all files
count = 0
for i in range(1,51):
    payload = "Hello world" + str(count)
    client.publish('test', payload, 2)
    count = count+1
    print(count, ' msg sent: ', payload)
    sleep(0.1)

client.loop_stop()
client.disconnect()

...然后我不再看到丢弃的数据包。我在这里使用start_loop/stop_loop 方法，它异步运行mqtt 循环。我不确定究竟是什么导致您的数据包丢失，但我怀疑代码退出时最终消息仍在发布者的发送队列中。

【讨论】：

【解决方案3】：

结果证明这是一个愚蠢的错误。正如 hardillb 建议的那样，我查看了代理日志。它表明订阅者客户端已经连接。很长一段时间后，我正在使用 Pycharm。所以我不小心运行了很多次发布者和订阅者，以至于它们在输出控制台中并行运行。难怪他们断开连接，因为客户端 ID 相同。抱歉，添麻烦了。 BTW client.loop() 发布后不需要。谢谢hardillb。

【讨论】：