paho mqtt 中的 on_message 是否在新线程中运行？

Question

假设我订阅了某个主题，并且 mosquitto 服务器正在持续发布有关该主题的一些消息。

当新消息从 mosquitto 代理到达时，如何调用 on_message 方法？它是在主线程上运行从而阻塞它并仅在处理当前消息时才处理下一条消息，还是在每次新消息到达时使用 on_message 方法生成一个新线程？

import paho.mqtt.client as mqtt

def on_message(client, userdata, message):
    print message.payload
    return True

topics = ["topic1",]
client = mqtt.Client("testclient",protocol=mqtt.MQTTv31,clean_session=True)
client.on_message=on_message
client.connect("127.0.0.1", 1883, 60)
for tpc in topics:
    client.subscribe(tpc,0)
client.loop_forever()

Answer 1

回调都在客户端网络线程上运行。

这取决于您如何“启动”网络线程。

如果您使用该client.loop_forever()方法，那么它将使用当前线程来运行客户端的网络线程，并因此在该调用中阻塞。

如果您使用该client.loop_start()方法，则会在后台启动一个新线程来运行网络循环和所有回调。

所有回调都会阻塞网络线程的执行，这就是为什么你不应该直接在回调中运行长时间运行或阻塞的任务。如果您有长时间运行的处理要做，您应该将传入的消息排队并让另一个线程（或线程池）处理它。

ps 对于您发布的代码，您应该真正移动 for 循环和订阅on_connect回调的调用，以便：

1. 他们只能在成功的连接上运行
2. 网络线程正在运行以处理发送订阅的数据包并处理在您完成订阅所有主题之前可能到达的任何传入消息（如果数组长于 1）