3. 传感器客户端QoS等级选择与配置

为什么矿山传感器不能统一用一个 QoS

QoS 决定的是“消息从发布端到 Broker、再到订阅端”这一段的投递保证强度，而不是数据本身的价值。矿山井下不同传感器的数据特征差异很大：瓦斯浓度报警要求“绝不能丢、且不能重复触发误动作”，水位监测要求“不能丢、偶发重复可接受”，而风速趋势、温度巡检这类高频数据更看重“新鲜度”，旧值补发反而没有意义。因此本例按“实时性 vs 可靠性”两个维度为不同主题分配 QoS，并在客户端代码中落地。

按数据语义分配 QoS 的判定规则

选择 QoS 的核心不是“越高越安全”，而是看三个问题：丢一条会不会造成安全后果、重复一条会不会造成误动作、旧数据补发还有没有价值。据此得到本例的分配表：

客户端配置：发布端按主题绑定 QoS

下面用 Python 的 paho-mqtt 编写一个可运行的发布端，把“主题 → QoS”映射写成常量表，避免在业务代码里散落魔法数字。示例参数为教学假设值，实际部署时按现场链路质量调整。

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

BROKER = "192.168.10.20"      # 矿山本地 Broker 地址（示例）
PORT = 1883
CLIENT_ID = "sensor-gw-A1"

# 主题 -> QoS 映射：与上面的分配表一一对应
TOPIC_QOS = {
    "mine/gas/A1/CH4/alarm":      2,
    "mine/water/B2/level":        1,
    "mine/fan/C3/wind/speed":     0,
    "mine/temp/D4/value":         0,
}

def on_connect(client, userdata, flags, rc):
    print("connected rc=", rc)

def on_publish(client, userdata, mid):
    print("puback mid=", mid)

client = mqtt.Client(client_id=CLIENT_ID, clean_session=False)
client.username_pw_set("sensor_gw", "Gw@2024")
client.on_connect = on_connect
client.on_publish = on_publish
client.connect(BROKER, PORT, keepalive=30)
client.loop_start()

def publish(topic, payload):
    qos = TOPIC_QOS[topic]
    info = client.publish(topic, json.dumps(payload), qos=qos)
    # QoS 1/2 时等待 PUBACK/PUBCOMP，确认本次投递完成
    if qos > 0:
        info.wait_for_publish(timeout=5)
    return info.rc

# 模拟一轮采集
publish("mine/gas/A1/CH4/alarm", {"ch4": 1.12, "ts": int(time.time()), "level": "ALARM"})
publish("mine/water/B2/level",  {"level_m": 2.35, "ts": int(time.time())})
publish("mine/fan/C3/wind/speed", {"mps": 3.8, "ts": int(time.time())})
publish("mine/temp/D4/value",   {"c": 26.4, "ts": int(time.time())})

time.sleep(1)
client.loop_stop()
client.disconnect()

关键点：clean_session=False 配合固定 client_id，使 QoS 1/2 的会话状态在断线重连后仍可续传；wait_for_publish 只对 QoS>0 有意义，QoS 0 调用它会立即返回，不能当作“已送达”的证据。

订阅端配置：订阅 QoS 必须与发布端对齐

MQTT 的实际投递等级取 min(发布 QoS, 订阅 QoS)。如果发布端用 QoS 2 而订阅端只订阅 QoS 0，报警仍会以 QoS 0 下发，等于白配。因此订阅端要按主题分别指定：

import paho.mqtt.client as mqtt

SUB_QOS = [
    ("mine/gas/A1/CH4/alarm", 2),
    ("mine/water/B2/level",   1),
    ("mine/fan/C3/wind/speed", 0),
    ("mine/temp/D4/value",    0),
]

def on_connect(client, userdata, flags, rc):
    for topic, qos in SUB_QOS:
        client.subscribe(topic, qos=qos)
        print(f"subscribe {topic} qos={qos}")

def on_message(client, userdata, msg):
    print(f"[qos={msg.qos}] {msg.topic} -> {msg.payload.decode()}")

c = mqtt.Client(client_id="monitor-ground-01", clean_session=False)
c.username_pw_set("monitor", "Mon@2024")
c.on_connect = on_connect
c.on_message = on_message
c.connect("192.168.10.20", 1883, keepalive=30)
c.loop_forever()

连接测试与结果判读

配置完成后需要验证“实际生效的 QoS”而不是“代码里写的 QoS”。用 mosquitto_sub 带 -v 与调试输出观察每条消息的 QoS 字段，同时用 mosquitto_pub 分别以不同 QoS 发布，形成对照：

# 终端 1：订阅并显示 QoS（-d 打印协议交互，-q 指定订阅 QoS）
mosquitto_sub -h 192.168.10.20 -u monitor -P 'Mon@2024' \
  -t 'mine/gas/A1/CH4/alarm' -q 2 -d -v

# 终端 2：以 QoS 2 发布报警
mosquitto_pub -h 192.168.10.20 -u sensor_gw -P 'Gw@2024' \
  -t 'mine/gas/A1/CH4/alarm' -q 2 -m '{"ch4":1.12,"level":"ALARM"}'

# 对照：订阅端故意降为 QoS 0，观察实际收到的 QoS
mosquitto_sub -h 192.168.10.20 -u monitor -P 'Mon@2024' \
  -t 'mine/gas/A1/CH4/alarm' -q 0 -v

判读标准：

• 订阅端 -q 2 时，-d 输出中应能看到 PUBLISH 与 PUBREC/PUBREL/PUBCOMP 的往返，消息 QoS 字段为 2。
• 订阅端 -q 0 时，即使发布端是 QoS 2，收到的消息 QoS 字段会降为 0，说明“订阅端降级会削弱保证”，需要修正订阅配置。
• 断开订阅端网络 10 秒后恢复，QoS 1/2 主题应能补收到断线期间的消息（受 Broker 持久化与 max_queued_messages 限制），QoS 0 主题则不会补发——这正是“旧值无意义”的预期行为。