使用Python Paho处理短时间大量MQTT消息的并发异常问题

问题原因 #

你遇到的核心问题是竞态条件：虽然Paho MQTT的on_message回调默认在单个网络线程中串行执行，但如果代码存在多线程触发（比如误启动多个网络循环），或者Python的非原子操作在高频率回调下出现执行间隙，会导致多个回调同时读取同一个self.counter值，各自执行+1后覆盖结果，最终计数器未按预期递增。数据库重复写入也是同理：多个回调同时判断“满足60秒保存条件”，进而重复执行写入操作。

解决方案 #

1. 给计数器操作加线程锁 #

使用threading.Lock确保counter的读写是原子操作，避免竞态条件：

import threading
import time

class MQTTClientHandler:
    def __init__(self):
        self.counter = 0
        # 初始化线程锁
        self.lock = threading.Lock()

    def on_message(self, client, userdata, msg):
        timestamp_now = time.time()
        print(timestamp_now)
        # 用锁包裹计数器操作
        with self.lock:
            print(self.counter)
            self.counter += 1

with语句会自动处理锁的获取与释放，确保同一时间只有一个回调能操作counter。

2. 检查MQTT网络循环是否重复启动 #

如果代码中多次调用client.loop_start()，会启动多个网络线程处理消息，直接导致on_message并行执行。确保只启动一次网络循环：

# 正确示例：仅启动一次MQTT网络线程
client = paho.mqtt.client.Client()
client.on_message = handler.on_message
client.connect("localhost", 1883, 60)
# 启动一次即可，不要重复调用
client.loop_start()

3. 优化数据库批量保存逻辑 #

将消息缓存到线程安全队列，单独用一个线程负责定期批量写入数据库，彻底避免回调中的并发问题：

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

class MQTTDataHandler:
    def __init__(self):
        self.data_queue = queue.Queue()
        self.last_save_time = time.time()
        # 启动后台数据库写入线程
        threading.Thread(target=self.db_save_loop, daemon=True).start()

    def parse_msg(self, msg):
        # 解析MQTT消息为可存储的数据格式
        return {
            "topic": msg.topic,
            "payload": msg.payload.decode(),
            "timestamp": time.time()
        }

    def on_message(self, client, userdata, msg):
        data = self.parse_msg(msg)
        # 将数据放入线程安全队列
        self.data_queue.put(data)

    def batch_insert_db(self, batch_data):
        # 实现批量写入数据库的逻辑
        print(f"批量写入{len(batch_data)}条数据到数据库")
        # 示例：替换为你的数据库操作
        # with db_connection.cursor() as cursor:
        #     cursor.executemany("INSERT INTO data_table VALUES (%s, %s, %s)", batch_data)
        #     db_connection.commit()

    def db_save_loop(self):
        while True:
            current_time = time.time()
            # 每60秒触发一次保存，或队列有数据时立即处理
            if current_time - self.last_save_time >= 60 or not self.data_queue.empty():
                batch_data = []
                # 取出队列中所有待保存数据
                while not self.data_queue.empty():
                    batch_data.append(self.data_queue.get())
                if batch_data:
                    self.batch_insert_db(batch_data)
                    self.last_save_time = current_time
            # 每秒检查一次
            time.sleep(1)

# 初始化并启动MQTT客户端
handler = MQTTDataHandler()
client = mqtt.Client()
client.on_message = handler.on_message
client.connect("localhost", 1883, 60)
client.loop_start()

# 保持主线程运行
while True:
    time.sleep(1)

原理说明 #

计数器异常的本质是非原子操作的竞态：self.counter += 1在Python中会拆分为“读取当前值→加1→写入新值”三步，若多个回调在这三步之间切换执行，就会出现重复读取旧值、覆盖新值的情况。线程锁强制这些操作成为一个不可分割的原子单元，确保执行顺序正确。

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