如何通过mqtt实现一个“邮箱”服务？

Question

我是所有 mqtt 的新手，作为第一个练习，我想通过持久的 mqtt 会话创建一个“邮箱”服务。激励是一个低功耗 ESP8266 设备，它大部分时间都处于睡眠状态，并定期唤醒并检查是否有任何待处理的命令。

我尝试使用 python 和 paho mqtt 在我的 Linux 主机上通过发送者和接收者来实现它。Mosquitto 作为代理在后台运行。

首先是“mbox”发件人，它会在每次按下 Enter 时发送另一条消息。

import paho.mqtt.client as mqtt 

broker_address='127.0.0.1'

client = mqtt.Client('MBoxClient') 

client.connect(broker_address)

counter = 1

while True:

  print('Press Enter to send msg #'+str(counter)+': ', end='')

  if input().startswith('q'):

    break

  client.publish("mbox/mail","Hello "+str(counter), qos=1)

  counter += 1

client.disconnect()

print('done!')

这是我的 mbox 接收器：

import paho.mqtt.client as mqtt

import time

def on_message(client, userdata, message):

  print("message:", message.topic + ': ' + str(message.payload.decode("utf-8")))

print('I\'m listening for mbox messages!')

broker_address="127.0.0.1"

client_name='mbox'

is_first=True

while 1:

  client = mqtt.Client(client_name, clean_session=is_first)

  is_first=False

  print("polling")

  client.on_message=on_message 

  client.connect(broker_address) 

  client.subscribe('mbox/#',qos=1)

  client.loop_start()

  time.sleep(0.1) # How long should this time be?

  client.loop_stop()

#  client.loop(0.1) # why doesn't this do the same action as the previous three lines?

  client.disconnect()

  time.sleep(5)

尽管这可行，但我觉得我的解决方案非常骇人听闻。client.loop_start()并client.loop_stop()创建另一个线程。但是当我尝试client.loop(0.1)这样做时，它没有用。

所以我的问题是：

• 是否有直接轮询消息的方法，而不是使用的间接方法loop_start();…;loop_stop()？

• 如果 usingloop_start();time.sleep(t);loop_end()是惯用的，我怎么知道要睡多久？

• loop(0.1); instead of 为什么在我执行loop_start()时接收器不工作；睡眠（0.1）；循环停止（）`？有什么不同？

• 接收方是否保证收到所有消息？

• 有没有更好的方法来实现这种模式？

Answer 1

问题按顺序回答。不，轮询完全违背了像 MQTT 这样的发布/订阅协议的意义你真的应该client.loop()在一个循环中调用，它默认只在提供的超时期限内处理 1 个数据包。QOS 1 需要多个数据包才能完成传送。调用client.loop(0.1)将阻塞 0.1 秒以等待传入消息，然后返回。如果消息在 0.1 秒后到达，它将驻留在操作系统 TCP/IP 堆栈中，直到您client.loop()再次调用。如果您没有定期调用它，那么代理将启动客户端，因为 KeepAlive 测试将失败。客户端循环还处理发送所有订阅消息。假设消息在 QOS > 0 时发布并且您已在 QOS > 0 时订阅并且客户端 ID 保持不变并且干净会话为 false 代理应该在订阅者离线时传递和发布消息如前所述，您需要client.loop()为每条消息调用多次，因为每个唤醒周期您只调用一次。启动后台线程将在您让它运行的时间长度内处理所有必需的消息。