2. MQTT协议基础与RT-Thread MQTT组件使用

任务目标

在RT-Thread系统中启用MQTT组件，配置并连接公共MQTT Broker（如EMQX的公共服务器或broker.emqx.io），完成主题订阅与消息发布，验证MQTT通信链路可用。

MQTT协议要点

MQTT基于发布/订阅模型，包含三个核心角色：Broker（服务器）、Publisher（发布者）、Subscriber（订阅者）。关键概念：

• 主题（Topic）：消息路由标识，支持层级分隔符/和通配符+（单层）、#（多层）
• QoS（服务质量）：0最多一次、1至少一次、2恰好一次
• 会话（Session）：Clean Session标志决定是否持久化会话状态
• 遗嘱消息（LWT）：客户端异常断开时Broker代发消息

环境准备

本案例基于RT-Thread Studio或ENV工具，目标平台为STM32系列开发板（如STM32F407），需具备以太网或WiFi模块（如ESP8266 AT方式）。核心依赖：

• RT-Thread 5.0+（含AT组件或LWIP网络协议栈）
• MQTT组件包：paho_mqtt（RT-Thread官方维护）

启用MQTT组件

使用ENV工具在BSP目录执行：

menuconfig

按以下路径启用组件：

RT-Thread online packages → IoT - internet of things → Paho MQTT
    → Enable MQTT client
    → Enable MQTT example（可选，用于参考）

保存后执行pkgs --update拉取源码，然后重新编译烧录。

关键完整代码

以下代码实现MQTT初始化、连接公共Broker、订阅主题并发布消息。将代码保存为mqtt_demo.c，放入应用目录：

#include <rtthread.h>
#include <stdio.h>
#include "mqtt_client.h"

#define MQTT_URI       "tcp://broker.emqx.io:1883"
#define MQTT_CLIENT_ID "rtthread_demo_001"
#define MQTT_USERNAME  ""
#define MQTT_PASSWORD  ""
#define SUB_TOPIC      "rtthread/demo/sub"
#define PUB_TOPIC      "rtthread/demo/pub"

static void mqtt_sub_callback(void *client, message_data_t *msg)
{
    if (msg == RT_NULL || msg->message == RT_NULL) return;
    rt_kprintf("Subscribed msg: %.*s\n",
               msg->message->payloadlen,
               (char *)msg->message->payload);
}

static void mqtt_connect_callback(void *client)
{
    rt_kprintf("MQTT connected\n");
}

static void mqtt_online_callback(void *client)
{
    rt_kprintf("MQTT online\n");
}

static void mqtt_offline_callback(void *client)
{
    rt_kprintf("MQTT offline\n");
}

static void mqtt_demo_thread(void *param)
{
    mqtt_client_t *client = RT_NULL;
    mqtt_client_config_t config = {0};

    config.uri = MQTT_URI;
    config.client_id = MQTT_CLIENT_ID;
    config.username = MQTT_USERNAME;
    config.password = MQTT_PASSWORD;
    config.connect_callback = mqtt_connect_callback;
    config.online_callback = mqtt_online_callback;
    config.offline_callback = mqtt_offline_callback;
    config.sub_callback = mqtt_sub_callback;
    config.clean_session = 1;
    config.keepalive_interval = 60;
    config.buf_size = 1024;
    config.conn_timeout = 3000;
    config.reconnect_timeout = 5000;

    client = mqtt_client_create(&config);
    if (client == RT_NULL) {
        rt_kprintf("MQTT client create failed\n");
        return;
    }

    /* 连接Broker */
    if (mqtt_client_connect(client) != RT_EOK) {
        rt_kprintf("MQTT connect failed\n");
        mqtt_client_destroy(client);
        return;
    }

    /* 订阅主题 */
    mqtt_client_subscribe(client, SUB_TOPIC, QOS1);
    rt_kprintf("Subscribed to: %s\n", SUB_TOPIC);

    /* 循环发布消息 */
    char buf[128];
    int count = 0;
    while (1) {
        rt_snprintf(buf, sizeof(buf), "RT-Thread MQTT msg #%d", count++);
        mqtt_client_publish(client, PUB_TOPIC, buf, rt_strlen(buf), QOS1, 0);
        rt_kprintf("Published: %s\n", buf);
        rt_thread_mdelay(5000);
    }
}

static int mqtt_demo_init(void)
{
    rt_thread_t tid = rt_thread_create("mqtt_demo",
                                       mqtt_demo_thread,
                                       RT_NULL,
                                       2048,
                                       25,
                                       10);
    if (tid != RT_NULL) rt_thread_startup(tid);
    return 0;
}
INIT_APP_EXPORT(mqtt_demo_init);

运行与验证

1. 编译烧录后，打开串口终端（115200-8-N-1）
2. 观察日志依次出现：MQTT connected → MQTT online → Subscribed to: rtthread/demo/sub
3. 每5秒输出一条Published: RT-Thread MQTT msg #N，N递增

验证方法：使用MQTT客户端工具（如MQTTX或mosquitto_sub）连接同一Broker：

mosquitto_sub -h broker.emqx.io -t "rtthread/demo/pub"

若能看到设备发布的递增消息，说明发布链路正常。再执行：

mosquitto_pub -h broker.emqx.io -t "rtthread/demo/sub" -m "hello from pc"

开发板串口应打印Subscribed msg: hello from pc，证明订阅链路正常。

常见问题

• 连接超时：检查网络连通性，确认开发板能ping通broker.emqx.io
• 认证失败：公共Broker通常无需用户名密码，保持为空字符串
• 内存不足：增大线程栈至2048以上，或减小buf_size