Python中接收MQTT消息后重连MySQL的最佳实践

你遇到的2006, 'MySQL server has gone away'问题太常见了——本质就是MySQL服务器会主动断开长时间闲置的连接，而你一开始创建的长连接在消息间隔久了之后就失效了。增加超时时间确实是饮鸩止渴，毕竟你没法预估消息什么时候来。你想的「收到消息时再创建连接」完全是可行的，而且是这类场景的推荐实践之一，下面给你两种靠谱的方案：

方案一：每次收到消息时创建新连接（最稳妥） #

这种方式完全规避长连接的问题，每次处理消息时才建立数据库连接，操作完成后立即关闭。你之前的问题是变量作用域，那我们把数据库连接的逻辑封装成独立函数，在on_message里调用就行。

修改后的完整代码示例：

import paho.mqtt.client as mqtt
import MySQLdb
import json

# 封装数据库插入函数
def insert_data_into_mysql(some_value):
    db = None
    cursor = None
    try:
        # 每次调用都新建连接
        db = MySQLdb.connect(
            host="localhost",
            user="admin",
            passwd="somepass",
            db="mydb"
        )
        cursor = db.cursor()
        sql = """INSERT INTO data(something) VALUES (%s)"""
        cursor.execute(sql, (some_value,))  # 注意：单个参数要传元组，逗号不能少
        db.commit()  # 务必提交事务，否则数据不会写入数据库
        print("数据插入成功")
    except MySQLdb.Error as e:
        print(f"数据库错误: {e}")
        if db:
            db.rollback()
    except Exception as e:
        print(f"其他错误: {e}")
    finally:
        # 确保游标和连接一定会关闭，避免资源泄漏
        if cursor:
            cursor.close()
        if db:
            db.close()

# MQTT消息回调函数
def on_message(client, userdata, msg):
    try:
        # 解析MQTT消息中的JSON数据
        payload = json.loads(msg.payload.decode('utf-8'))
        some_value = payload.get("something")  # 根据你的实际JSON结构取值
        if some_value:
            insert_data_into_mysql(some_value)
        else:
            print("消息中未找到目标字段")
    except json.JSONDecodeError as e:
        print(f"JSON解析错误: {e}")

# 初始化MQTT客户端
def init_mqtt_client():
    client = mqtt.Client()
    client.on_message = on_message
    # 替换为你的MQTT broker地址和端口
    client.connect("mqtt_broker_host", 1883, 60)
    # 替换为你要订阅的主题
    client.subscribe("your_target_topic")
    return client

if __name__ == "__main__":
    client = init_mqtt_client()
    client.loop_forever()

方案优势： #

• 彻底摆脱连接失效的困扰，每次都是全新的有效连接
• 代码职责分离清晰，MQTT逻辑和数据库操作互不干扰
• 无需处理复杂的连接重连逻辑，新手也能快速上手

注意事项： #

• 一定要执行db.commit()提交事务，否则数据只会停留在内存中，不会写入数据库
• 传入SQL的参数必须是元组（哪怕只有一个值，也要加逗号写成(some_value,)），否则MySQLdb会把字符串当成序列处理，触发参数数量不匹配的错误
• 用finally块确保游标和连接一定会关闭，避免资源泄漏

方案二：复用连接，但操作前自动重连（适合高频率消息） #

如果你的消息频率很高，每次创建连接会产生额外开销，那可以复用连接，但每次操作前先检查连接有效性，失效则自动重连。这种方式可以把连接和游标作为全局变量（或封装成类），示例如下：

import paho.mqtt.client as mqtt
import MySQLdb
import json

# 全局数据库连接和游标
db = None
cursor = None

# 初始化或重连数据库
def get_valid_db_connection():
    global db, cursor
    try:
        if db:
            # 检查连接是否有效，ping失败会自动重连
            db.ping(reconnect=True)
        else:
            # 新建连接
            db = MySQLdb.connect(
                host="localhost",
                user="admin",
                passwd="somepass",
                db="mydb"
            )
            cursor = db.cursor()
        return db, cursor
    except MySQLdb.Error as e:
        print(f"数据库连接失败: {e}")
        # 连接失败后重置，下次重新创建
        db = None
        cursor = None
        raise

# MQTT消息回调函数
def on_message(client, userdata, msg):
    try:
        payload = json.loads(msg.payload.decode('utf-8'))
        some_value = payload.get("something")
        if not some_value:
            print("消息中未找到目标字段")
            return
        
        # 获取有效连接
        db_conn, db_cursor = get_valid_db_connection()
        sql = """INSERT INTO data(something) VALUES (%s)"""
        db_cursor.execute(sql, (some_value,))
        db_conn.commit()
        print("数据插入成功")
    except json.JSONDecodeError as e:
        print(f"JSON解析错误: {e}")
    except MySQLdb.Error as e:
        print(f"数据库操作错误: {e}")
        if db_conn:
            db_conn.rollback()
    except Exception as e:
        print(f"其他错误: {e}")

# 初始化MQTT客户端
def init_mqtt_client():
    client = mqtt.Client()
    client.on_message = on_message
    client.connect("mqtt_broker_host", 1883, 60)
    client.subscribe("your_target_topic")
    return client

if __name__ == "__main__":
    client = init_mqtt_client()
    client.loop_forever()

方案优势： #

• 减少连接创建的开销，适合消息频繁的场景
• db.ping(reconnect=True)会自动处理连接失效的情况，不用手动编写重连逻辑

注意事项： #

• 全局变量在多线程环境下要注意线程安全，如果使用loop_start()而非loop_forever()，需要加锁保护数据库操作
• 同样要记得提交事务和异常回滚

为什么你之前的尝试没成功？ #

你之前把连接逻辑放进on_message但仍报错，大概率是这几个原因：

1. 没有正确处理连接的创建和关闭，比如with db:的用法在部分MySQLdb版本中不支持（需要用游标作为上下文管理器）
2. 变量作用域问题：如果db和cursor是函数外定义的，函数内需要用global声明，否则会被当成局部变量处理
3. 缺少事务提交操作，导致数据没写入，误以为是连接问题

总结 #

• 如果消息频率不高，方案一是首选，简单可靠，不用操心连接维护
• 如果消息频率很高，方案二更高效，但要注意线程安全问题

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