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

Lần đầu làm IoT, team mình mắc một lỗi rất phổ biến: coi MQTT broker như RabbitMQ. Thiết bị publish dữ liệu cảm biến, một worker Python subscribe rồi ghi vào database.

Mọi thứ chạy ổn cho đến hôm worker bị restart lúc deploy. Đúng 40 giây downtime mà mất sạch dữ liệu của hơn 200 thiết bị. Không có error log, không có dead letter, dữ liệu cứ thế biến mất.

Lý do rất đơn giản: MQTT không có message queue theo nghĩa bạn đang nghĩ. MQTT là giao thức pub/sub cực nhẹ (fixed header chỉ 2 byte). Broker chỉ chuyển message tới những ai đang subscribe. Muốn có thứ gì đó giống queue thì bạn phải tự bật đúng cờ, và từng cờ đều có giới hạn riêng.

Bài này tổng hợp những gì mình học được sau vài lần bị mất dữ liệu trên production.

1. Mặc định MQTT là "fire and forget"

Trong Kafka hay RabbitMQ, message nằm trong queue/log cho tới khi có consumer đọc. MQTT thì khác. Nếu client kết nối với clean_session=true (MQTT 3.1.1) hoặc clean_start=true và session expiry bằng 0 (MQTT 5), thì lúc client ngắt kết nối broker sẽ xoá hết subscription của nó. Message đến trong lúc đó không được lưu cho ai cả.

Có ba khái niệm hay bị nhầm thành "queue":

• QoS (0, 1, 2): chỉ đảm bảo việc giao nhận trên một chặng (publisher → broker, hoặc broker → subscriber). QoS 2 không phải exactly-once end-to-end.
• Retained message: broker giữ duy nhất message cuối cùng của mỗi topic. Nó là "last known value", không phải hàng đợi.
• Persistent session: lúc client offline, broker giữ subscription và gom các message QoS 1/2 cho client đó. Đây mới là thứ gần với queue nhất.

sequenceDiagram
participant D as Device
participant B as Broker
participant W as Worker
W->>B: CONNECT clean_start=false, expiry=3600
W->>B: SUBSCRIBE sensors/+/temp QoS1
W--xB: mất kết nối (deploy)
D->>B: PUBLISH QoS1 temp=31.2
B->>B: lưu vào session của worker
W->>B: CONNECT lại cùng client_id
B->>W: CONNACK session_present=1
B->>W: PUBLISH temp=31.2 (gửi bù)
W->>B: PUBACK

2. Bật persistent session cho đúng cách

Thử nhanh với Mosquitto 2.0.x. Từ bản 2.0, Mosquitto mặc định chỉ listen trên localhost và chặn anonymous, nên phải có config tối thiểu:

# mosquitto.conf
cat > mosquitto.conf << 'EOF'
listener 1883
allow_anonymous true
persistence true
persistence_location /tmp/mosquitto/
max_queued_messages 10000
EOF
mkdir -p /tmp/mosquitto/ && mosquitto -c mosquitto.conf -v

# Terminal 2: subscriber có session bền, expiry 1 giờ
mosquitto_sub -V mqttv5 -i worker-01 -c -x 3600 -q 1 -t 'sensors/+/temp'
# Ctrl+C để giả lập worker chết, sau đó publish:
mosquitto_pub -V mqttv5 -q 1 -t sensors/dev42/temp -m '31.2'
# Chạy lại lệnh mosquitto_sub y như cũ -> message được gửi bù

Có ba điều kiện phải thoả cùng lúc: client_id cố định (-i), không dùng clean session (-c), và cả subscribe lẫn publish đều QoS >= 1. QoS thực tế mà subscriber nhận được là min(QoS của publish, QoS của subscribe). Thiết bị publish QoS 0 thì dù subscriber có xin QoS 1 cũng không được queue (Mosquitto có tuỳ chọn queue_qos0_messages nhưng mình không khuyên dùng).

Trong Python, dùng paho-mqtt 2.1 (bản 2.x bắt buộc khai báo CallbackAPIVersion):

import json
import paho.mqtt.client as mqtt
from paho.mqtt.properties import Properties
from paho.mqtt.packettypes import PacketTypes

seen = set() # demo; production nên dùng Redis SETNX hoặc unique key trong DB

def on_connect(client, userdata, flags, reason_code, properties):
    print(f"connected rc={reason_code} session_present={flags.session_present}")
    # Luôn subscribe lại: nếu session đã hết hạn, broker không còn nhớ gì
    client.subscribe("sensors/+/temp", qos=1)

def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    key = (msg.topic, data["seq"])
    if key in seen: # QoS 1 = at-least-once -> sẽ có duplicate
        return
    seen.add(key)
    save_to_db(msg.topic, data)

client = mqtt.Client(
    mqtt.CallbackAPIVersion.VERSION2,
    client_id="worker-01",
    protocol=mqtt.MQTTv5,
)
client.on_connect = on_connect
client.on_message = on_message

props = Properties(PacketTypes.CONNECT)
props.SessionExpiryInterval = 3600 # giữ session 1 giờ sau khi disconnect
client.connect("localhost", 1883, keepalive=30, clean_start=False, properties=props)
client.loop_forever()

Hai bẫy mình đã dính:

• Hai instance dùng chung client_id: broker sẽ kick instance cũ, instance cũ reconnect thì kick lại instance mới, tạo thành vòng lặp disconnect liên tục. Log Mosquitto sẽ hiện Client worker-01 already connected, closing old connection. Gặp dòng này thì kiểm tra ngay số replica.
• max_queued_messages bị tràn: mặc định là 1000. Nếu worker chết lâu và thiết bị gửi nhiều, message mới vượt ngưỡng sẽ bị drop một cách âm thầm. Hãy tính số message/giây × thời gian downtime tối đa rồi cấu hình cho đủ, kèm max_queued_bytes để broker không ăn hết RAM.

3. Scale worker bằng shared subscription

Persistent session giải quyết chuyện offline, nhưng chạy 3 worker để chia tải thì sao? Nếu cả 3 cùng subscribe sensors/+/temp, mỗi worker đều nhận đủ mọi message, tức là dữ liệu bị ghi 3 lần. MQTT 5 có shared subscription để giải bài toán này. Cú pháp là $share/<group>/<topic>, và broker sẽ phân phối mỗi message cho đúng một thành viên trong group.

graph LR
D1[Device A] --> B((Broker))
D2[Device B] --> B
B -->|$share/ingest| W1[Worker 1]
B -->|$share/ingest| W2[Worker 2]
B -->|$share/ingest| W3[Worker 3]
B -->|sensors/#| M[Dashboard: nhận tất cả]

Ví dụ với mqtt.js 5.x trên Node.js:

import mqtt from 'mqtt';
const id = `ingest-${process.env.HOSTNAME}`; // mỗi pod một id riêng, ổn định
const client = mqtt.connect('mqtt://localhost:1883', {
  protocolVersion: 5,
  clientId: id,
  clean: false,
  properties: { sessionExpiryInterval: 600 },
});
client.on('connect', (connack) => {
  console.log('session present:', connack.sessionPresent);
  client.subscribe('$share/ingest/sensors/+/temp', { qos: 1 });
});
client.on('message', async (topic, payload) => {
  await saveToDb(topic, JSON.parse(payload.toString()));
});

Lưu ý: mỗi broker xử lý shared subscription khi có member offline theo cách khác nhau. EMQX 5.x có các strategy như round_robin, sticky, hash_clientid và có thể dispatch lại cho member khác. Mosquitto thì đơn giản hơn nhiều. Đừng giả định gì cả, hãy test bằng cách kill một worker giữa chừng rồi đếm số message nhận được.

4. Khi nào nên thôi ép MQTT làm queue

MQTT rất mạnh ở chặng "thiết bị ↔ cloud": kết nối chập chờn, băng thông thấp, hàng chục nghìn client. Nhưng nếu bạn cần replay dữ liệu từ 3 ngày trước, cần retention theo dung lượng, hoặc cần nhiều consumer group đọc lại độc lập, thì đó là việc của Kafka, Redpanda hay NATS JetStream.

Pattern mình dùng ổn định nhất:

• Thiết bị → MQTT broker (QoS 1).
• Một bridge nhỏ (hoặc rule engine có sẵn của EMQX) chuyển ngay sang Kafka/JetStream.
• Các service backend đọc từ Kafka, không subscribe trực tiếp MQTT.

Làm vậy thì persistent session chỉ cần đủ cho bridge, tức là một client duy nhất, dễ theo dõi, dễ tính max_queued_messages.

Những việc nên làm ngay nếu hệ thống của bạn đang chạy MQTT:

• Kiểm tra QoS ở cả hai đầu. Publisher dùng QoS 0 thì mọi cấu hình session phía subscriber đều vô nghĩa.
• Dùng client_id cố định, clean_start=false và đặt SessionExpiryInterval rõ ràng cho mọi consumer quan trọng.
• Luôn subscribe lại trong on_connect.
• Thiết kế idempotent. QoS 1 là at-least-once, vì vậy hãy gắn seq hoặc message ID trong payload và dedupe ở tầng ghi DB.
• Tính và cấu hình max_queued_messages / max_queued_bytes theo thời gian downtime tối đa bạn chấp nhận được, đừng để giá trị mặc định 1000.
• Scale bằng $share/group/topic, đừng nhân bản subscriber thường. Và đừng bao giờ để hai instance dùng chung client_id.
• Cần retention hay replay thì bridge sang Kafka/JetStream. MQTT là đường ống vận chuyển, không phải nơi lưu trữ.

Hãy nhớ: retained message là "giá trị cuối", persistent session là "hộp thư tạm", còn queue thật sự thì MQTT chưa bao giờ hứa sẽ cung cấp. Hiểu đúng điều này là đã tránh được phần lớn các vụ mất dữ liệu "không rõ nguyên nhân" trong hệ thống IoT.