我有一个场景，其中移动应用程序调用由我的应用程序托管的 rest API。在此过程中，我需要通过 MQTT 向下游系统发送消息，并等待直到我收到该消息的响应。然后我回复了移动应用程序。
这里的挑战是，通过 MQTT 的消息传递是异步的。所以我收到的消息将在不同的线程中(一些侦听器类，监听 messageArrived(((。如何回到调用 http 线程？
我们是否有泛美卫生图书馆支持的同步通信？比如我发送一条消息，打开某个主题并等待它，直到收到某些消息或超时？
在此代码中，我通过使用队列和 while 循环同步管理异步 Paho Python MQTT 通信。
MTTQ 经纪商 https://mosquitto.org/
泛美卫生客户 https://pypi.org/project/paho-mqtt/
# python3.6
import random
from paho.mqtt import client as mqtt_client
from queue import Queue
import datetime
import time
broker = '127.0.0.1'#'broker.emqx.io'
port = 1883
topic_1 = "Parent/Topic1"
topic_2 = "Parent/Topic2"
msg1_start = "topic1: start"
msg1_started = "topic1: started"
msg1_finished = "topic1: finished"
msg2_start = "topic2: start"
msg2_started = "topic2: started"
msg2_finished = "topic2: finished"
# Generate a Client ID with the subscribe prefix.
client_id = f'subscribe-{random.randint(0, 100)}'
# username = 'emqx'
# password = 'public'
msgQueue = Queue()
def connect_mqtt() -> mqtt_client:
def on_connect(client, userdata, flags, rc):
if rc == 0:
print(f"{datetime.datetime.now()} [Info] Connected to MQTT Broker!")
else:
print(f"{datetime.datetime.now()} [Error] Failed to connect, return code %dn", rc)
client = mqtt_client.Client(client_id)
# client.username_pw_set(username, password)
client.on_connect = on_connect
client.connect(broker, port)
return client
def subscribe(client: mqtt_client):
def on_message(client, userdata, msg):
print(f"{datetime.datetime.now()} [Info] Received `{msg.payload.decode()}` from `{msg.topic}` topic")
msgQueue.put(msg)
client.subscribe(topic_1)
client.subscribe(topic_2)
client.on_message = on_message
def publish(client,topic,msg):
result = client.publish(topic, msg)
# result: [0, 1]
status = result[0]
if status == 0:
print(f"{datetime.datetime.now()} [Info] Send `{msg}` to topic `{topic}`")
else:
print(f"{datetime.datetime.now()} [Error] Failed to send message to topic {topic}")
def run():
client = connect_mqtt()
time.sleep(1)
publish(client,topic_1,msg1_start)
subscribe(client)
#client.loop_forever()
client.loop_start()
while True:
message = msgQueue.get()
if message is None:
continue
print(f"{datetime.datetime.now()} [Info] received from queue",str(message.payload.decode("utf-8")))
if message.topic == topic_1 and message.payload.decode() == msg1_finished:
publish(client,topic_2,msg2_start)
elif message.topic == topic_2 and message.payload.decode() == msg2_finished:
break
client.loop_stop() #client thread needs to be stopped when topic2 completed.
if __name__ == '__main__':
run()
MQTT本质上是异步的，所有Pub/Sub实现也是如此。在协议级别没有回复消息的概念，您无法知道是否会收到对已发布消息的响应(或者您可能会收到很多回复(，因为您甚至无法知道是否有订阅者您发布的主题。
可以构建一个以这种方式工作的系统，但您需要维护所有正在进行的请求的状态机，实施合理的超时策略，并计算出在收到多个响应时该怎么做。
您没有提到您正在使用哪些不同的 Paho 库，但我从方法名称中猜测 Java，但不知道您正在使用什么 HTTP 框架以及许多其他因素，我不会建议解决方案，特别是因为它将涉及大量轮询和同步。
移动应用程序不能直接发布和订阅 MQTT 主题有什么原因吗？这将消除对此的需求。