RT-Thread-RyanMqtt使用介绍和示例代码（一）RT-Thread问答社区

RyanMqtt使用介绍和示例代码（一）

发布于 2023-09-27 19:53:11 浏览：207 订阅该版

[tocm]

本文提供一个RyanMqtt的基本使用例子,示例工程放在文末
## 关于RyanMqtt
RyanMqtt是一个完整的mqtt3.1.1实现，支持特性非常丰富。你想要的特性它可能都有，推荐你们尝试一下！

阅读此篇文章前推荐看看https://github.com/Ryan-CW-Code/RyanMqtt 的readme，对RyanMqtt有一个基本认知。

**测试环境：stm32F401RCT6、RT-Thread版本: v4.1.0、RT-Thread Studio版本: 2.2.6、网络硬件使用ec800m移植at_socket使用sal框架。**

## 1、添加网络硬件
**RyanMqtt依赖SAL框架 / LWIP。**

此步骤不做过多解释，是lwip就用lwip，是at设备就用at_socket。推荐所有平台都使用SAL框架（RyanMqtt软件包会自动使能）。

## 2、添加RyanMqtt
这里使用的Studio开发环境，点击 RT-Thread Settings，选择添加软件包，搜索RyanMqtt添加。

**开启RyanMqtt msh示例，版本建议选择latest** (latest包含了最新的错误修复和特性，推送前都经过测试的。**但是rt-thread使用gitee镜像，同步需要1-3天，如果要使用最新latest版本请修改为github源**)

添加后如下图所示，直接编译然后烧录
![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/8e4b9d1978b05795621cfafae29177b1.png.webp)

## 3、使用msh测试例子
***网络硬件必须要能连接到网络，否则mqtt没有任何意义***
编译烧录后在串口终端输入 "help" / table键，打印shell命令信息，查看是否有mqtt命令，如下图所示。

*注：没有mqtt命令的请查看RyanMqtt软件包是否开启了msh示例*

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/7ba1ef16ac7cb4762d8eeabb08777d3c.png)

接着我们输入 "mqtt" / "mqtt help" 命令，打印信息如下图所示

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/6c34ac9370d09043262998b8ef0b76ed.png.webp)

***可以看到msh示例程序给出的部分mqtt功能命令，左侧为mqtt示例命令，中间为命令介绍，右侧为需要传递的参数。***

***接下来我们依次试试每个命令的功能。***

#### mqtt state （打印mqtt客户端状态）
打印当前mqtt的状态。状态定义为
```c
    typedef enum
    {
        RyanMqttInvalidState = -1, // 无效状态
        RyanMqttInitState = 0,     // 初始化状态
        RyanMqttStartState,        // 开始状态
        RyanMqttConnectState,      // 连接状态
        RyanMqttDisconnectState,   // 断开连接状态
        RyanMqttReconnectState,    // 重新连接状态
    } RyanMqttState_e;
```
让我们尝试一下
无效状态，因为我们还没有调用连接函数，mqtt客户端还没有被初始化，所以为无效状态
![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/dd71e2f4e136a5b31751e7ac3418d365.png)

#### mqtt connect （mqtt连接服务器）
连接mqtt服务器，可以看到参数为null。连接服务器的配置信息在RyanMqttTest.c文件的头部，如下所示
修改下面配置信息来连接你的mqtt服务器，"broker.emqx.io" 为emqx的开放mqtt服务器。
**注意：mqtt客户端id必须唯一，请务必进行修改！**
![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/2595e391915b3b66fa0b7b1564596211.png.webp)

让我们尝试一下connect

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/7871054421f357fa58bce4b20c16ebdb.png.webp)

可以看到打印了一条 "mqtt连接成功回调" 。为什么会打印呢？ 这我们就要看看mqtt connect命令做了什么？
首先我们要找到cmdTab，这里存放了示例的所有命令。
![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/10874d0aa8ec8cc5476a167f3e64df75.png.webp)

**我们点击connect示例对应的函数 “MqttConnect”**秘密就发生在mqttConfig的回调函数和注册事件中。我们在注册事件回调中选择了所有事件，其中就包括了mqtt连接成功事件。如下图所示
**当mqtt连接成功时就会调用mqttConfig中的mqttEventHandle函数指针**。在mqttEventHandle内部我们打印了一条 "mqtt连接成功回调"

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/a815de338175024c0858c9725c189edd.png.webp)

回调函数内部
![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/00d1ca847c5d3e0799cde90147e16a93.png.webp)

**可注册的回调事件定义为**
![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/e413dd857e78ee998b9b0159ab3294cc.png.webp)

#### mqtt disc （mqtt断开连接服务器）
主动断开mqtt服务器的连接，使用效果如下，回调函数中打印了 "mqtt断开连接回调"
**图中 "204" 为断开连接事件的eventData**，可查看可注册的回调事件定义对于eventData的定义。
***注：为什么断开连接后会自动重连呢？ 我们在调用connect示例的时候使能了 "自动重连" 并定义了重连时间。***
![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/0f0fffac43dfb9f3f6a8b2f4442c96c6.png.webp)

#### mqtt reconnect （mqtt断开连接时重新连接服务器）
当我们没有使能 "自动重连" 时，我们可以手动调用 RyanMqttReconnect 函数来进行重连
*注：如果mqtt客户端不为 RyanMqttDisconnectState 状态，此接口不会进行任何操作*

#### mqtt destory （mqtt销毁客户端）
销毁mqtt客户端，释放mqtt客户端申请的所有资源。

**销毁mqtt客户端因为安全问题，设置为异步操作，当接收到 RyanMqttEventDestoryBefore 事件后才真正开始销毁。真正开始销毁速度非常快！就只是释放资源。**
**至于最多要多长时间才可以销毁？如果网络层移植的没有问题，最长时间为初始化config里的recvTimeout**

*注：客户端需要初始化后的才可以被销毁，否则会忽视*

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/f75531183f2d81662138e1338aa0aaef.png)

#### mqtt pub （mqtt发布消息）
根据命令提示，需要传入 主题、消息等级、发送内容、发送条数、间隔时间(可以为0)
这里我们使用 mqttx工具来做上位机

***使用命令：mqtt pub testup 2 hello 10 0***
![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/17c8dd411d638e48405666d9645f23eb.png)

mqttx截图

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/9a6e7e16af69e73bdd2952b585f897a3.png.webp)

#### mqtt sub （mqtt订阅主题）
接下来我们试一下订阅主题，使用多通配符来测试

##### 先测试 "$" 通配符，"$"比较特殊，应用不能使用"$"开头的系统主题！ 
按规范服务端不能将 $ 字符开头的主题名匹配通配符 (#或+) 开头的主题过滤器

***由于应用不能使用"$"开头的系统主题所以 "broker.emqx.io" 不允许订阅 "$SYS", 但是为了测试我把我的mqtt服务器设置为允许订阅，下图示例使用我的mqtt服务器来进行测试。 如果您测试的时候发现订阅失败请查看您的mqtt服务器是否允许应用订阅"$"开头的系统主题***

*我使用emqx服务端，当订阅 "$SYS/#" 时会触发保留消息打印地址、应用名、版本等信息*

RyanMqtt订阅结果
![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/34d1b9be9e3e37c4f37c677055c6e03c.png.webp)

mqttx订阅结果
![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/3dc6607c0da3b2802f4f27397243a0aa.png.webp)

##### 再来测试 "/"、"#"、"+" 通配符 
**为了快速我就在一个主题里面使用多个通配符，发送符合不同通配符的消息来进行测试**
*订阅主题：testdown/+/nihao/#*

下图中红框发送的是符合订阅主题通配符的，黑框是不符合订阅主题通配符的。
**可以看到RyanMqtt可以准确的接收到通配符消息。**

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/7463fd88585c29618dd62be8c95786cd.png.webp)

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/0b29a87f5338e605d6da43d9e529dc6e.png.webp)

#### mqtt unsub （mqtt取消订阅主题）
取消订阅主题，**取消没订阅的主题时会自动忽略。**

可以看到取消订阅后再发送消息，RyanMqtt就不会收到了
*下图第一个红框为第一次取消可以触发回调，第二次取消就没有任何响应。*

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/6d66fcd16c959e5ffba389b4b34df500.png.webp)

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/df78533774d31481a133d4c43266b126.png.webp)

#### mqtt listsub （mqtt获取已订阅主题）
打印结果执行顺序为:

***打印已订阅主题 --> 订阅"testdown/+/nihao/#"主题 --> 打印已订阅主题 --> 订阅"testdown2"主题 --> 打印已订阅主题 --> 取消订阅"testdown/+/nihao/#"主题 -->  打印已订阅主题***

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/953a694f0f139e1dd6db9c01f0c579a3.png.webp)

#### mqtt listack （打印ack链表,辅助功能）
***ack链表包含发送qos1 / qos2 的ack报文、接收qos1 / qos2 的ack报文、订阅 / 取消订阅主题的ack报文。***
根据上面的描述可以知道 ack链表 通常都应该为空。只有在上诉情况下才会存在，但是碍于篇幅这里无法进行测试了，**等下一篇文章测试qos1 / qos2消息稳定性时再进行展示**

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/c3303677b0993e9ded6e99e71ceada52.png.webp)

#### mqtt listmsg （打印msg链表,辅助功能）
**msg链表保存着订阅主题的信息**，接收消息、取消订阅的时候都会操作msg链表。
所以**listmsg和listsub是一摸一样的操作**，结果自然也就一样，这里就不展示了

#### mqtt data （打印测试信息用户自定义的）
此接口我用来测试qos消息稳定性，没有实际意义。

## 4、将RyanMqtt添加到自己项目代码里，不使用msh示例
**上面我们使用msh示例来进行RyanMqtt的测试，但在项目中我们肯定不会通过msh来操作mqtt，所以我们将根据msh示例来将RyanMqtt添加到代码里**
首先我们思考下mqtt的执行流程，根据示例来看我们需要

**处理订阅消息**
连接mqtt服务器 --> 订阅主题 --> 收到订阅主题的消息 --> 在回调函数里面消费消息（调函数的执行环境是mqtt客户端的线程，**所以非常不建议在回调函数里面做复杂逻辑操作**，一是**会阻塞mqtt线程运行**，二是可能会**导致mqtt线程爆栈**。还是看使用场景如果需要串行处理mqtt消息在回调里面使用是挺好的）

**发布消息**
连接mqtt服务器 --> 发布主题消息(qos1 / qos2会有发送成功或者超过重发次数的回调)

**重连逻辑**
- 配置mqtt客户端自动重连 --> 连接服务器
- 不配置mqtt客户端自动重连 --> 连接服务器 --> 获取mqtt客户端状态（断连状态手动调用重连函数）

根据上面的处理方式我们来进行代码编写，为了方便我都放在main函数了

这是原始main函数，只设置了netdev状态变更回调
![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/3496ea467680d224d89de5d4d9845563.png.webp)

先添加头文件，如下图

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/902726c2383bf2666f6226e741b9bbe1.png.webp)

**1、先添加连接服务器函数**，这里直接将msh示例中的connect函数复制到main.c，**并处理报错地方(自行处理都很简单，都是资源未定义）**，图片放不下我就不截图了

**2、再添加订阅主题函数**，**订阅主要要等mqtt连接成功后**。所以不管有没有使能clearSession，都非常推荐**在连接成功回调函数中订阅主题** 如下图所示
![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/156b473a9ebe61b8eb729f8ece84948d.png.webp)

**3、再添加发布消息**，这里简单起见直接加个循环，每10秒发送一个hello。(**mqtt没有连接成功时也可以发布但是mqtt客户端不会进行处理，会直接丢弃不管qos等级**) 如下图所示
![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/66eef9836874f1f6e24bef4116d4a24a.png.webp)

**4、消费订阅主题的消息**，这里是直接**打印出来消息的主题、报文id、载荷长度、载荷指针**，推荐通过**消息队列发送到别的线程进行处理**，或者您如果知道在**回调函数处理的副作用**的话可以在回调函数中直接处理

***注：载荷指针最后一位没有 "\0"，可能会存在脏数据，需要用户手动处理。下图给出一种处理方案***
**或者mqtt消息一般为json，可以使用RyanJson(https://github.com/Ryan-CW-Code/RyanJson) / cJSON来处理json数据，都可以自动识别尾部脏数据**
![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/3c2c9895ad7f012b79f852ba7076cb98.png.webp)

**5、这样我们就添加完成了，烧录进行测试**
如下面两个图所示，每秒上传一次消息，接收到消息后进行打印出来。结果如我们所想
![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/f7ffb65910d8a0c4264718f9bb1909cd.png.webp)

![screenshot_image.png](https://oss-club.rt-thread.org/uploads/20230927/0e182f450eb1e78c8aaa2b2ef461f972.png.webp)

## 5、总结
这篇文章简单介绍了RyanMqtt的使用，包括msh示例和添加到自己的工程代码里面。但文章碍于篇幅限制**介绍的始终很浅，想深入的了解还是要看代码，RyanMqtt注释都为中文**。

接下来应该还会写两篇文章介绍RyanMqtt，一篇进行**RyanMqtt的qos1 / qos2消息等级的稳定性测试**，另一篇介绍下**RyanMqtt移植指南**

后面看要不要详细介绍一下mqtt协议，RyanMqtt代码仓库docs/下有mqtt3.1.1协议中文版本pdf，一般来说看文档就可以了。
以及mqtt5.0，5.0增加了很多激动人心的特性，在考虑要不要适配一下，看大家需求了，可能遥遥无期哈哈哈哈哈。

[项目代码示例.zip](https://club.rt-thread.org/file_download/d5ec35284f851ed1)