物联网协议 MQTT 快速开发与实践

MQTT (消息队列遥测传输，Message Queuing Telemetry Transport) 是OASIS和ISO制定的物联网消息协议，主要有3.1.1和5.0版规范，新项目普遍使用 MQTT 5.0。

• MQTT5 特点
  • 基于TCP
  • 发布/订阅模式，核心是 MQTT 客户端(Client) 和 MQTT代理(Broker)
  • 支持服务质量 (QoS) 等级、持久会话、会话管理、用户属性等

二、数据包格式

• Packet type：15 种控制报文类型，可分为连接、发布、订阅三类
  • ACK：确认消息
  • REC、REL、COMP：保证交付，收到、释放、完成

• Flag: 标志位，PUBLISH报文标识

  • Bit 3：DUP，是否重传报文。
  • Bit 2,1：QoS，服务质量等级。
  • Bit 0：Retain，是否保留消息。

• Remaining Length：从第2个字节开始，可变头（Variable header） + 消息体（payload）的长度

三、发布/订阅

• 主题Topic：消息路由通道，类似于 URL 路径，使用/分层，订阅时可使用+或#通配符

sensor/1/temperature
sensor/2/temperature

sensor/+/temperature   //单层通配符
sensor/#               //多层通配符

• QoS
  • QoS 0：消息最多传送一次。如果当前客户端不可用，它将丢失这条消息。
  • QoS 1：消息至少传送一次。
  • QoS 2：消息只传送一次。

四、开发实践

代码示例使用Mosquitto，当然Client也可以选Paho、Broker选EMQX、NanoMQ、HiveMQ等。

Broker （代理）

1. 下载 mosquitto broker安装，安装时勾选"Service"组件
2. 安装目录下打开mosquitto.conf，添加监听配置

listener 1883 0.0.0.0
allow_anonymous true

3. 打开Windows服务，重启服务
4. 安装目录下，切到命令行，运行测试

mosquitto_sub -h localhost -t <span>"test/topic"</span> -v
<span># 运行下一句pub后收到： test/topic Hello MQTT</span>

mosquitto_pub -h localhost -t <span>"test/topic"</span> -m <span>"Hello MQTT3"</span>

Client （发布者/订阅者）

1. cmake编译mosquitto及所需依赖如cJson等,可选的如需要持久化添加SQLite依赖
2. 创建C++项目，设置头文件包含目录和库目录，设置附加依赖项mosquitto.lib libcjson.lib...
3. Client代码编译成功后，拷贝dll到exe目录，先运行订阅者，再运行发布者，否则消息后丢失

Publisher 发布者

<span>#<span>include</span> <span><iostream></span></span>
<span>#<span>include</span> <span>"mosquitto.h"</span></span>

<span><span>void</span> <span>on_connect</span><span>(<span>struct</span> mosquitto* mosq, <span>void</span>* obj, <span>int</span> reason_code)</span>
</span>{
	<span>printf</span>(<span>"on_connect: %s\n"</span>, <span>mosquitto_connack_string</span>(reason_code));
	<span>if</span> (reason_code != <span>0</span>)
		<span>mosquitto_disconnect</span>(mosq);
}

<span><span>void</span> <span>on_publish</span><span>(<span>struct</span> mosquitto* mosq, <span>void</span>* obj, <span>int</span> mid)</span>
</span>{
    <span>printf</span>(<span>"Message with mid %d has been published.\n"</span>, mid);
}

<span><span>int</span> <span>main</span><span>()</span>
</span>{
    <span>mosquitto_lib_init</span>();
    <span>struct</span> <span>mosquitto</span>* mosq = <span>mosquitto_new</span>(<span>NULL</span>, <span>true</span>, <span>NULL</span>);

    <span>// 设置回调函数</span>
    <span>mosquitto_connect_callback_set</span>(mosq, on_connect); <span>// 收到CONNACK触发</span>
    <span>mosquitto_publish_callback_set</span>(mosq, on_publish); <span>// PUBLISH成功触发(QoS0) (QoS1:PUBACK , QoS2:PUBCOMP)</span>

    <span>mosquitto_connect</span>(mosq, <span>"127.0.0.1"</span>, <span>1883</span>, <span>60</span>);
    <span>mosquitto_loop_start</span>(mosq);

    <span>//推送消息</span>
    <span>char</span> payload[<span>20</span>] = {<span>0</span>};
    <span>snprintf</span>(payload, <span>sizeof</span>(payload), <span>"temperature is %d"</span>, <span>26</span>);
    <span>mosquitto_publish</span>(mosq, <span>NULL</span>, <span>"example/temperature"</span>, (<span>int</span>)<span>strlen</span>(payload), payload, <span>2</span>, <span>false</span>);

    <span>getchar</span>();

    <span>mosquitto_lib_cleanup</span>();

    <span>return</span> <span>0</span>;
}

Subscriber 订阅者

<span>#<span>include</span> <span><iostream></span></span>
<span>#<span>include</span> <span>"mosquitto.h"</span></span>

<span><span>void</span> <span>on_connect</span><span>(<span>struct</span> mosquitto* mosq, <span>void</span>* obj, <span>int</span> reason_code)</span>
</span>{
    <span>printf</span>(<span>"on_connect: %s\n"</span>, <span>mosquitto_connack_string</span>(reason_code));
    <span>if</span> (reason_code != <span>0</span>)
        <span>mosquitto_disconnect</span>(mosq);

    <span>//订阅，写在此处重连时能恢复</span>
    <span>mosquitto_subscribe</span>(mosq, <span>NULL</span>, <span>"example/temperature"</span>, <span>1</span>);
}

<span>// obj:mosquitto_new时的用户数据；mid:mosquitto_subscribe时的消息id；</span>
<span>// qos_count:Topic主题数量；granted_qos：每个订阅请求最终授予的 QoS</span>
<span><span>void</span> <span>on_subscribe</span><span>(<span>struct</span> mosquitto* mosq, <span>void</span>* obj, <span>int</span> mid, <span>int</span> qos_count, <span>const</span> <span>int</span>* granted_qos)</span>
</span>{
    <span>bool</span> have_any_subscription = <span>false</span>;
    
    <span>for</span> (<span>int</span> i = <span>0</span>; i < qos_count; i++) {
        <span>printf</span>(<span>"on_subscribe: %d:granted qos = %d\n"</span>, i, granted_qos[i]);
        <span>if</span> (granted_qos[i] <= <span>2</span>) {
			have_any_subscription = <span>true</span>;<span>//qos 0,1,2表示订阅成功，128表示订阅失败</span>
        }
    }
    <span>if</span> (have_any_subscription == <span>false</span>) {
        <span>fprintf</span>(stderr, <span>"Error: All subscriptions rejected.\n"</span>);
        <span>mosquitto_disconnect</span>(mosq);
    }
}

<span><span>void</span> <span>on_message</span><span>(<span>struct</span> mosquitto* mosq, <span>void</span>* obj, <span>const</span> <span>struct</span> mosquitto_message* msg)</span>
</span>{
    <span>printf</span>(<span>"%s %d %s\n"</span>, msg->topic, msg->qos, (<span>char</span>*)msg->payload);
}

<span><span>int</span> <span>main</span><span>()</span>
</span>{
    <span>mosquitto_lib_init</span>();

    <span>struct</span> <span>mosquitto</span>* mosq = <span>mosquitto_new</span>(<span>NULL</span>, <span>true</span>, <span>NULL</span>);

    <span>// 设置回调函数</span>
    <span>mosquitto_connect_callback_set</span>(mosq, on_connect);     <span>// 收到CONNACK触发</span>
    <span>mosquitto_subscribe_callback_set</span>(mosq, on_subscribe); <span>// 收到SUBACK触发，即订阅成功</span>
    <span>mosquitto_message_callback_set</span>(mosq, on_message);     <span>// 收到消息触发</span>

    <span>mosquitto_connect</span>(mosq, <span>"127.0.0.1"</span>, <span>1883</span>, <span>60</span>);

    <span>mosquitto_loop_start</span>(mosq);

    <span>getchar</span>();

    <span>mosquitto_lib_cleanup</span>();

    <span>return</span> <span>0</span>;
}

五、资料参考

• HiveMQ - MQTT5
• EMQX - # MQTT 协议快速入门 2025：基础知识和实用教程