如何在Python中使用Paho MQTT实现多事件处理

咱们这个实现把MQTT核心事件（连接、订阅、消息接收、断开重连）都做了完整处理，同时结合Arduino实现了传感器数据发布和远程引脚控制，直接就能拿来用，我给你梳理下细节：

完整实现代码 #

#!/usr/bin/env python
import sys
from arduino.Arduino import Arduino
import paho.mqtt.client as mqtt
import serial
import time

# 基础配置参数
PIN_LED = 13
BROKER_ADDRESS = "10.0.2.190"
SERIAL_PORT = '/dev/ttyACM0'
BAUD_RATE = 9600
# 定义要订阅和发布的MQTT主题
MQTT_TOPICS = [
    ('/Deutschland/Osnabrueck/Coffee-bike-1/Control/LED', 0),  # 控制LED的主题
    ('/Deutschland/Osnabrueck/Coffee-bike-1/Sensor_1', 0)     # 传感器数据主题
]

# 添加Arduino库路径（根据你的实际路径调整）
sys.path.append("/home/hu/Schreibtisch/Arduino_BA_2.0/Probe_Programmierung/Python-Arduino-Proto-API-v2/arduino")

# 初始化Arduino和串口通信
b = Arduino(SERIAL_PORT)
b.output([PIN_LED])
b.setLow(PIN_LED)
ser = serial.Serial(SERIAL_PORT, BAUD_RATE)
time.sleep(2)  # 给串口初始化留缓冲时间

# ---------------------- MQTT多事件回调函数 ----------------------
def on_connect(client, userdata, flags, rc):
    """MQTT连接成功/失败的回调"""
    if rc == 0:
        print("✅ 成功连接到MQTT Broker")
        # 连接成功后自动订阅预设主题
        client.subscribe(MQTT_TOPICS)
    else:
        print(f"❌ 连接失败，错误码: {rc}")

def on_subscribe(client, userdata, mid, granted_qos):
    """订阅主题成功后的回调"""
    print(f"✅ 成功订阅主题，QoS等级: {granted_qos}")

def on_message(client, userdata, msg):
    """收到MQTT消息后的处理逻辑"""
    topic = msg.topic
    payload = msg.payload.decode('utf-8')
    print(f"📩 收到消息 - 主题: {topic}, 内容: {payload}")
    
    # 根据不同主题执行对应动作，比如控制LED
    if topic == '/Deutschland/Osnabrueck/Coffee-bike-1/Control/LED':
        if payload.lower() == 'on':
            b.setHigh(PIN_LED)
            print("💡 LED已打开")
        elif payload.lower() == 'off':
            b.setLow(PIN_LED)
            print("💡 LED已关闭")

def on_disconnect(client, userdata, rc):
    """意外断开连接后的处理（自动尝试重连）"""
    if rc != 0:
        print("⚠️ 意外断开与MQTT Broker的连接，正在尝试重连...")
        client.reconnect()

# ---------------------- 传感器数据读取函数 ----------------------
def read_gas_sensor():
    """从串口读取气体传感器数据"""
    try:
        if ser.in_waiting > 0:
            sensor_data = ser.readline().decode('utf-8').strip()
            return sensor_data
        return "无传感器数据"
    except Exception as e:
        print(f"❌ 读取传感器失败: {e}")
        return "读取失败"

# ---------------------- 主程序逻辑 ----------------------
if __name__ == "__main__":
    # 初始化MQTT客户端
    client = mqtt.Client()
    
    # 绑定所有事件回调
    client.on_connect = on_connect
    client.on_subscribe = on_subscribe
    client.on_message = on_message
    client.on_disconnect = on_disconnect
    
    # 尝试连接MQTT Broker
    try:
        client.connect(BROKER_ADDRESS, 1883, 60)
    except Exception as e:
        print(f"❌ 无法连接到Broker: {e}")
        sys.exit(1)
    
    # 启动后台线程处理MQTT网络流量（非阻塞模式）
    client.loop_start()
    
    # 循环发布传感器数据
    try:
        while True:
            gassensor_value = read_gas_sensor()
            # 发布传感器数据到指定主题
            client.publish('/Deutschland/Osnabrueck/Coffee-bike-1/Sensor_1', gassensor_value)
            print(f"📤 发布传感器数据: {gassensor_value}")
            time.sleep(5)  # 每5秒发布一次数据
    except KeyboardInterrupt:
        print("\n🔄 用户终止程序，正在清理资源...")
        b.setLow(PIN_LED)
        ser.close()
        client.loop_stop()
        client.disconnect()

关键功能说明 #

• 多事件覆盖：把MQTT从连接到断开的全流程事件都做了处理，不管是连接成功后的自动订阅，还是意外断连的自动重连，都不用手动操心
• 双向交互：既可以定时读取Arduino传感器数据发布到MQTT，也能通过MQTT消息远程控制Arduino的LED引脚
• 异常处理：针对串口读取、Broker连接都做了异常捕获，程序稳定性更高

注意事项 #

1. 确保Arduino端已经上传了对应串口输出传感器数据的代码
2. 检查MQTT Broker的地址和端口是否和你的环境匹配
3. 串口权限问题：如果出现Permission denied错误，执行sudo chmod 666 /dev/ttyACM0赋予权限
4. 依赖安装：确保安装了需要的库，执行pip install paho-mqtt pyserial即可

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