📚 课程目录第4章 QoS 0/1/2 消息质量测试

4. QoS 0/1/2 消息质量测试

任务目标

本案例将使用 Mosquitto 作为 MQTT Broker，通过 Python 的 paho-mqtt 客户端分别以 QoS 0、1、2 发布消息，并验证不同服务质量等级下消息的送达行为与确认机制差异。

环境准备

安装 Mosquitto Broker：sudo apt install mosquitto mosquitto-clients（或对应平台安装方式）
安装 Python 依赖：pip install paho-mqtt
启动 Broker：mosquitto -v（默认监听 1883 端口）

完整实现代码

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

BROKER = "127.0.0.1"
PORT = 1883
TOPIC = "test/qos"
MESSAGES = [f"msg-{i}" for i in range(5)]

# 用于统计各 QoS 下收到的消息
received = {"qos0": [], "qos1": [], "qos2": []}
pub_results = {}

def on_connect(client, userdata, flags, rc, properties=None):
    print(f"Connected with result code {rc}")

def on_message(client, userdata, msg):
    qos_key = userdata["qos_key"]
    received[qos_key].append(msg.payload.decode())
    print(f"[{qos_key}] Received: {msg.payload.decode()} (msg QoS: {msg.qos})")

def on_publish(client, userdata, mid, reason_code=None, properties=None):
    qos_key = userdata["qos_key"]
    pub_results[qos_key] = pub_results.get(qos_key, 0) + 1
    print(f"[{qos_key}] Published mid={mid}")

def run_qos_test(qos_level):
    qos_key = f"qos{qos_level}"
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, userdata={"qos_key": qos_key})
    client.on_connect = on_connect
    client.on_message = on_message
    client.on_publish = on_publish

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

    # 等待连接建立
    time.sleep(0.5)

    # 订阅主题（使用相同 QoS）
    client.subscribe(TOPIC, qos=qos_level)

    # 等待订阅完成
    time.sleep(0.5)

    # 发布 5 条消息
    for i, msg in enumerate(MESSAGES):
        info = client.publish(TOPIC, msg, qos=qos_level)
        info.wait_for_publish(2)  # 等待发布完成（QoS 1/2 会等待确认）
        time.sleep(0.2)

    # 等待消息回传
    time.sleep(1.0)

    client.loop_stop()
    client.disconnect()

def main():
    for qos in [0, 1, 2]:
        print(f"\n===== Testing QoS {qos} =====")
        run_qos_test(qos)

    print("\n===== Summary =====")
    for qos_key, msgs in received.items():
        expected = len(MESSAGES)
        actual = len(msgs)
        print(f"{qos_key}: received {actual}/{expected} messages")
        if msgs != MESSAGES:
            print(f"  Order mismatch or missing: {msgs}")

if __name__ == "__main__":
    main()

运行方式

确保 Mosquitto Broker 已在本地运行（mosquitto -v）
保存上述代码为 qos_test.py
执行：python3 qos_test.py

验证方法与判定标准

程序依次对 QoS 0、1、2 各执行一轮测试，每轮发布 5 条消息并统计回传结果。判定标准如下：

QoS 0：消息可能全部到达，也可能因网络拥塞丢失。本地回环测试通常全部到达，但 on_publish 回调在 QoS 0 下也会触发（表示已交给 Broker 发送，不代表送达确认）。
QoS 1：至少一次送达，可能重复。本地测试通常 5/5 到达且无重复。
QoS 2：恰好一次送达，无重复。本地测试应 5/5 到达且顺序一致。

观察程序输出的 Summary 部分：若各 QoS 均收到 5/5 条消息且顺序正确，说明本地回环下三种 QoS 都能可靠送达。要观察差异，可在发布端与订阅端之间人为增加网络延迟或丢包（如使用 tc netem），此时 QoS 0 会出现丢消息，QoS 1/2 仍能保证送达。

关键说明

QoS 0 的 on_publish 回调仅表示消息已交给 Broker 的 socket 发送，不保证对端收到。
QoS 1 通过 PUBACK 确认，QoS 2 通过 PUBREC/PUBREL/PUBCOMP 四次握手完成确认，因此发布耗时依次增加。
本测试中订阅与发布使用相同 QoS 等级，实际应用中可分别指定。