mqtt - 如何从 paho python mqtt 中的多进程发布

Question

我在树莓派上运行一个 python 脚本，它在收到 MQTT 消息后，在多进程中运行一个函数。从主脚本发布 mqtt 消息工作正常并被代理接收。但是，在新进程中运行的函数无法发布。没有错误信息。该函数确实会打印测试日志，因此它确实在运行。

    ### on message, run function in a new process
    def on_message(client, obj, msg):
        def threaded_message():
            print("Hello, process is running")
            ### This publish does not work!
            mqttc.publish(topicStatus, "message received")

    myProcess = multiprocessing.Process(target=threaded_message)
    myProcess.start()      
    ### MQTT setup
    mqttc = mqtt.Client()
    mqttc.on_message = on_message
    mqttc.on_connect = on_connect
    mqttc.on_publish = on_publish
    mqttc.on_subscribe = on_subscribe

    url_str = 'm24.cloudmqtt.com'
    url_port = '16310'

    topicStatus = "Home/Status"
    topicCommands = "Home/Commands"

    mqttc.username_pw_set(myUsername, myPassword)
    mqttc.connect(url_str, url_port)
    ### This publish does work!
    mqttc.publish(topicStatus, "Online")

    mqttc.loop_forever()

在主脚本中运行 mqttc.publish 成功发布消息。运行多进程打印消息，但不发布消息。

为什么发布功能在这种情况下不起作用？

Answer 1

您的 MQTT 未发布的原因是您尝试将多个 MQTT 线程连接到同一个客户端，而您只为所有线程连接一个客户端。

解决方法:对于多处理，您必须创建多个 mqtt 客户端，这样您的多个客户端才能并行发布消息。

from datetime import datetime, time
import paho.mqtt.client as mqtt
import time
import multiprocessing


def on_connect(client, userdata, flags, rc):
    
    if rc == 0:
        global Connected               
        Connected = True               
    
    else:
        print("Connection failed")

def client_conn():
    # MQTT connection 
    cli = "test"+str(datetime.now())
    client = mqtt.Client(cli)
    client.on_connect = on_connect
    
                
    client.connect("mqttServer", "mqttPort")
    client.username_pw_set("mqttUsername", "mqttPassword")
    client.loop_start()
    while Connected != True:  
        time.sleep(0.1)
    return client

def mqtt_publish():
    # Publish on mqtt
    client = client_conn()
    client.publish("topic", "payload")
    


def mul_process():
    # Process creation
    p = multiprocessing.Process(target=mqtt_publish)
    p.start()
    p.join()
    
# Execution starts here
mul_process()