MQTT¶

MQTT là giao thức publish/subscribe chạy trên TCP, được thiết kế cho thiết bị tài nguyên thấp và mạng không ổn định. Đây là giao thức phổ biến nhất giữa gateway/thiết bị và cloud IoT. Hai phiên bản đang dùng là 3.1.1 (chuẩn OASIS năm 2014) và 5.0 (chuẩn OASIS năm 2019).

Mô hình publish/subscribe¶

flowchart LR
    S1["Sensor A"] -- "PUBLISH site1/room2/temp" --> B(("Broker"))
    S2["Sensor B"] -- "PUBLISH site1/room3/temp" --> B
    B -- "khớp site1/+/temp" --> D["Dashboard"]
    B -- "khớp site1/#" --> L["Logger"]

Client không gửi trực tiếp cho nhau. Mọi bản tin đi qua broker, broker chuyển tiếp cho các client đã subscribe topic khớp.
Publisher và subscriber không cần biết nhau, không cần online cùng lúc (nếu dùng persistent session hoặc retained message).
Port mặc định: 1883 (TCP), 8883 (TLS). MQTT over WebSocket thường dùng 8083/443.

Topic¶

Chuỗi UTF-8 phân cấp bằng /, ví dụ site1/gw-07/sensor/temp. Không có thao tác tạo topic: publish là topic tồn tại.
Wildcard chỉ dùng khi subscribe:
- +: đúng một cấp. site1/+/temp khớp site1/room2/temp, không khớp site1/room2/a/temp.
- #: mọi cấp còn lại, phải đứng cuối. site1/# khớp mọi thứ dưới site1, kể cả chính site1.
Topic bắt đầu bằng $ dành cho broker (ví dụ $SYS/ chứa thống kê của Mosquitto) và không khớp với wildcard # ở cấp đầu.
Quy ước nên theo: không bắt đầu bằng /, không dùng khoảng trắng, đặt định danh thiết bị trong topic để dễ phân quyền theo ACL (dev/<device-id>/...).

Cấu trúc gói tin¶

Mỗi gói tin gồm fixed header, variable header và payload:

| Byte 1: type (4 bit) | flags (4 bit) |  Remaining Length (1–4 byte, varint) | Variable header | Payload |

Gói nhỏ nhất (PINGREQ, PINGRESP) chỉ có 2 byte. Overhead thấp là lý do MQTT hợp với mạng cellular tính tiền theo dung lượng.
Các loại gói chính: CONNECT/CONNACK, PUBLISH/PUBACK/PUBREC/PUBREL/PUBCOMP, SUBSCRIBE/SUBACK, UNSUBSCRIBE/UNSUBACK, PINGREQ/PINGRESP, DISCONNECT, và AUTH (chỉ có từ MQTT 5).
Payload là dãy byte tùy ý, broker không đọc. Định dạng (JSON, CBOR, Protobuf) do hai đầu tự thỏa thuận.

QoS¶

QoS áp dụng cho từng chặng: publisher → broker và broker → subscriber. QoS thực tế mà subscriber nhận được là mức thấp hơn giữa QoS lúc publish và QoS lúc subscribe.

QoS	Đảm bảo	Trao đổi	Dùng khi
0	Nhiều nhất một lần (có thể mất)	`PUBLISH`	Telemetry định kỳ, mất một mẫu không sao
1	Ít nhất một lần (có thể trùng)	`PUBLISH` → `PUBACK`	Phần lớn trường hợp; bên nhận tự loại trùng nếu cần
2	Đúng một lần	`PUBLISH` → `PUBREC` → `PUBREL` → `PUBCOMP`	Lệnh không được lặp (trừ tiền, đếm sự kiện); tốn 4 gói và trạng thái

sequenceDiagram
    participant C as Client
    participant B as Broker
    Note over C,B: QoS 1
    C->>B: PUBLISH (id=10, QoS 1)
    B-->>C: PUBACK (id=10)
    Note over C,B: QoS 2
    C->>B: PUBLISH (id=11, QoS 2)
    B-->>C: PUBREC (id=11)
    C->>B: PUBREL (id=11)
    B-->>C: PUBCOMP (id=11)

Session, retained message và Last Will¶

Persistent session: khi client kết nối với Clean Session = 0 (3.1.1), hoặc Clean Start = 0 kèm Session Expiry Interval > 0 (5.0), broker giữ danh sách subscription và các bản tin QoS 1/2 gửi tới trong lúc client offline. Client phải dùng Client ID cố định.
Retained message: publish với cờ retain, broker lưu bản tin cuối cùng của topic và gửi ngay cho subscriber mới. Hợp cho trạng thái ("online", cấu hình hiện tại). Xóa bằng cách publish payload rỗng có cờ retain.
Last Will and Testament (LWT): client khai báo trong CONNECT một bản tin "di chúc". Nếu client mất kết nối mà không gửi DISCONNECT (mất điện, mất sóng, keep alive hết hạn), broker publish bản tin đó.
Keep alive: client cam kết gửi ít nhất một gói trong mỗi khoảng keep alive (gửi PINGREQ nếu không có gì). Broker ngắt kết nối nếu quá 1,5 lần khoảng đó mà không nhận được gì.

Mẫu hay dùng để báo trạng thái thiết bị: LWT dev/<id>/status = offline (retain), ngay sau khi kết nối thì publish dev/<id>/status = online (retain).

MQTT 5 có gì mới¶

Tính năng	Ý nghĩa
Reason code trong mọi ACK	Biết vì sao bị từ chối (sai quyền, quota, topic không hợp lệ)
User properties	Cặp key/value tùy ý gắn vào bản tin, giống HTTP header
Session / Message Expiry Interval	Session và bản tin tự hết hạn, tránh broker giữ dữ liệu cũ mãi mãi
Shared subscription `$share/<group>/<topic>`	Chia tải bản tin cho nhiều consumer trong một nhóm
Topic alias	Thay topic dài bằng số nguyên, giảm byte trên mạng hẹp
Request/response (Response Topic, Correlation Data)	Chuẩn hóa mẫu gửi lệnh và chờ trả lời
Flow control (Receive Maximum)	Giới hạn số bản tin QoS 1/2 đang chờ xác nhận
Will Delay Interval	Trì hoãn LWT, tránh báo offline khi client chỉ kết nối lại nhanh
Enhanced authentication (`AUTH`)	Xác thực nhiều bước (SCRAM, Kerberos)

Thiết bị mới nên dùng MQTT 5 nếu broker và cloud hỗ trợ. Một số dịch vụ cloud chỉ hỗ trợ một phần tính năng của MQTT 5, cần kiểm tra trước.

Broker¶

Broker	Ngôn ngữ	Đặc điểm	Phù hợp
Eclipse Mosquitto	C	Nhẹ, một luồng, có sẵn trong Yocto (`meta-oe`) và Buildroot	Gateway, hệ thống vừa và nhỏ
NanoMQ	C	Đa luồng, thiết kế cho edge	Gateway cần thông lượng cao
EMQX	Erlang	Cluster, rule engine, hàng triệu kết nối	Cloud, self-host quy mô lớn
HiveMQ	Java	Cluster, bản thương mại	Doanh nghiệp
VerneMQ	Erlang	Cluster, mã nguồn mở	Self-host quy mô lớn

Cấu hình Mosquitto¶

Từ Mosquitto 2.0, nếu không khai báo listener thì broker chỉ nghe trên localhost, và mặc định không cho kết nối ẩn danh.

/etc/mosquitto/conf.d/gateway.conf

# Mỗi listener có cấu hình xác thực riêng
per_listener_settings true

persistence true
persistence_location /var/lib/mosquitto/

# Cổng nội bộ cho các service trên gateway
listener 1883 127.0.0.1
allow_anonymous true

# Cổng TLS cho thiết bị, xác thực bằng client certificate
listener 8883
allow_anonymous false
cafile   /etc/mosquitto/certs/ca.crt
certfile /etc/mosquitto/certs/server.crt
keyfile  /etc/mosquitto/certs/server.key
require_certificate true
use_identity_as_username true
# Mỗi thiết bị chỉ được ghi vào cây topic của chính nó
acl_file /etc/mosquitto/acl

/etc/mosquitto/acl

# %u là username, ở đây là CN của client certificate
pattern write dev/%u/#
pattern read  cmd/%u/#

Lưu ý: nếu không có per_listener_settings true, các tùy chọn xác thực (allow_anonymous, acl_file, password_file) là toàn cục, áp dụng cho mọi listener, nên cổng nội bộ ẩn danh sẽ mở luôn cổng TLS. Bảo vệ kênh TLS xem thêm TLS.

Client¶

Dòng lệnh¶

# Theo dõi mọi topic (debug)
mosquitto_sub -h localhost -t '#' -v

# Gửi một bản tin QoS 1, có retain
mosquitto_pub -h localhost -t dev/gw-07/status -m online -q 1 -r

# Kết nối TLS với client certificate
mosquitto_sub -h broker.example.com -p 8883 \
    --cafile ca.crt --cert gw-07.crt --key gw-07.key \
    -t 'cmd/gw-07/#' -V mqttv5

C với libmosquitto¶

#include <mosquitto.h>
#include <stdio.h>
#include <string.h>

int main(void)
{
    const char *id = "gw-07";
    char payload[64];
    struct mosquitto *m;

    mosquitto_lib_init();
    m = mosquitto_new(id, false, NULL);       /* clean_session = false: giữ session */

    /* LWT phải đặt trước khi connect */
    mosquitto_will_set(m, "dev/gw-07/status", 7, "offline", 1, true);

    if (mosquitto_connect(m, "localhost", 1883, 60) != MOSQ_ERR_SUCCESS) {
        fprintf(stderr, "connect failed\n");
        return 1;
    }
    mosquitto_loop_start(m);                   /* thread riêng lo I/O và keep alive */

    mosquitto_publish(m, NULL, "dev/gw-07/status", 6, "online", 1, true);

    snprintf(payload, sizeof(payload), "{\"temp\":%.1f}", 27.5);
    mosquitto_publish(m, NULL, "dev/gw-07/sensor/temp",
                      strlen(payload), payload, 1, false);

    /* Ví dụ tối giản: thực tế nên chờ callback on_publish trước khi ngắt kết nối */
    mosquitto_disconnect(m);
    mosquitto_loop_stop(m, false);
    mosquitto_destroy(m);
    mosquitto_lib_cleanup();
    return 0;
}

gcc mqtt_pub.c -o mqtt_pub -lmosquitto

Thư viện khác thường gặp: Eclipse Paho (C, C++, Python, Java), coreMQTT (FreeRTOS, cho MCU), client MQTT có sẵn trong Zephyr và ESP-IDF.

Lưu ý khi triển khai¶

Client ID duy nhất: hai client cùng ID sẽ đá nhau liên tục, vì broker ngắt kết nối cũ khi có kết nối mới cùng ID.
Kết nối lại có backoff: hàng nghìn thiết bị cùng kết nối lại ngay khi broker khởi động lại sẽ gây quá tải. Dùng exponential backoff có jitter.
Đệm khi mất mạng: QoS 1 chỉ đảm bảo trong lúc kết nối còn sống hoặc session còn. Gateway nên lưu dữ liệu xuống đĩa (store-and-forward) nếu không được phép mất. Xem Edge Gateway.
Kích thước payload: broker và cloud thường giới hạn kích thước bản tin (Mosquitto: message_size_limit). Không gửi firmware qua MQTT; chỉ gửi URL và hash, thiết bị tự tải qua HTTPS.
Keep alive và NAT: NAT của nhà mạng có thể xóa ánh xạ TCP sau vài phút không có dữ liệu. Chọn keep alive ngắn hơn khoảng thời gian đó, cân đối với năng lượng tiêu thụ.
Không dùng # trong sản phẩm: subscribe # trên broker lớn sẽ kéo toàn bộ lưu lượng về một client.

Liên quan¶

Giao thức thay thế trên UDP cho thiết bị hạn chế: CoAP & LwM2M.
Bridge dữ liệu từ BLE, Zigbee, Modbus sang MQTT: Edge Gateway.