MQTT Is Not a Message Queue: QoS, Persistent Sessions and Shared Subscriptions Without Losing Data

本文为原文前 6,000 字符的节选翻译，完整内容请查看原文。

When I first started working with IoT, my team made a very common mistake: treating an MQTT broker like RabbitMQ. Devices published sensor data, and a Python worker subscribed to it and wrote it to a database. Everything ran smoothly until the worker was restarted during a deployment. Exactly 40 seconds of downtime resulted in the loss of all data from over 200 devices. There were no error logs, no dead letters; the data simply vanished. The reason is simple: MQTT does not have a message queue in the way you are thinking. MQTT is an extremely lightweight pub/sub protocol (a fixed header is only 2 bytes). The broker only forwards messages to those who are subscribed. If you want something like a queue, you must enable the correct flags yourself, and each flag has its own limitations. This article summarizes what I learned after losing data in production several times.

第一次做物联网项目时，我们的团队犯了一个非常普遍的错误：把 MQTT Broker 当作 RabbitMQ 来用。设备发布传感器数据，一个 Python Worker 订阅并将其写入数据库。一切运行正常，直到部署时 Worker 重启。仅仅 40 秒的停机时间，就导致了 200 多台设备的数据全部丢失。没有错误日志，没有死信队列，数据就这样消失了。原因很简单：MQTT 并没有你所认为的那种消息队列。MQTT 是一种极其轻量级的发布/订阅协议（固定报头仅 2 字节）。Broker 只会将消息转发给订阅者。如果你想要类似队列的功能，必须手动开启正确的标志位，且每个标志位都有其局限性。本文总结了我多次在生产环境丢失数据后学到的经验。

1. MQTT is “fire and forget” by default

In Kafka or RabbitMQ, messages remain in a queue/log until a consumer reads them. MQTT is different. If a client connects with clean_session=true (MQTT 3.1.1) or clean_start=true and a session expiry of 0 (MQTT 5), the broker will delete all its subscriptions when the client disconnects. Messages arriving during that time are not stored for anyone. There are three concepts often mistaken for a “queue”:

QoS (0, 1, 2): Only guarantees delivery over a single hop (publisher → broker, or broker → subscriber). QoS 2 is not exactly-once end-to-end.
Retained message: The broker keeps only the last message of each topic. It is a “last known value,” not a queue.
Persistent session: When a client is offline, the broker keeps the subscription and collects QoS 1/2 messages for that client. This is the closest thing to a queue.

1. MQTT 默认是“即发即弃” (fire and forget)

在 Kafka 或 RabbitMQ 中，消息会保留在队列/日志中，直到消费者读取为止。MQTT 则不同。如果客户端连接时设置了 clean_session=true（MQTT 3.1.1）或 clean_start=true 且会话过期时间为 0（MQTT 5），那么当客户端断开连接时，Broker 会删除其所有订阅。在此期间到达的消息不会为任何人保存。有三个概念常被误认为是“队列”：

QoS (0, 1, 2)： 仅保证单跳（发布者 → Broker，或 Broker → 订阅者）的交付。QoS 2 并非端到端的精确一次（exactly-once）。
保留消息 (Retained message)： Broker 只保留每个主题的最后一条消息。它是“最后已知值”，而不是队列。
持久会话 (Persistent session)： 当客户端离线时，Broker 会保留订阅并为该客户端收集 QoS 1/2 消息。这才是最接近队列的东西。

2. Enabling persistent sessions correctly

To test this quickly with Mosquitto 2.0.x: Since version 2.0, Mosquitto defaults to listening only on localhost and blocks anonymous access, so a minimal config is required:

# mosquitto.conf
listener 1883
allow_anonymous true
persistence true
persistence_location /tmp/mosquitto/
max_queued_messages 10000

Three conditions must be met simultaneously: a fixed client_id (-i), no clean session (-c), and both subscribe and publish must use QoS >= 1. The actual QoS received by the subscriber is min(QoS of publish, QoS of subscribe). If a device publishes at QoS 0, even if the subscriber requests QoS 1, it will not be queued.

2. 正确开启持久会话

使用 Mosquitto 2.0.x 进行快速测试：从 2.0 版本开始，Mosquitto 默认仅监听 localhost 并禁止匿名访问，因此需要最简配置：

# mosquitto.conf
listener 1883
allow_anonymous true
persistence true
persistence_location /tmp/mosquitto/
max_queued_messages 10000

必须同时满足三个条件：固定的 client_id (-i)，不使用 clean session (-c)，且订阅和发布都必须使用 QoS >= 1。订阅者实际收到的 QoS 是 min(发布 QoS, 订阅 QoS)。如果设备以 QoS 0 发布，即使订阅者请求 QoS 1，消息也不会进入队列。

3. Scaling workers with shared subscriptions

Persistent sessions solve the offline issue, but what if you run 3 workers to share the load? If all 3 subscribe to sensors/+/temp, each worker receives every message, meaning data is written 3 times. MQTT 5 has shared subscriptions to solve this. The syntax is $share/<group>/<topic>, and the broker distributes each message to exactly one member of the group. Note: each broker handles shared subscriptions differently when a member is offline. EMQX 5.x has strategies like round_robin, sticky, hash_clientid, etc.

3. 使用共享订阅扩展 Worker

持久会话解决了离线问题，但如果运行 3 个 Worker 来分担负载呢？如果这 3 个 Worker 都订阅 sensors/+/temp，每个 Worker 都会收到所有消息，这意味着数据会被写入 3 次。MQTT 5 引入了共享订阅来解决这个问题。语法为 $share/<group>/<topic>，Broker 会将每条消息分发给组内恰好一个成员。注意：当成员离线时，每个 Broker 处理共享订阅的方式不同。EMQX 5.x 提供了 round_robin（轮询）、sticky（粘性）、hash_clientid 等策略。