如何在不使用全局变量的情况下在Paho MQTT回调外处理Payload

1. 类封装方式 #

把MQTT客户端、消息存储和处理逻辑封装到类中，利用实例属性传递消息，避免全局变量。类的on_message方法可以将收到的Payload存入实例变量，或者直接调用类内的处理方法，其他模块通过类实例来获取处理结果。

示例代码：

import paho.mqtt.client as mqtt

class MQTTProcessor:
    def __init__(self, broker_host, broker_port):
        self.client = mqtt.Client()
        self.client.on_connect = self.on_connect
        self.client.on_message = self.on_message
        self.received_payload = None  # 存储收到的payload
        self.processed_result = None  # 存储处理结果
        self.client.connect(broker_host, broker_port, 60)

    def on_connect(self, client, userdata, flags, rc):
        print("Connected with result code "+str(rc))
        self.client.subscribe("your/topic")

    def on_message(self, client, userdata, msg):
        self.received_payload = msg.payload.decode()
        self.process_payload()

    def process_payload(self):
        # 自定义Payload处理逻辑
        if self.received_payload:
            self.processed_result = f"Processed: {self.received_payload.upper()}"

# 其他模块调用示例
if __name__ == "__main__":
    processor = MQTTProcessor("localhost", 1883)
    processor.client.loop_start()
    
    import time
    while True:
        if processor.processed_result:
            print(processor.processed_result)
            processor.processed_result = None  # 重置结果避免重复输出
        time.sleep(1)

2. 线程安全队列传递 #

利用Python标准库的queue.Queue实现消息的线程安全传递。on_message回调将收到的Payload放入队列，其他线程或模块从队列中取出数据进行处理，适合异步处理场景。

示例代码：

import paho.mqtt.client as mqtt
from queue import Queue
import threading
import time

# 模块级队列（仅在当前模块可见，其他模块可通过导入使用）
message_queue = Queue()

def on_connect(client, userdata, flags, rc):
    print("Connected with result code "+str(rc))
    client.subscribe("your/topic")

def on_message(client, userdata, msg):
    payload = msg.payload.decode()
    message_queue.put(payload)  # 将Payload放入队列

def payload_processor():
    # 该函数可放在其他模块，负责从队列取数据处理
    while True:
        payload = message_queue.get()
        # 自定义处理逻辑
        processed_result = f"Handled: {payload}"
        print(processed_result)
        message_queue.task_done()

if __name__ == "__main__":
    client = mqtt.Client()
    client.on_connect = on_connect
    client.on_message = on_message
    client.connect("localhost", 1883, 60)

    # 启动处理线程
    threading.Thread(target=payload_processor, daemon=True).start()

    # 启动MQTT循环
    client.loop_forever()

3. 注册外部处理函数 #

允许其他模块注册自定义的处理函数，on_message回调触发时直接调用这些注册的函数，将Payload传递过去，实现处理逻辑的解耦。

示例代码：

import paho.mqtt.client as mqtt

# 存储注册的处理函数
message_handlers = []

def register_handler(handler_func):
    message_handlers.append(handler_func)

def on_connect(client, userdata, flags, rc):
    print("Connected with result code "+str(rc))
    client.subscribe("your/topic")

def on_message(client, userdata, msg):
    payload = msg.payload.decode()
    # 调用所有注册的处理函数
    for handler in message_handlers:
        handler(payload)

# 其他模块注册处理函数示例
def my_custom_handler(payload):
    processed = f"Custom processed: {payload}"
    print(processed)
    # 可将处理结果存储到模块内变量或返回给调用方

if __name__ == "__main__":
    register_handler(my_custom_handler)

    client = mqtt.Client()
    client.on_connect = on_connect
    client.on_message = on_message
    client.connect("localhost", 1883, 60)

    client.loop_forever()

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