2. 发布与订阅基础消息

任务目标

使用 MQTT 协议建立两个客户端：一个作为发布者向指定主题发送消息，另一个作为订阅者接收该主题的消息，验证消息能够实时从发布端传递到订阅端。

环境依赖

本案例使用 paho-mqtt 客户端库，需要先安装：

pip install paho-mqtt

同时需要一个可用的 MQTT Broker。本案例默认使用公共 Broker broker.emqx.io（端口 1883），无需本地搭建服务端。

完整实现代码

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

BROKER = "broker.emqx.io"
PORT = 1883
TOPIC = "demo/iot/basic"
QOS = 1

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

def on_connect_sub(client, userdata, flags, rc):
    if rc == 0:
        print("[订阅端] 已连接 Broker")
        client.subscribe(TOPIC, qos=QOS)
        print(f"[订阅端] 已订阅主题: {TOPIC}")
    else:
        print(f"[订阅端] 连接失败，返回码: {rc}")

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

def run_subscriber():
    sub_client = mqtt.Client(client_id="subscriber_demo_001")
    sub_client.on_connect = on_connect_sub
    sub_client.on_message = on_message
    sub_client.connect(BROKER, PORT, keepalive=60)
    sub_client.loop_forever()

# ---------- 发布端 ----------
def run_publisher():
    time.sleep(1)  # 等待订阅端完成连接与订阅
    pub_client = mqtt.Client(client_id="publisher_demo_001")
    pub_client.connect(BROKER, PORT, keepalive=60)
    pub_client.loop_start()

    messages = ["Hello MQTT", "IoT 实战案例", "消息实时接收验证"]
    for i, msg in enumerate(messages):
        result = pub_client.publish(TOPIC, payload=msg, qos=QOS)
        result.wait_for_publish()
        print(f"[发布端] 已发送: {msg}")
        time.sleep(1)

    pub_client.loop_stop()
    pub_client.disconnect()

# ---------- 启动 ----------
if __name__ == "__main__":
    sub_thread = threading.Thread(target=run_subscriber, daemon=True)
    sub_thread.start()

    run_publisher()

    # 等待消息处理完成
    time.sleep(2)
    print("\n===== 验证结果 =====")
    if len(received_messages) == 3:
        print("通过：订阅端成功接收到全部 3 条消息")
        for m in received_messages:
            print(f"  - {m}")
    else:
        print(f"失败：预期 3 条，实际收到 {len(received_messages)} 条")

运行方式

python mqtt_pub_sub.py

验证方法与判定标准

• 预期输出：程序运行后，订阅端先显示连接成功并订阅主题，随后发布端依次发送 3 条消息，订阅端控制台同步打印出收到的每条消息内容。
• 通过条件：最终输出“通过：订阅端成功接收到全部 3 条消息”，且下方列出与发布端发送内容完全一致的 3 条消息文本。
• 实时性验证：观察发布端发送与订阅端接收的打印时间差，两者间隔应小于 1 秒（受网络延迟影响），说明消息是实时推送而非轮询获取。

关键说明

• 代码中通过 threading 让订阅端在后台线程运行 loop_forever()，主线程执行发布逻辑，从而在单进程中模拟两个独立客户端。
• 发布端使用 loop_start() 启动网络循环，发送完成后调用 loop_stop() 清理资源。
• 若使用本地 Broker，只需将 BROKER 变量改为 127.0.0.1 并确保服务已启动即可，其余代码无需修改。
• 公共 Broker 可能因网络原因偶发连接失败，若出现连接错误可稍后重试或更换为 test.mosquitto.org。