3. Python Paho客户端连接测试

任务目标

使用Python Paho MQTT客户端库，实现一个最小可运行的发布/订阅闭环：订阅端连接到公共Broker并订阅主题，发布端向同一主题发送消息，订阅端收到消息后打印内容。通过该案例验证Paho库的基本连接、订阅、发布与回调机制。

环境准备

安装Paho MQTT客户端（支持Python 3.7+）：

pip install paho-mqtt

本案例使用公共测试Broker：broker.emqx.io，端口1883，无需认证。若网络受限，可替换为本地部署的Mosquitto（默认端口1883）。

完整实现代码

以下代码包含订阅端与发布端，通过两个线程在同一个进程中运行，形成完整闭环。实际使用时可将两端拆分为独立脚本。

import threading
import time
import paho.mqtt.client as mqtt

BROKER = "broker.emqx.io"
PORT = 1883
TOPIC = "paho/test/connect"
QOS = 1

# ---------- 订阅端 ----------
received_messages = []

def on_connect(client, userdata, flags, rc, properties=None):
    if rc == 0:
        print(f"[Subscriber] 连接成功，订阅主题: {TOPIC}")
        client.subscribe(TOPIC, qos=QOS)
    else:
        print(f"[Subscriber] 连接失败，返回码: {rc}")

def on_message(client, userdata, msg):
    payload = msg.payload.decode()
    print(f"[Subscriber] 收到消息: {payload} (QoS={msg.qos})")
    received_messages.append(payload)

def run_subscriber():
    sub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
    sub.on_connect = on_connect
    sub.on_message = on_message
    sub.connect(BROKER, PORT, keepalive=60)
    sub.loop_forever()

# ---------- 发布端 ----------
def run_publisher():
    time.sleep(1)  # 等待订阅端完成连接
    pub = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
    pub.connect(BROKER, PORT, keepalive=60)
    pub.loop_start()
    for i in range(3):
        message = f"Hello MQTT #{i+1}"
        result = pub.publish(TOPIC, payload=message, qos=QOS)
        result.wait_for_publish()
        print(f"[Publisher] 已发布: {message}")
        time.sleep(1)
    pub.loop_stop()
    pub.disconnect()

# ---------- 启动 ----------
if __name__ == "__main__":
    sub_thread = threading.Thread(target=run_subscriber, daemon=True)
    sub_thread.start()
    run_publisher()
    time.sleep(2)  # 等待最后一条消息处理完成
    print(f"\n验证结果: 共收到 {len(received_messages)} 条消息")
    if len(received_messages) == 3:
        print("测试通过：发布与订阅闭环正常")
    else:
        print("测试失败：消息数量不匹配")

运行与验证

运行方式：将上述代码保存为 paho_test.py，直接执行：

python paho_test.py

预期输出（确定性结果）：由于代码逻辑明确，输出应严格为以下顺序：

[Subscriber] 连接成功，订阅主题: paho/test/connect
[Publisher] 已发布: Hello MQTT #1
[Subscriber] 收到消息: Hello MQTT #1 (QoS=1)
[Publisher] 已发布: Hello MQTT #2
[Subscriber] 收到消息: Hello MQTT #2 (QoS=1)
[Publisher] 已发布: Hello MQTT #3
[Subscriber] 收到消息: Hello MQTT #3 (QoS=1)

验证结果: 共收到 3 条消息
测试通过：发布与订阅闭环正常

通过判定标准：程序退出码为0，且控制台打印“测试通过：发布与订阅闭环正常”。若网络延迟导致消息顺序略有变化，只要最终收到3条消息且内容匹配即为通过。

常见问题

• 连接失败（返回码≠0）：检查网络能否访问 broker.emqx.io:1883，可尝试改用 test.mosquitto.org 或本地Broker地址。
• 收不到消息：确认订阅端已连接成功后再启动发布端（代码中已通过time.sleep(1)保证）；检查主题名称是否完全一致（区分大小写）。
• 回调不触发：确保使用 mqtt.CallbackAPIVersion.VERSION2 并保持 loop_forever() 或 loop_start() 运行，回调函数签名需与示例一致。