Python MQTT接收传感器数据存DataFrame定期存CSV无报错停收问题

• Paho MQTT的on_message回调运行在客户端独立的网络线程中，你在回调内执行的pandas赋值、本地文件写入操作属于阻塞类操作，且未做异常捕获：一旦执行过程中抛出任何异常，回调线程会直接静默退出，仅保留主进程运行，表现为程序存活但不再接收消息。
• 全局变量bufferDF、counter未加线程锁保护，多线程读写时容易出现竞态条件，引发死锁或赋值异常。
• 代码存在多处未显式处理的隐患：未导入os模块、json.loads直接读取message对象而非其payload属性、InitComm方法缺少语法冒号、未实现连接断开自动重连逻辑，任意一处触发异常都会直接终止消息接收流程。

1. 新增异常捕获逻辑，避免回调线程静默退出
   所有on_message内的业务逻辑用try-except包裹，打印异常栈便于排查。
2. 移除全局变量，改用类实例属性+线程锁保护
   将bufferDF、counter移入DataSaver类作为实例属性，读写时加线程锁，避免竞态问题。
3. 补全MQTT连接异常处理逻辑
   新增on_connect、on_disconnect回调，实现断开自动重连，实时感知连接状态。
4. 优化回调逻辑，避免阻塞网络线程
   如果后续业务复杂度继续提升，建议引入队列将消息转交给独立工作线程处理，on_message仅负责消息入队，不做任何阻塞操作。

import pandas as pd
import json
import paho.mqtt.client as mqtt
import os
import threading
from datetime import timedelta

# 补全你缺失的初始化周时间范围函数
def InitDateRange(current_ts):
    # 按周生成时间序列，步长15分钟，可根据你的实际需求调整
    week_start = current_ts.floor('W')
    week_end = week_start + timedelta(days=7) - timedelta(minutes=15)
    return pd.date_range(start=week_start, end=week_end, freq='15T')

class DataSaver():
    def __init__(self, filesfolderpath, sensorslist):
        self.filesfolderpath = filesfolderpath
        self.sensorslist = sensorslist
        self.client = None
        # 替换全局变量为实例属性
        self.bufferDF = None
        self.counter = 1
        # 新增线程锁保护数据读写
        self.lock = threading.Lock()

    def SaveSensorRead(self, client, userdata, message):
        try:
            message_dict = json.loads(message.payload.decode())
            timestamp = pd.to_datetime(message_dict["timestamp"])
            sensorname = message_dict["sensorname"]
            read = message_dict["read"]

            # 加锁保护数据操作
            with self.lock:
                if self.bufferDF is None:
                    daterange = InitDateRange(timestamp)
                    self.bufferDF = pd.DataFrame(index=daterange, columns=self.sensorslist)
                
                if timestamp > max(self.bufferDF.index):
                    filename = f"week{self.counter}.csv"
                    self.bufferDF.to_csv(os.path.join(self.filesfolderpath, filename))
                    daterange = InitDateRange(timestamp)
                    self.bufferDF = pd.DataFrame(index=daterange, columns=self.sensorslist)
                    self.counter += 1
                
                self.bufferDF.loc[timestamp, sensorname] = read
        except Exception as e:
            # 打印异常，避免线程静默退出
            print(f"消息处理出错: {str(e)}")
            import traceback
            traceback.print_exc()
    
    def on_connect(self, client, userdata, flags, rc):
        print(f"MQTT连接成功，返回码: {rc}")
        # 连接成功后重新订阅，避免重连后订阅丢失
        client.subscribe(self.channelname)
    
    def on_disconnect(self, client, userdata, rc):
        print(f"MQTT连接断开，返回码: {rc}，尝试重连")
        if rc != 0:
            # 非主动断开的情况，自动重连
            try:
                client.reconnect()
            except Exception as e:
                print(f"重连失败: {str(e)}")

    def InitComm(self, brokerip, channelname):
        self.channelname = channelname
        self.client = mqtt.Client("client")
        self.client.on_message = self.SaveSensorRead
        self.client.on_connect = self.on_connect
        self.client.on_disconnect = self.on_disconnect
        # 设置自动重连间隔
        self.client.reconnect_delay_set(min_delay=1, max_delay=30)
        self.client.connect(brokerip, 1883, keepalive=60)
        self.client.loop_start()

# 调用示例
# filesfolderpath = "./sensor_data"
# sensorslist = ["sensor1", "sensor2", "sensorxx"]
# brokerip = "127.0.0.1"
# channelname = "sensor/topic"
# saver = DataSaver(filesfolderpath, sensorslist)
# saver.InitComm(brokerip, channelname)

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