4. MQTT客户端开发（Python）：使用paho-mqtt库

好，咱们进入实战环节了。前面聊了那么多MQTT协议的理论，说白了都是为这一刻做准备。Python开发MQTT客户端，我几乎每个项目都会用到paho-mqtt这个库。它轻量、稳定，而且社区活跃。我个人习惯是，不管项目大小，都优先选它。

4.1 环境准备与安装

先装库。这个没什么好说的，一行命令搞定：

pip install paho-mqtt

嗯，这里要注意一下版本。我遇到过几次因为版本太新导致回调函数签名不兼容的问题。建议用 1.6.1 或以上稳定版。如果你用的是Python 3.8+，基本不会有兼容性问题。

4.2 连接与断开

连接MQTT Broker，说白了就是让客户端和服务器握个手。来看一个最基础的连接示例：

import paho.mqtt.client as mqtt

# 创建客户端实例
client = mqtt.Client()

# 连接到Broker
client.connect("broker.emqx.io", 1883, 60)

# 启动网络循环
client.loop_start()

# ... 做一些事情 ...

# 断开连接
client.disconnect()
client.loop_stop()

你想想看，这里有几个关键点：

• Client ID：如果不传参，paho-mqtt会自动生成一个随机ID。但我在生产环境中从来不用自动生成的，因为断线重连时客户端ID必须唯一。我习惯自己生成，比如 device_001_xxxx 这种格式。
• 端口号：1883是明文端口，8883是加密端口。如果你传输的是敏感数据，比如设备控制指令，一定要用TLS加密。我踩过这个坑，有一次设备上报的GPS数据被中间人截获了，从那以后所有生产环境我都强制用8883。
• Keep Alive：第三个参数60表示心跳间隔。这个值设得太小会浪费带宽，设得太大又可能检测不到断线。我个人建议30-60秒之间。

4.3 订阅与发布

连接建立之后，就可以开始真正的数据交互了。订阅和发布是MQTT最核心的两个操作。

4.3.1 订阅主题

def on_connect(client, userdata, flags, rc):
    print("连接成功，返回码:", rc)
    # 订阅主题
    client.subscribe("sensor/temperature")
    client.subscribe("sensor/humidity", qos=1)

client.on_connect = on_connect

这里有个细节：订阅操作最好放在 on_connect 回调里。为什么？因为如果网络断开重连，MQTT协议会自动重新建立连接，但不会自动重新订阅。你想想看，如果订阅逻辑写在主程序里，断线重连后设备就收不到消息了。我曾经因为这个原因，导致一批温湿度传感器离线了整整一个晚上才被发现。

4.3.2 发布消息

import json

# 构造消息
payload = {
    "device_id": "sensor_001",
    "temperature": 25.6,
    "humidity": 68.3,
    "timestamp": "2024-01-15T10:30:00Z"
}

# 发布消息
client.publish("sensor/temperature", json.dumps(payload), qos=1)

发布时我建议：

• 消息体用JSON格式，方便解析和扩展
• QoS等级根据业务需求选择。控制指令用QoS 2，数据上报用QoS 1，日志类用QoS 0
• 保留消息（Retain）慎用。我见过有人把所有消息都设成保留，结果Broker内存爆了

4.4 回调函数处理

回调函数是paho-mqtt的灵魂。说白了，就是当某个事件发生时，库会调用你预先注册的函数。常用的回调有这几个：

回调名称	触发时机	典型用途
on_connect	客户端连接到Broker时	订阅主题、记录日志
on_disconnect	客户端断开连接时	清理资源、触发重连
on_message	收到订阅主题的消息时	处理业务逻辑
on_subscribe	订阅成功时	确认订阅状态
on_publish	消息发布完成时	确认发送成功

来看一个完整的回调示例：

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("连接成功")
        client.subscribe("sensor/#")
    else:
        print(f"连接失败，返回码: {rc}")

def on_message(client, userdata, msg):
    print(f"收到消息: {msg.topic} -> {msg.payload.decode()}")
    # 这里可以加业务逻辑，比如解析JSON、存入数据库等

def on_disconnect(client, userdata, rc):
    print("连接断开，尝试重连...")
    # 自动重连逻辑
    while True:
        try:
            client.reconnect()
            break
        except:
            time.sleep(5)

client.on_connect = on_connect
client.on_message = on_message
client.on_disconnect = on_disconnect

4.5 完整示例：温湿度传感器客户端

最后，我把上面所有知识点串起来，给你一个可以直接运行的示例：

import paho.mqtt.client as mqtt
import json
import time
import random

BROKER = "broker.emqx.io"
PORT = 1883
TOPIC = "sensor/temperature"
CLIENT_ID = f"python_mqtt_{random.randint(1000, 9999)}"

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print(f"客户端 {CLIENT_ID} 连接成功")
        client.subscribe("sensor/control")
    else:
        print(f"连接失败，错误码: {rc}")

def on_message(client, userdata, msg):
    print(f"收到控制指令: {msg.payload.decode()}")

def on_disconnect(client, userdata, rc):
    print("连接断开，5秒后重连...")
    time.sleep(5)

client = mqtt.Client(CLIENT_ID)
client.on_connect = on_connect
client.on_message = on_message
client.on_disconnect = on_disconnect

client.connect(BROKER, PORT, 60)
client.loop_start()

try:
    while True:
        payload = {
            "temperature": round(random.uniform(20, 30), 1),
            "humidity": round(random.uniform(40, 70), 1),
            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
        }
        client.publish(TOPIC, json.dumps(payload), qos=1)
        print(f"发布数据: {payload}")
        time.sleep(5)
except KeyboardInterrupt:
    print("程序退出")
    client.loop_stop()
    client.disconnect()

这个示例我用了很多次，基本覆盖了日常开发中80%的场景。你运行一下试试，应该能看到客户端每隔5秒上报一次温湿度数据。

好了，MQTT客户端开发的核心内容就这些。说白了就是三板斧：连接、订阅/发布、回调处理。你把这几个点吃透了，大部分物联网项目都能搞定。下一章咱们聊聊数据持久化和存储策略，到时候见。