使用MQTT回调统计消息接收数时如何避免全局变量？

Paho的回调签名虽然固定，但有几种可行的方式能绕开全局变量，下面是具体实现方案：

1. 利用Paho的userdata参数传递计数器对象 #

Paho客户端初始化时允许传入userdata参数，该参数会直接传递给回调函数。你可以把计数器和互斥锁封装成一个对象，作为userdata传入，回调里直接操作这个对象的属性即可。

示例代码：

import threading
import paho.mqtt.client as paho

# 封装计数器与锁的类
class MessageCounter:
    def __init__(self):
        self.count = 0
        self.mutex = threading.Lock()

    def increment(self):
        with self.mutex:
            self.count += 1

# 初始化计数器实例
msg_counter = MessageCounter()

# 回调函数
def on_message_callback_v3(message_client, userdata, message):
    # userdata就是传入的MessageCounter实例
    userdata.increment()
    msg = str(message.payload.decode("utf-8"))
    print(f"▼▼ ON MESSAGE ▼▼")
    print(f"  Message received for client: {message_client}")
    print(f"  Message user data: {userdata}")
    print(f"  Message topic: {message.topic}")
    print(f"  Message body: {msg}")

# 初始化客户端并绑定回调与userdata
client = paho.mqtt.Client(client_id="your_client_id")
client.user_data_set(msg_counter)
client.on_message = on_message_callback_v3

后续对比计数时，直接取msg_counter.count与published_count即可。

2. 使用闭包封装计数器 #

把计数器和回调函数放在闭包内部，让计数器成为闭包的私有变量，避免污染全局作用域。

示例代码：

import threading
import paho.mqtt.client as paho

def create_message_callback():
    message_received_count = 0
    mutex = threading.Lock()

    def callback(message_client, userdata, message):
        nonlocal message_received_count
        with mutex:
            message_received_count += 1
        msg = str(message.payload.decode("utf-8"))
        print(f"▼▼ ON MESSAGE ▼▼")
        print(f"  Message received for client: {message_client}")
        print(f"  Message user data: {userdata}")
        print(f"  Message topic: {message.topic}")
        print(f"  Message body: {msg}")
    
    # 返回回调函数和获取计数的方法
    return callback, lambda: message_received_count

# 创建回调与计数获取函数
on_message_callback_v3, get_received_count = create_message_callback()

# 绑定回调到客户端
client = paho.mqtt.Client(client_id="your_client_id")
client.on_message = on_message_callback_v3

对比时调用get_received_count()就能拿到当前接收数。

3. 用类封装回调与计数器 #

把回调函数作为类的实例方法，计数器作为实例属性，所有状态都封装在类内部，结构更清晰，也便于后续扩展。

示例代码：

import threading
import paho.mqtt.client as paho

class MessageHandler:
    def __init__(self):
        self.message_received_count = 0
        self.mutex = threading.Lock()

    def on_message(self, message_client, userdata, message):
        with self.mutex:
            self.message_received_count += 1
        msg = str(message.payload.decode("utf-8"))
        print(f"▼▼ ON MESSAGE ▼▼")
        print(f"  Message received for client: {message_client}")
        print(f"  Message user data: {userdata}")
        print(f"  Message topic: {message.topic}")
        print(f"  Message body: {msg}")

# 实例化处理器
handler = MessageHandler()

# 绑定回调到客户端
client = paho.mqtt.Client(client_id="your_client_id")
client.on_message = handler.on_message

对比时直接使用handler.message_received_count即可。

以上三种方案都能有效避免全局变量，同时保留了线程安全的互斥锁机制。其中利用userdata的方式最贴合Paho的原生设计，类封装的方式更适合复杂业务场景的扩展。

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