物联网设备有一个共同的需求：把数据实时传出去。温度传感器要上报温度，门磁要上报开关状态，烟雾报警器要在触发时第一时间通知手机。

这个需求听起来简单，但用 HTTP 做会很快碰到墙。MQTT 就是为了解决这个问题而生的。

一、为什么需要 MQTT

HTTP 的问题

HTTP 的模型是这样的：客户端发请求，服务器回响应。一问一答，客户端主动，服务器被动。

如果你想用 HTTP 实时监控一个温度传感器，只能轮询：

1
2
3
4
while (1) {
    http_get("http://sensor/temperature");
    sleep(1);
}

每秒发一个请求。大部分时候数据根本没变，这个请求是完全无效的。100 个设备同时轮询，就是 100 个并发连接一直占着服务器，带宽和服务器资源都在做无用功。

对嵌入式设备来说还有另一个问题：HTTP 本身比较重，每个请求都要带 Header，一个简单的温度数据可能只有几个字节，但 HTTP 的 Header 就有几百字节。电池供电的设备承受不起这个开销。

还有一个根本性的问题：HTTP 没有”推送”能力。服务器有新数据想通知客户端，做不到，只能等客户端下次来问。

发布/订阅模型

MQTT 引入了一个完全不同的模型：发布/订阅。

1
2
设备A  →  发布数据  →  Broker（中间人）
设备B  ←  推送数据  ←  Broker（中间人）

中间有一个 Broker，所有人都连到它。有数据的一方发布出去，想要数据的一方订阅，Broker 负责把数据从发布者推送到所有订阅者。

发布者不需要知道谁在订阅，订阅者不需要知道数据从哪来。两边完全解耦，通过 Broker 这个中间人沟通。

这个模型有几个好处：

推送而不是轮询。数据一产生就推出去，不需要订阅者反复来问，延迟低，带宽利用率高。

一对多。发布者发一条消息，所有订阅了这个主题的人都能收到，不需要发布者单独给每个人发一遍。

轻量。MQTT 的控制报文头最小只有 2 个字节，比 HTTP 精简得多，电池供电的设备也能用。

MQTT 诞生于 1999 年，最初是为油气管道监控设计的——卫星链路带宽极贵，每一个字节都要算。这个出身决定了它天生适合资源受限的场景。

二、核心概念

Broker

Broker 是 MQTT 架构里的中间人，负责接收所有发布者的消息，再转发给对应的订阅者。

用王者荣耀的匹配服务器来类比最直观。你开排位，不需要知道对手的 IP，你们都连到腾讯的匹配服务器，服务器把你们撮合在一起，转发双方的消息。没有这个服务器，两个玩家根本不知道对方在哪。

Broker 做的事情也一样：

  • 维护所有客户端的连接
  • 记录谁订阅了哪些主题
  • 收到消息后查表，把消息推给所有订阅者
  • 处理离线消息的存储和重传

Broker 本身不产生数据，不消费数据，只负责路由。

目前最常用的开源 Broker 是 Mosquitto，轻量、文档完善，嵌入式场景用得很多。企业级场景会用 EMQX，支持集群、高并发。

Topic

Topic 是消息的主题标签，是一个有层级结构的字符串，用 / 分隔，类似文件路径：

1
2
3
home/bedroom/temperature
home/kitchen/humidity
factory/line1/sensor/pressure

发布者把消息发到某个 Topic，订阅者告诉 Broker “我要订阅这个 Topic”，之后这个 Topic 上的所有消息都会被推送过来。

通配符

订阅时可以用通配符，发布时不行。

+ 匹配单层：

1
2
3
4
home/+/temperature
    匹配  home/bedroom/temperature
    匹配  home/kitchen/temperature
    不匹配 home/bedroom/sensor/temperature  （多了一层）

# 匹配任意多层，只能放在末尾：

1
2
3
4
5
home/#
    匹配  home/bedroom/temperature
    匹配  home/kitchen/humidity
    匹配  home/bedroom/sensor/temperature
    匹配  home/anything/a/b/c/d

实际使用中，# 常用来调试——订阅 # 可以收到这个 Broker 上的所有消息，方便排查问题。

QoS

QoS（Quality of Service）是消息可靠性等级，有三档。

QoS 0 — 最多发一次

发出去不管结果，网络丢包就丢了，不重传。开销最小，延迟最低。

适合传感器实时数据。丢一条温度数据无所谓，下一条很快就来。

QoS 1 — 至少发一次

发出去之后等 ACK 确认，没收到就重传。保证消息不丢，但可能重复收到同一条消息。

适合告警消息这类不能丢但能接受重复的场景。

QoS 2 — 恰好一次

四次握手，保证不丢不重复，开销最大。

适合支付、指令控制这类绝对不能重复执行的场景。嵌入式设备很少用到。

实际项目里，传感器数据用 QoS 0，告警和控制指令用 QoS 1，QoS 2 基本不碰。

Retain 消息

正常情况下，订阅者连上来之前发的消息就错过了。加了 Retain 标志，Broker 会把这个 Topic 最后一条消息存下来，新订阅者一连上就立刻收到，不需要等下一次发布。

1
2
3
4
5
6
# 发布一条 retain 消息
mosquitto_pub -t "home/bedroom/temperature" -m "25.1" --retain

# 1小时后，新客户端订阅这个 Topic
mosquitto_sub -t "home/bedroom/temperature"
# 立刻收到 25.1，不需要等传感器下一次上报

Retain 适合存设备状态、配置信息这类”最新值比历史值更重要”的数据。每个 Topic 只保留一条 retain 消息，新的会覆盖旧的。

Will 消息（遗嘱）

客户端连接时可以预先告诉 Broker：”如果我异常断线，帮我发这条消息”。

1
2
3
4
5
连接时设置：
  will_topic   = "home/device/status"
  will_payload = "offline"
  will_qos     = 1
  will_retain  = true

正常调用 DISCONNECT 断开不会触发，只有网络中断、程序崩溃这类异常情况才触发。

其他订阅者可以订阅 home/device/status，收到 offline 就知道设备掉线了，可以触发报警或者重连逻辑。Will 消息是 MQTT 里监控设备在线状态的标准做法。

三、连接与会话

连接参数

客户端连接 Broker 时需要提供几个参数。

ClientID

每个客户端的唯一标识字符串。同一个 Broker 上不能有两个客户端用同一个 ClientID，如果重复连接，后来的会把前面的踢掉。

ClientID 可以自己指定，也可以让 Broker 自动分配（发送空字符串，同时 CleanSession 必须为 true）。

用户名和密码

可选。Broker 配置了认证才需要提供，局域网测试通常不配，公网部署必须开。

Keep Alive

客户端告诉 Broker：”我最长 N 秒会发一次数据”。如果超过 1.5×N 秒没有收到任何报文，Broker 认为客户端已经掉线，触发 Will 消息。

客户端通过发送 PINGREQ 来维持心跳，Broker 回 PINGRESP：

1
2
客户端  ──PINGREQ──►  Broker
客户端  ◄──PINGRESP──  Broker

典型值是 60 秒，电池设备可以设更长来省电。

Clean Session

这是连接时的一个标志位，控制会话是否持久化，理解它很重要。

Clean Session = true

每次连接都是全新的，断线后 Broker 清除这个客户端的所有状态：订阅关系丢失，离线期间的 QoS 1/2 消息全部丢弃。重连后需要重新订阅。

适合发布者，或者对离线消息不感兴趣的场景。

Clean Session = false（持久会话）

Broker 用 ClientID 记住这个客户端的状态。断线后：

  • 订阅关系保留
  • QoS 1/2 的未送达消息排队等重连后补发
  • 重连后自动恢复，不需要重新订阅

适合设备频繁掉线但不能错过消息的场景，比如信号不稳定的传感器节点。代价是 Broker 要持久化存储会话，消耗资源，对于消息量大的场景要注意队列不能无限增长。

完整的连接流程

把上面这些放在一起，一次完整的 MQTT 会话是这样的：

CONNACK 的 returnCode 不为 0 说明连接被拒绝，常见原因是 ClientID 格式不对、用户名密码错误、Broker 不允许匿名连接等。

四、动手：在本机跑起来

理解概念之后，最快的学习方式是直接在本机跑起来，用命令行工具收发消息，不需要写一行代码。

安装 Mosquitto

Ubuntu / Debian：

1
2
3
sudo apt install mosquitto mosquitto-clients
sudo systemctl start mosquitto
sudo systemctl enable mosquitto   # 开机自启

默认监听 1883 端口，允许本地匿名连接。

验证是否跑起来：

1
sudo systemctl status mosquitto

命令行收发消息

Mosquitto 附带了两个命令行工具：mosquitto_sub（订阅）和 mosquitto_pub（发布）。

开一个终端，订阅 Topic：

1
mosquitto_sub -h localhost -t "test/hello" -v

-h 指定 Broker 地址，-t 指定 Topic，-v 打印 Topic 名称和消息内容。这个终端会一直阻塞，等待消息进来。

再开一个终端，发布消息：

1
mosquitto_pub -h localhost -t "test/hello" -m "你好，MQTT"

立刻能在第一个终端看到消息打印出来。

用通配符订阅多个 Topic：

1
mosquitto_sub -h localhost -t "home/#" -v

然后发布几条不同 Topic 的消息：

1
2
3
mosquitto_pub -h localhost -t "home/bedroom/temp" -m "25.1"
mosquitto_pub -h localhost -t "home/kitchen/humidity" -m "60"
mosquitto_pub -h localhost -t "home/living/light" -m "350"

三条消息都会出现在订阅窗口里。

测试 Retain：

1
2
3
4
5
6
# 发一条 retain 消息
mosquitto_pub -h localhost -t "home/status" -m "online" --retain

# 断开重连，订阅这个 Topic
mosquitto_sub -h localhost -t "home/status" -v
# 立刻收到 "online"，不需要等新消息

MQTTX 图形客户端

命令行够用，但如果想更直观地查看消息，可以装 MQTTX。

MQTTX 是一个跨平台的 MQTT 客户端工具，支持连接管理、消息收发、历史记录、图表展示。从 mqttx.app 下载对应平台的安装包。

连接配置很简单：新建连接，Host 填 localhost，Port 填 1883，其他默认，点 Connect。连上之后订阅 # 就能看到所有流经这个 Broker 的消息。

五、在 C 里用 libmosquitto

命令行工具验证通了之后，看怎么在 C 程序里用 MQTT。libmosquitto 是 Mosquitto 项目提供的 C 客户端库，API 简洁，嵌入式项目常用。

安装

1
sudo apt install libmosquitto-dev

编译时加 -lmosquitto。

核心 API

初始化和连接：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
#include <mosquitto.h>

mosquitto_lib_init();   // 全局初始化，程序启动时调一次

struct mosquitto *mosq = mosquitto_new(
    "my-client-id",  // ClientID
    true,            // CleanSession
    NULL             // 用户数据，回调里可以拿到
);

mosquitto_connect_callback_set(mosq, on_connect);
mosquitto_disconnect_callback_set(mosq, on_disconnect);
mosquitto_message_callback_set(mosq, on_message);

// 异步连接，不阻塞
mosquitto_connect_async(mosq, "localhost", 1883, 60);

// 启动后台线程处理网络 I/O，publish 线程安全
mosquitto_loop_start(mosq);

发布消息：

1
2
3
4
5
6
7
8
9
10
11
const char *payload = "{\"temp\": 25.1}";

mosquitto_publish(
    mosq,
    NULL,              // message id，不需要填 NULL
    "home/bedroom/temperature",
    strlen(payload),
    payload,
    0,                 // QoS
    false              // retain
);

订阅和接收：

1
2
3
4
5
6
7
8
9
10
11
12
static void on_connect(struct mosquitto *mosq, void *userdata, int rc)
{
    if (rc == 0)
        mosquitto_subscribe(mosq, NULL, "home/#", 1);
}

static void on_message(struct mosquitto *mosq, void *userdata,
                       const struct mosquitto_message *msg)
{
    printf("Topic: %s\n", msg->topic);
    printf("Payload: %.*s\n", msg->payloadlen, (char *)msg->payload);
}

退出清理：

1
2
3
4
mosquitto_disconnect(mosq);
mosquitto_loop_stop(mosq, true);
mosquitto_destroy(mosq);
mosquitto_lib_cleanup();

异步模式说明

mosquitto_loop_start 启动了一个内部后台线程专门处理网络 I/O。这意味着 mosquitto_publish 可以在任意线程安全调用，不需要自己加锁。但回调函数（on_connect、on_message）在这个后台线程里执行，如果回调里要操作共享数据，需要加锁。

如果不想用线程，可以用 mosquitto_loop_forever 阻塞模式，或者自己在主循环里定期调用 mosquitto_loop。

一个完整的发布者示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <mosquitto.h>

static void on_connect(struct mosquitto *mosq, void *ud, int rc)
{
    if (rc == 0)
        printf("连接成功\n");
    else
        fprintf(stderr, "连接失败，code=%d\n", rc);
}

int main(void)
{
    mosquitto_lib_init();

    struct mosquitto *mosq = mosquitto_new("publisher-demo", true, NULL);
    mosquitto_connect_callback_set(mosq, on_connect);
    mosquitto_connect_async(mosq, "localhost", 1883, 60);
    mosquitto_loop_start(mosq);

    sleep(1);   // 等连接建立

    for (int i = 0; i < 10; i++) {
        char buf[64];
        snprintf(buf, sizeof(buf), "{\"count\": %d}", i);
        mosquitto_publish(mosq, NULL, "demo/counter", strlen(buf), buf, 0, false);
        printf("发布: %s\n", buf);
        sleep(1);
    }

    mosquitto_disconnect(mosq);
    mosquitto_loop_stop(mosq, true);
    mosquitto_destroy(mosq);
    mosquitto_lib_cleanup();
    return 0;
}

编译：

1
gcc demo.c -lmosquitto -o publisher

跑之前先开一个终端订阅：

1
mosquitto_sub -h localhost -t "demo/counter" -v

六、Topic 设计建议

Topic 的设计看起来随意，实际上直接影响系统的可维护性和扩展性。

基本原则

从大到小，从粗到细。 最顶层是业务域或者设备类型，往下细化到具体指标：

1
2
3
4
5
{业务}/{设备类型}/{设备ID}/{数据类型}

home/sensor/bedroom-01/temperature
factory/motor/line1-m3/rpm
vehicle/truck/vin123456/location

这样设计的好处是通配符很自然：

1
2
home/sensor/#          订阅所有传感器数据
home/sensor/+/temperature    订阅所有传感器的温度

不要在 Topic 里放动态数据。 Topic 是路由键，不是 payload，不要把时间戳、序列号这类每条消息都不一样的东西放进去，那会让订阅变得不可能。

大小写统一，用小写加连字符。 home/bedroom/temperature 比 Home/BedRoom/Temperature 好维护，减少出错的机会。

不要以 / 开头。 home/temp 和 /home/temp 是两个不同的 Topic，前者有 2 层，后者有 3 层（第一层是空字符串）。统一不加开头的 /。

设备状态的标准模式

设备上线时发布：

1
2
3
Topic:   home/device/sensor-01/status
Payload: online
Retain:  true

设备的 Will 消息设置为：

1
2
3
Topic:   home/device/sensor-01/status
Payload: offline
Retain:  true

这样任何时候订阅 home/device/+/status，立刻能知道所有设备的当前状态，不需要等设备主动上报。

控制指令的模式

如果设备需要接收控制指令，通常用 /set 或 /cmd 后缀区分：

1
2
home/light/living/state        →  状态上报（设备发布）
home/light/living/set          →  控制指令（外部发布，设备订阅）

发布者和订阅者的角色对调，但 Topic 结构一致，一眼能看出关系。

七、安全：别裸连公网

本机测试用匿名连接没问题，但设备一旦上公网，裸连就是在把所有数据明文暴露出去，任何人都能订阅你的 Topic。

用户名密码

Mosquitto 配置文件 /etc/mosquitto/mosquitto.conf 加两行：

1
2
allow_anonymous false
password_file /etc/mosquitto/passwd

创建账号：

1
2
3
sudo mosquitto_passwd -c /etc/mosquitto/passwd your_username
# 输入密码
sudo systemctl restart mosquitto

客户端连接时带上认证：

1
mosquitto_pub -h localhost -u your_username -P your_password -t "test" -m "hello"

libmosquitto 里：

1
2
3
mosquitto_username_pw_set(mosq, "your_username", "your_password");
// 必须在 connect 之前调用
mosquitto_connect_async(mosq, "broker.example.com", 1883, 60);

TLS

用户名密码在明文连接下没有意义——密码本身也是明文传的，抓包一眼就看到。公网部署必须开 TLS，端口改用 8883。

Mosquitto 加 TLS 配置：

1
2
3
4
listener 8883
cafile   /etc/mosquitto/certs/ca.crt
certfile /etc/mosquitto/certs/server.crt
keyfile  /etc/mosquitto/certs/server.key

libmosquitto 客户端：

1
2
3
4
5
mosquitto_tls_set(mosq,
    "/path/to/ca.crt",   // CA 证书，验证服务端身份
    NULL, NULL, NULL, NULL
);
mosquitto_connect_async(mosq, "broker.example.com", 8883, 60);

如果只是内网设备，Mosquitto 加用户名密码就够了；设备要上公网，TLS 不能省。

ACL（访问控制）

用户名密码只控制”能不能连上”，ACL 控制”连上之后能读写哪些 Topic”。

1
2
3
4
5
6
# /etc/mosquitto/acl
user sensor_node
topic write home/sensor/#

user dashboard
topic read home/#

sensor_node 只能发布，dashboard 只能订阅，互相不干扰。生产环境建议都配上，最小权限原则。

八、常见坑

断线后不重连

mosquitto_loop_start 启动的后台线程只处理 I/O，不负责重连。网络抖动断线之后，程序不会自动恢复，需要自己处理 on_disconnect 回调：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
static void on_disconnect(struct mosquitto *mosq, void *ud, int rc)
{
    if (rc != 0) {
        // rc != 0 说明是异常断线，不是主动 DISCONNECT
        // 不要在回调里直接 reconnect，会在 loop 线程里死锁
        // 设一个标志位，让主线程重连
        g_need_reconnect = 1;
    }
}

// 主循环里
while (1) {
    if (g_need_reconnect) {
        g_need_reconnect = 0;
        mosquitto_reconnect(mosq);  // 用原来的参数重连
    }
    sleep(1);
}

或者直接用 mosquitto_reconnect_delay_set 让库自动处理：

1
2
// 断线后 1 秒重连，最长退避到 30 秒，加随机抖动
mosquitto_reconnect_delay_set(mosq, 1, 30, true);

加了这行，断线后库会自动重连，不需要自己写重连逻辑。

Clean Session 和 ClientID 乱配

持久会话（clean_session=false）必须配固定 ClientID，否则每次连接用不同 ClientID，Broker 会认为是全新客户端，持久会话完全没效果。

反过来，如果你用固定 ClientID 但 clean_session=true，每次连接都会清掉上次的会话，离线消息也丢了，和持久会话的目的背道而驰。

规则很简单：要持久会话，固定 ClientID + clean_session=false 必须同时满足。

订阅泄漏（持久会话下）

持久会话里，重连后不需要重新订阅，因为 Broker 记住了上次的订阅关系。但如果你在 on_connect 回调里无条件调 mosquitto_subscribe：

1
2
3
4
static void on_connect(struct mosquitto *mosq, void *ud, int rc)
{
    mosquitto_subscribe(mosq, NULL, "home/#", 1);  // 每次连接都订阅一次
}

每次重连都会给同一个 Topic 加一次订阅，虽然大多数 Broker 会去重，但这是不必要的行为，也掩盖了对持久会话语义的误解。持久会话下，只在首次连接时订阅一次就够了。CONNACK 报文里有一个 session_present 标志位，可以用它判断：

1
2
3
4
5
6
7
static void on_connect(struct mosquitto *mosq, void *ud, int rc)
{
    // rc==0 且 session_present==0，说明是新会话，需要重新订阅
    // 但 libmosquitto 的 on_connect 回调不直接暴露 session_present
    // 简单做法：只在 clean_session=true 时无条件订阅；
    //           clean_session=false 时首次连接订阅，重连跳过
}

实际项目里，最简单的做法是直接用 clean_session=true，每次连接重新订阅，逻辑清晰，不容易出错；只有明确需要离线消息时才用持久会话。

QoS 不匹配

发布者用 QoS 1 发，订阅者用 QoS 0 订阅——消息实际投递的 QoS 取两边的最小值，也就是 QoS 0，消息可能丢。这不是 bug，是 MQTT 规范的设计，但容易踩。

如果你要保证告警消息不丢，发布和订阅都要用 QoS 1，缺一不可。

Broker 上线前 publish 会丢

如果程序启动后立刻调 mosquitto_publish，而此时连接还没建立，消息会直接丢掉，不会排队等连接成功再发。

mosquitto_connect_async 是非阻塞的，连接在后台建立，调用后不代表已经连上。正确做法是等 on_connect 回调触发再开始发布，或者在发布前判断连接状态。

九、MQTT vs 其他协议

vs HTTP

HTTP MQTT
模型 请求/响应 发布/订阅
推送 不支持（需轮询或 WebSocket） 原生支持
报文头开销 几百字节 最小 2 字节
连接 短连接为主 长连接
适合 Web API、一次性请求 实时数据、IoT 设备

HTTP 适合”我要查一个资源”这类场景，MQTT 适合”有数据了就推给我”这类场景。两者不是替代关系，很多系统会同时用：设备用 MQTT 上报数据，后台用 HTTP 提供 API 给前端查询。

vs WebSocket

WebSocket 是全双工的持久连接，解决了 HTTP 的推送问题，但它只是一个传输层，本身没有消息路由、QoS、持久会话这些机制，需要自己实现。

MQTT 可以跑在 WebSocket 之上（MQTT over WebSocket，端口 8083/8084），这是微信小程序、浏览器端接入 MQTT 的标准方式，因为浏览器不能直接建 TCP 连接。

vs Modbus

Modbus 是工业领域的老协议（1979 年），用在 PLC、变频器、温控仪这类工业设备之间通信，物理层通常是 RS-485 串口。主从模型，一问一答。

两者完全不是同一个场景：Modbus 是工厂车间里设备互联用的，MQTT 是 IoT 设备往云端或手机推数据用的。如果你要对接一台带 Modbus 接口的工业仪表，读出数据之后再通过 MQTT 上报，这两个协议可以同时出现在一个系统里，各司其职。