4、MQTT客户端开发（Python）：paho-mqtt库安装、连接Broker、订阅与发布消息、回调函数处理

好，咱们进入实战环节了。前面聊了那么多MQTT协议的原理，说白了都是纸上谈兵。今天咱们就用Python，把MQTT客户端真正跑起来。

我个人习惯用paho-mqtt这个库，它是Eclipse基金会维护的，稳定、文档全，社区也活跃。我在好几个物联网项目里都用它，没出过什么大乱子。

4.1 安装paho-mqtt库

安装很简单，一行命令搞定：

pip install paho-mqtt

嗯，这里要注意。如果你用的是Python 3，建议用pip3而不是pip，避免装到Python 2的环境里。我曾经在一个老服务器上踩过这个坑，折腾了半天才发现装错了版本。

验证安装是否成功：

python -c "import paho.mqtt.client as mqtt; print(mqtt.__version__)"

如果能打印出版本号，说明装好了。我目前用的是1.6.1版本，新版本API基本兼容。

4.2 连接Broker

连接Broker是第一步。说白了，就是让你的设备和MQTT服务器建立TCP连接。

先看一个最基础的连接示例：

import paho.mqtt.client as mqtt

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

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

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

这里我解释几个关键点：

• Client()：创建一个客户端对象。你可以给它起个名字，比如mqtt.Client("my_device_001")，方便调试时识别。
• connect()：三个参数分别是Broker地址、端口、keepalive时间（秒）。keepalive的意思是，如果60秒内没有消息交互，客户端会自动发送PINGREQ保活。
• loop_start()：启动一个后台线程来处理网络流量。如果不调用这个，你的客户端收不到任何消息。

你想想看，如果Broker地址写错了，或者端口被防火墙挡住了，客户端就会一直卡在那里。所以生产环境里，我通常会这样写：

import time

def connect_with_retry(client, broker, port, keepalive, retries=3):
    for i in range(retries):
        try:
            client.connect(broker, port, keepalive)
            print("连接成功")
            return True
        except Exception as e:
            print(f"第{i+1}次连接失败: {e}")
            time.sleep(2)
    return False

4.3 订阅与发布消息

连接上Broker之后，就可以干正事了——订阅和发布。

4.3.1 订阅主题

client.subscribe("sensor/temperature", qos=1)

subscribe()有两个参数：主题名和QoS等级。QoS默认是0，但我建议至少用1，不然丢消息你都不知道。

为什么会这样？因为QoS 0是"最多一次"，发出去就不管了。网络一抖，消息就丢了。我在一个农业大棚项目里吃过这个亏，温度数据丢了一整夜，差点把菜给冻坏了。

4.3.2 发布消息

client.publish("sensor/temperature", "25.3", qos=1)

publish()的三个参数：主题、消息内容、QoS。消息内容必须是字符串或者字节流。如果你要发JSON，记得先json.dumps()序列化。

4.4 回调函数处理

回调函数是MQTT客户端的灵魂。说白了，就是当某个事件发生时，库会自动调用你注册的函数。

常用的回调有这几个：

来看一个完整的示例：

import paho.mqtt.client as mqtt

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

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

def on_disconnect(client, userdata, rc):
    print("连接断开，尝试重连...")
    # 自动重连逻辑可以写在这里

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

client.connect("broker.emqx.io", 1883, 60)
client.loop_forever()

这里有个关键点：loop_forever()会阻塞主线程，适合做长期运行的服务。而loop_start()是后台线程，适合做GUI程序或者需要同时做其他事情的场景。

4.5 完整实战：温湿度传感器模拟

最后，咱们写一个完整的示例。模拟一个温湿度传感器，每5秒发布一次数据，同时订阅控制指令。

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

BROKER = "broker.emqx.io"
PORT = 1883
TOPIC_DATA = "device/sensor/data"
TOPIC_CMD = "device/sensor/cmd"

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("已连接到Broker")
        client.subscribe(TOPIC_CMD, qos=1)
    else:
        print(f"连接失败: {rc}")

def on_message(client, userdata, msg):
    print(f"收到指令: {msg.payload.decode()}")
    # 这里可以解析指令，比如"set_interval:10"来改变采集频率

client = mqtt.Client("sensor_simulator")
client.on_connect = on_connect
client.on_message = on_message

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

try:
    while True:
        data = {
            "temperature": round(random.uniform(20.0, 30.0), 1),
            "humidity": round(random.uniform(40.0, 70.0), 1),
            "timestamp": int(time.time())
        }
        client.publish(TOPIC_DATA, json.dumps(data), qos=1)
        print(f"发布数据: {data}")
        time.sleep(5)
except KeyboardInterrupt:
    print("程序退出")
    client.loop_stop()
    client.disconnect()

这个示例里，我用了loop_start()，因为主循环还要做定时发布。如果你用loop_forever()，就得把发布逻辑放到另一个线程里。

好了，这一章的内容就到这里。你动手试试看，把代码跑起来，感受一下MQTT客户端是怎么工作的。下一章咱们会聊更高级的话题——如何保证消息的可靠性和安全性。