多个并发 MQTT 订阅者

Question

我正在尝试运行多个 pub - 多个子消息交换。 当我尝试启动多个订阅者时，问题就出现了（想法是为每个订阅者订阅不同的主题）。

我打开 2 个终端：一个用于酒吧，一个用于订阅者。

这里是子代码：

import random
from paho.mqtt import client as mqtt_client
import threading

broker='172.18.0.3'
port = 1885
topics = ["sensors/humidity","sensors/temperature"]
# Generate a Client ID with the subscribe prefix.
client_id = f'subscribe-{random.randint(0, 100)}'

# ++ OBJECT ORIENTED PROGRAMMING?

def connect_mqtt() -> mqtt_client:
    def on_connect(client, userdata, flags, rc):
        if rc == 0:
            print("Connected to MQTT Broker!")
        else:
            print("Failed to connect, return code %d\n", 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, topic):
    
    def on_message(client, userdata, msg):
       
        print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic")

    client.subscribe(topic)
    client.on_message = on_message

def run():
    threads=[]
    for topic in topics:
        
        client = connect_mqtt()
        subscribe(client, topic)
        
        thread=threading.Thread(target=client.loop_forever)
        
        threads.append(thread)

    for thread in threads:
        
        thread.start()

if __name__ == '__main__':
    run()

它似乎没有按我想要的方式工作，因为它可能每次都会重新连接。相反，我希望他们永远保持清醒，等待我通过运行 pub Python 代码的另一个终端向他们发送消息。 哪里有问题？有没有更聪明的方法来实现这个目标？我对 MQTT 还很陌生，所以也许我没有以正确的方式面对这个问题。

Answer 1

您没有在代码中的任何位置启动 MQTT 客户端事件/网络循环

我建议您在此处阅读有关运行事件/网络循环的不同方法的 Paho 文档 https://eclipse.dev/paho/files/paho.mqtt.python/html/index.html#network-loop。

最简单的选择可能是在

client.loop_start()

调用之后添加

client.connect()

。