玩转RT-Thread系列教程(13)--MQTT协议通信

Posted 2021-07-17 Rb菌

玩转RT-Thread系列教程(13)–MQTT协议通信

一、了解一下MQTT

1.MQTT介绍

​ 客户端 Client

​ 使用MQTT的程序或设备。客户端总是通过网络连接到服务端。它可以

• 发布应用消息给其它相关的客户端。

• 订阅以请求接受相关的应用消息。

• 取消订阅以移除接受应用消息的请求。

• 从服务端断开连接。

  服务端 Server
  一个程序或设备，作为发送消息的客户端和请求订阅的客户端之间的中介。服务端

• 接受来自客户端的网络连接。

• 接受客户端发布的应用消息。

• 处理客户端的订阅和取消订阅请求。

• 转发应用消息给符合条件的已订阅客户端。

说白了一方为供应商，一方为消费者（Client），供应商（Server）一旦和消费者产生了联系，那么供应商（Server）就会提供商品给消费者，同时消费者（Client）也可以向供应商提供意见。

订阅 Subscription
订阅包含一个主题过滤器（Topic Filter）和一个最大的服务质量（QoS）等级。订阅与单个会话（Session）关联。会话可以包含多于一个的订阅。会话的每个订阅都有一个不同的主题过滤器。

• QoS0，At most once，至多一次；Sender 发送的一条消息，Receiver 最多能收到一次，如果发送失败，也就算了。
• QoS1，At least once，至少一次；Sender 发送的一条消息，Receiver 至少能收到一次，如果发送失败，会继续重试，直到 Receiver 收到消息为止，但Receiver 有可能会收到重复的消息
• QoS2，Exactly once，确保只有一次。Sender 尽力向 Receiver 发送消息，如果发送失败，会继续重试，直到 Receiver 收到消息为止，同时保证 Receiver 不会因为消息重传而收到重复的消息。

2.MQTT协议数据包结构

一个MQTT数据包由：固定头（Fixed header）、可变头（Variable header）、有效载荷（payload）三部分构成。

• （1）固定头（Fixed header）。存在于所有MQTT数据包中，表示数据包类型及数据包的分组类标识。
• （2）可变头（Variable header）。存在于部分MQTT数据包中，数据包类型决定了可变头是否存在及其具体内容。
• （3）有效载荷（Payload）。存在于部分MQTT数据包中，表示客户端收到的具体内容。

固定报头

byte1		MQTT报文类型（1）	Reserved 保留位
	10	0	0	0	1	0	0	0	0
byte2	剩余长度
	27	0	0	1	0	0	1	1	1

可变报头

	说明	7	6	5	4	3	2	1	0
协议名
byte 1	长度 MSB (0)	0	0	0	0	0	0	0	0
byte 2	长度 LSB (4)	0	0	0	0	0	1	0	0
byte 3	‘M’	0	1	0	0	1	1	0	1
byte 4	‘Q’	0	1	0	1	0	0	0	1
byte 5	‘T’	0	1	0	1	0	1	0	0
byte 6	‘T’	0	1	0	1	0	1	0	0

这里我用Onenet的连接报文示例，其他的同理

二、RTT中MQTT组件

1.menuconfig进入env，选择IOT组件

2.选择Paho MQTT组件以及CJSON解析组件

Eclipse paho是eclipse基金会下面的一个开源项目，基于MQTT协议的客户端

json是一种轻量级的数据交换格式，简洁和清晰的层次结构使得 JSON 成为理想的数据交换语言，易于人阅读和编写

cjson组件则是可以对服务器端发来的json数据进行解析，对比于常规的使用C库解析字符串方式，cjson为我们封装好了解析方法，调用更加灵活方便。

3.使用pkgs --update更新、下载软件包

4.使用scons --target=md5生成mdk工程

5.打开mdk工程

可以看见，我们选择的paho以及cjson组件已经添加到了我们的工程中

三、MQTT的使用

3.1、编写MQTT客户端

1.宏定义连接mqtt服务器需要的参数：

#define MQTT_Uri    "tcp://xxx.xxx.xxx:1883"     		// MQTT服务器的地址和端口号
#define ClientId    "751061401"                     // ClientId需要唯一
#define UserName    "rb"                        		// 用户名
#define PassWord    "123456"                        // 用户名对应的密码

2.定义一个mqtt客户端结构体变量

/* 定义一个MQTT客户端结构体 */
static MQTTClient client;

3.对MQTT进行配置

    /* 用以存放邮件 */
    rt_ubase_t* buf;
    /* 对MQTT客户端结构体变量进行配置 */
    client.isconnected = 0;
    client.uri = MQTT_Uri;

    /* 配置MQTT的连接参数 */
    MQTTPacket_connectData condata = MQTTPacket_connectData_initializer;
    memcpy(&client.condata, &condata, sizeof(condata));
    client.condata.clientID.cstring = ClientId;
    client.condata.keepAliveInterval = 30;
    client.condata.cleansession = 1;
    client.condata.username.cstring = UserName;
    client.condata.password.cstring = PassWord;

4.为MQTT的消息缓存申请内存

    /* 为mqtt申请内存 */
    client.buf_size = client.readbuf_size = 1024;
    client.buf = rt_calloc(1, client.buf_size);
    client.readbuf = rt_calloc(1, client.readbuf_size);

    if (!(client.buf && client.readbuf))
    {
        rt_kprintf("no memory for MQTT client buffer!\\r\\n");
        return;
    }

5.设置回调函数，以及订阅主题；设置默认的回调函数，是在如果有订阅的 Topic 没有设置回调函数时，则使用该默认回调函数

    /* 设置回调函数  */
    client.connect_callback = mqtt_connect_callback;
    client.online_callback = mqtt_online_callback;
    client.offline_callback = mqtt_offline_callback;

    /* 订阅一个主题，并设置其回调函数 */
    client.messageHandlers[0].topicFilter = rt_strdup(ONENET_MQTT_SUBTOPIC);
    client.messageHandlers[0].callback = mqtt_sub_callback;
    client.messageHandlers[0].qos = QOS1;

    /* 设置默认的回调函数 */
    client.defaultMessageHandler = mqtt_sub_default_callback;

6.启动 mqtt client

    /* 启动 mqtt client */
		paho_mqtt_start(&client);

7.实现各个回调函数

/* 默认的订阅回调函数，如果有订阅的 Topic 没有设置回调函数，则使用该默认回调函数 */
static void mqtt_sub_default_callback(MQTTClient *c, MessageData *msg_data)
{
    *((char *) msg_data->message->payload + msg_data->message->payloadlen) = '\\0';
    rt_kprintf("%.*s\\r\\n", msg_data->message->payloadlen, (char *) msg_data->message->payload);
}

/* 连接成功回调函数 */
static void mqtt_connect_callback(MQTTClient *c)
{
    rt_kprintf("success connect to mqtt! \\r\\n");
}

/* 上线回调函数 */
static void mqtt_online_callback(MQTTClient *c)
{
    rt_kprintf("mqtt is online \\r\\n");
    client.isconnected = 1;
}

/* 下线回调函数 */
static void mqtt_offline_callback(MQTTClient *c)
{
    rt_kprintf("mqtt is offline \\r\\n");
}

8.设置收到订阅主题的消息时的回调函数

static void mqtt_sub_callback(MQTTClient *c, MessageData *msg_data)
{
    cJSON *root = RT_NULL, *object = RT_NULL;
    root = cJSON_Parse((const char *) msg_data->message->payload);

    if (!root)
    {
        rt_kprintf("No memory for cJSON root!\\n");
				cJSON_Delete(root);
        return;
    }
		
    object = cJSON_GetObjectItem(root, "led1");
    if (object->type == cJSON_Number)
        led1 = object->valueint;		
		
    object = cJSON_GetObjectItem(root, "led2");
    if (object->type == cJSON_Number)
        led2 = object->valueint;			
		
		led1==1?LED1_ON:LED1_OFF;
		led2==1?LED2_ON:LED2_OFF;	
			
		if (NULL != root)
		{
				cJSON_Delete(root);
				root = NULL;
		}		
}

3.2、MQTT消息推送

既然我们已经使设备连接上了服务器，那么如何将我们的数据发送到MQTT服务器呢？

没错，我们需要向服务器端订阅的topic发送消息，使得服务器可以收到我们的“悄悄话”

paho已经已经为我们封装好了消息推送的函数–paho_mqtt_publish(MQTTClient *client, enum QoS qos, const char *topic, const char *msg_str)

3.2.1、对之前处理进程同步的优化

对于之前我们处理两个线程之间消息同步，我们采用的是通过事件来进行消息的同步。但经过测试发现了问题，我们没有必要去进行两次的事件的接收，

对于第二次我们完全可以直接通过消息邮箱进行判断是否收到了消息。

当我们收到了ADC采集到的光敏数据以及485的温湿度数据，我们就可以将传感器的数据发送到我们的服务器了。

//ADC数据接收事件if (rt_event_recv(Sensor_event, EVENT_ADC_FLAG, RT_EVENT_FLAG_OR | RT_EVENT_FLAG_CLEAR, RT_WAITING_FOREVER, &e) != RT_EOK)  continue;//5s等待接收邮件if(rt_mb_recv(Sensor_msg_mb, (rt_ubase_t*)&sensor_msg, rt_tick_from_millisecond(5000)) == RT_EOK){  sprintf((char*)r_buff, "{\\"temperature\\":%.2f,\\"humidity\\":%.2f}", (float)sensor_msg->temp, (float)sensor_msg->hum); //拼接到温度数组里  if(connect_sta == RT_TRUE && client.isconnected)  {    if(paho_mqtt_publish(&client, QOS1, "topic_pub", (char *) r_buff) != RT_EOK)      rt_kprintf("mqtt publish failed...\\n");    else      rt_kprintf("onenet upload OK >>> %s\\n", r_buff);  }  //释放内存块  rt_mp_free(sensor_msg);  sensor_msg = RT_NULL;  //继续接收  continue;}//10s超时直接存数据rt_kputs("@5s接收事件超时--存储数据\\n");//发送采集数据指令rt_device_write(serial, 0, sensor_T_H, sizeof(sensor_T_H));  continue;

3.3、MQTT消息订阅

既然我们已经成功的将数据推送到了服务器，那么我们的设备如何收到MQTT服务器端发来的消息呢？

没错，接下来我们需要订阅服务器端发布的topic，使得设备可以收到服务器的“悄悄话”

我们只需要实现一下订阅主题的回调函数

static void mqtt_sub_callback(MQTTClient *c, MessageData *msg_data){}

3.3.1、使用CJSON解析数据

因为我们服务器发送的数据是json格式的，这时候cjson便派上了用场

---

常用的cjson函数：

void cJSON_Delete(cJSON *c)

删除 cJSON 指针，由于我们会频繁的使用cJSON，也就是会频繁的申请内存， 这就相当于向内存借空间。

如果有借不还，很快就会将内存用空，导致系统崩溃。

char *cJSON_Print(cJSON *item)

它是将cJSON数据解析成JSON字符串，并会在堆中开辟一块char *的内存空间，存放JSON字符串。

函数成功后会返回一个char *指针，该指针指向位于堆中JSON字符串。

cJSON *cJSON_Parse(const char *value)

将一个JSON数据包，按照cJSON结构体的结构序列化整个数据包，并在堆中开辟一块内存存储cJSON结构体

返回值：成功返回一个指向内存块中的cJSON的指针，失败返回NULL

cJSON *cJSON_GetObjectItem(cJSON *object,const char *string)

获取JSON字符串字段值，成功返回一个指向cJSON类型的结构体指针，失败返回NULL

3.3.2、MQTT消息订阅

cJSON *root = RT_NULL, *object = RT_NULL;
root = cJSON_Parse((const char *) msg_data->message->payload);

if (!root)
{
  rt_kprintf("No memory for cJSON root!\\n");
  cJSON_Delete(root);
  return;
}

object = cJSON_GetObjectItem(root, "led1");
if (object->type == cJSON_Number)
  led1 = object->valueint;		

object = cJSON_GetObjectItem(root, "led2");
if (object->type == cJSON_Number)
  led2 = object->valueint;			

led1==1?LED1_ON:LED1_OFF;
led2==1?LED2_ON:LED2_OFF;	

if (NULL != root)
{
  cJSON_Delete(root);
  root = NULL;
}		

使用cJSON_GetObjectItem获取JSON字符串字段值，解析服务器发送的json字符串{“ledX”:1}

这里调用函数就会将字符串{“ledX”:1}进行拆分：“ledX的数据为1”，那么我们获取到了ledX的数值之后，岂不是想干嘛干嘛~

四、编译、下载、验证

以上为设备启动后的消息，可以看出，当我们没有成功联网时mqtt会一直进行重连，直到成功联网后，才会订阅到topic以及进行消息的转发处理

这里我使用emq测试mqtt的通信，可以看到我们订阅的topic_pub主题成功的收到了设备发送的消息

同时向topic_sub推送消息设备也可以正常接收，至此我们的MQTT通信成功的得以已验证。