3. 使用RT-Thread连接EMQX云平台

任务目标

在EMQX Cloud创建MQTT Broker部署，配置设备用户名与密码认证；基于RT-Thread的paho_mqtt软件包完成设备接入，实现主题订阅与消息发布，验证端到端通信链路。

前置准备

  • EMQX Cloud（serverless或专有版）已创建部署，获取连接地址、端口（默认8083/8084）与连接凭证。
  • RT-Thread工程已开启paho_mqtt软件包（ENV中menuconfig勾选，依赖netutils与lwIP）。
  • 开发板已能正常联网（ping通外网）。

EMQX Cloud侧配置

  1. 在EMQX Cloud控制台创建部署，选择版本与地域。
  2. 进入认证 → 添加认证，选择用户名密码方式，添加设备：device_001 / pass1234。
  3. 在访问控制 → 授权中，为device_001添加发布/订阅权限（允许test/topic/#）。
  4. 记录部署的连接地址（形如broker.emqx.io或自定义域名）与TCP端口（默认1883，TLS为8883）。

核心代码实现

以下代码基于RT-Thread的paho_mqtt软件包，完成MQTT连接、订阅与发布。将代码保存为mqtt_emqx_sample.c，加入工程编译。

#include <rtthread.h>
#include <stdio.h>
#include <string.h>
#include "paho_mqtt.h"

#define MQTT_HOST       "your-deployment-id.emqxsl.com"  // 替换为实际地址
#define MQTT_PORT        1883
#define MQTT_USERNAME   "device_001"
#define MQTT_PASSWORD   "pass1234"
#define MQTT_CLIENTID   "rtthread-device-001"
#define SUB_TOPIC       "test/topic/data"
#define PUB_TOPIC       "test/topic/cmd"

static MQTTClient client;
static int msg_count = 0;

/* 消息到达回调 */
static void mqtt_message_arrived(MQTTClient *c, char *topicName, int topicLen,
                                 MQTTMessage *message)
{
    rt_kprintf(" [SUB] topic: %.*s, payload: %.*s\n",
                topicLen, topicName,
                (int)message->payloadlen, (char *)message->payload);
}

/* 连接成功回调 */
static void mqtt_connect_callback(MQTTClient *c)
{
    rt_kprintf(" MQTT connected to EMQX\n");
}

/* 连接断开回调 */
static void mqtt_disconnect_callback(MQTTClient *c)
{
    rt_kprintf(" MQTT disconnected\n");
}

/* 发布线程 */
static void mqtt_publish_thread(void *param)
{
    char buf[64];
    MQTTMessage msg;

    while (1)
    {
        rt_thread_mdelay(5000);  // 每5秒发布一次

        if (MQTTIsConnected(&client))
        {
            msg_count++;
            rt_snprintf(buf, sizeof(buf), "hello from rt-thread #%d", msg_count);
            msg.qos = 1;
            msg.retained = 0;
            msg.payload = buf;
            msg.payloadlen = strlen(buf);

            if (MQTTPublish(&client, PUB_TOPIC, &msg) == SUCCESS)
                rt_kprintf(" [PUB] %s -> %s\n", PUB_TOPIC, buf);
            else
                rt_kprintf(" [PUB] failed\n");
        }
    }
}

/* MQTT初始化与连接 */
static int mqtt_emqx_init(void)
{
    MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
    Network n;
    int rc;

    /* 初始化网络 */
    NetworkInit(&n);
    if (NetworkConnect(&n, MQTT_HOST, MQTT_PORT) != 0)
    {
        rt_kprintf(" Network connect failed\n");
        return -1;
    }

    /* 配置连接参数 */
    data.clientID.cstring = MQTT_CLIENTID;
    data.username.cstring = MQTT_USERNAME;
    data.password.cstring = MQTT_PASSWORD;
    data.keepAliveInterval = 30;
    data.cleansession = 1;

    MQTTClientInit(&client, &n, 1024, 1024);
    MQTTClientSetCallbacks(&client, NULL, mqtt_connect_callback,
                           mqtt_message_arrived, mqtt_disconnect_callback);

    rc = MQTTConnect(&client, &data);
    if (rc != SUCCESS)
    {
        rt_kprintf(" MQTT connect failed, rc=%d\n", rc);
        return -1;
    }

    /* 订阅主题 */
    if (MQTTSubscribe(&client, SUB_TOPIC, 1) != SUCCESS)
    {
        rt_kprintf(" Subscribe failed\n");
        return -1;
    }
    rt_kprintf(" Subscribed to %s\n", SUB_TOPIC);

    /* 创建发布线程 */
    rt_thread_t tid = rt_thread_create("mqtt_pub", mqtt_publish_thread,
                                       RT_NULL, 2048, 20, 10);
    if (tid)
        rt_thread_startup(tid);

    return 0;
}

/* 导出到MSH命令 */
static void mqtt_start(void)
{
    if (mqtt_emqx_init() != 0)
        rt_kprintf(" MQTT start failed\n");
}
MSH_CMD_EXPORT(mqtt_start, start mqtt connection to emqx);

编译与烧录

  1. 将上述代码放入工程applications目录。
  2. 在ENV中确认已开启paho_mqtt软件包，并配置网络（如ESP8266/以太网）。
  3. 使用scons编译，烧录固件至开发板。

运行与验证

  1. 开发板启动后，在MSH控制台输入mqtt_start。
  2. 观察串口输出，预期出现MQTT connected to EMQX与Subscribed to test/topic/data。
  3. 每5秒应看到[PUB] test/topic/cmd -> hello from rt-thread #N，N递增。
  4. 在EMQX Cloud控制台的在线客户端页面，应能看到rtthread-device-001处于在线状态。
  5. 使用MQTT X或EMQX控制台的WebSocket客户端，向test/topic/data发送消息，开发板串口应打印[SUB]开头的接收日志。

通过条件：开发板成功连接EMQX并保持在线；发布消息能在MQTT X中收到；MQTT X发送的消息能被开发板订阅接收。若连接失败，优先检查地址端口、用户名密码及EMQX授权规则。

← 上一章 📖 返回目录 下一章 →