Python多进程处理MQTT订阅数据的实时并行执行问题求助

你的脚本存在几个关键错误，导致无法实现并行运行和实时处理，以下是修正后的代码及具体解决思路：

修正后的代码 #

import paho.mqtt.client as mqtt
from multiprocessing import Manager, Process
import time

topics_to_subscribe=['test/topic1','test/topic2','test/topic3','test/topic4']

def process(buffer1, buffer2):
    # 持续监听缓冲区，实时处理数据
    while True:
        # 处理buffer1的数据
        if buffer1:
            data = buffer1.pop(0)
            print(f"处理buffer1数据: {data.decode('utf-8')}")
            # 在这里添加你的具体处理逻辑
        
        # 处理buffer2的数据
        if buffer2:
            data = buffer2.pop(0)
            print(f"处理buffer2数据: {data.decode('utf-8')}")
            # 在这里添加你的具体处理逻辑
        
        time.sleep(0.1)  # 避免过度占用CPU

def mqtt_subscribe(buffer):
    def on_connect(client, userdata, flags, rc):
        for topic in topics_to_subscribe:
            client.subscribe(topic)
      
    def on_message(client, userdata, message):
        buffer.append(message.payload)

    client = mqtt.Client()
    client.on_connect = on_connect
    client.on_message = on_message

    broker_address="192.168.1.184"
    client.connect(broker_address)
    client.loop_forever()

def mqtt_subscribe2(buffer2):
    def on_connect(client, userdata, flags, rc):
        client.subscribe('topic1')
      
    def on_message(client, userdata, message):
        buffer2.append(message.payload)

    client = mqtt.Client()
    client.on_connect = on_connect
    client.on_message = on_message

    broker_address="192.168.1.184"
    client.connect(broker_address)
    client.loop_forever()
   
if __name__ == '__main__':
    manager = Manager()

    buffer1 = manager.list()
    buffer2 = manager.list()
    # 修正函数名拼写错误，以及args参数的元组格式（必须加逗号）
    p1 = Process(target=mqtt_subscribe, args=(buffer1,))
    p2 = Process(target=mqtt_subscribe2, args=(buffer2,))
    p3 = Process(target=process, args=(buffer1, buffer2))
    
    p1.start()
    p2.start()
    p3.start()

    # 可选：等待子进程结束
    p1.join()
    p2.join()
    p3.join()

关键错误修正点 #

• 函数名拼写错误：你创建进程时写的mqtt_subscriber2是不存在的，正确名称是mqtt_subscribe2；同时p1应该绑定mqtt_subscribe对应buffer1，而非重复使用mqtt_subscribe2。
• 参数传递错误：args=(buffer2)会被Python解析为单个对象而非元组，必须写成args=(buffer2,)（末尾加逗号），否则会导致进程启动时参数不匹配报错。
• 处理函数无监听逻辑：原process函数没有持续运行的逻辑，无法实时检测缓冲区的新数据，添加while True循环轮询缓冲区实现实时处理。
• 冗余代码清理：主进程中创建的client = mqtt.Client()没有实际作用，直接删除即可。

优化思路 #

1. 使用Queue替代List：multiprocessing.Queue是更适合进程间数据传递的结构，自带阻塞等待机制，无需手动轮询，能更高效地实现实时触发处理。示例如下：

# 替换Manager.list为Queue
buffer1 = manager.Queue()
buffer2 = manager.Queue()

# 处理函数改为阻塞获取数据
def process(buffer1, buffer2):
    while True:
        # 阻塞等待buffer1的数据
        data = buffer1.get()
        print(f"处理buffer1数据: {data.decode('utf-8')}")
        
        # 阻塞等待buffer2的数据
        data = buffer2.get()
        print(f"处理buffer2数据: {data.decode('utf-8')}")

2. 添加进程锁：如果需要对共享数据做复杂操作，可使用manager.Lock()确保进程间操作的原子性，避免数据混乱。
3. 异常处理：在MQTT连接、数据处理逻辑中添加异常捕获，避免单个进程崩溃导致整个程序停止。

内容的提问来源于stack exchange，提问作者user18430327