【Ai-WB2高级篇】MQTT 协议连接

【Ai-WB2高级篇】MQTT 协议连接

MQTT
MQTT（Message Queuing Telemetry Transport）是一种轻量级适用于资源受限的设备下使用，特别是嵌入式领域

MQTT特点

• 轻量级：MQTT开销低、报文小非常适合资源受限的设备
• 可靠：MQTT支持多种QoS等级、会话感知和持久连接
• 安全通信： 支持TLS和SSL加密功能，还可以通过用户名/密码方式进行认证
• 双向通信： MQTT基于**发布-订阅**模式为设备之间提供了无缝双向通信，既可以发布消息也可以订阅消息，无须设备之间相互耦合，易于扩展
• 大规模物联网设备支持：MQTT轻量级、低带宽消耗，通过发布-订阅模式解耦设备与设备之间的关系，从而有效的网络流量和资源的使用
• 语言支持： 编程语言支持多
  

MQTT基本组件
发布订阅模式： 为了解决物联网设备之间的解耦问题，MQTT采用了发布-订阅模式，与传统的客户端-服务端的模式不同，MQTT发布者和订阅者无须知道对方的存在，无须建立直接的连接，通过MQTT Broker来负责消息的路由和分发

MQTT设备将温度数据发布到指定的Temperature主题，MQTT Broker收到消息后会将改数据转发给该主题的订阅者

主题topic
MQTT通过topic对发布和订阅的客户端进行管理，类似于分组，发布者可以发送订阅topic的所有组内成员，topic通过/来分割，支持+和#两种通配符

• +： 表示单层匹配，例如：/a/+，匹配/a/b，和a/c
• #： 表示多层匹配，例如：a/#，匹配/a/x，/a/b/c
  

需要注意的是通配符只能用于订阅，不能用于发布

服务质量 QoS
为了适应不同的网络环境，MQTT提供了三种服务质量等级

• QoS 0： 消息最多传送一次。如果客户端不可用，将丢失这条消息
• QoS 1：消息至少传送一次，保证可以收到消息，但可能会重复
• QoS 2： 消息只传递一次，保证即不丢失也不重复
  

QoS等级从低到高消息可靠性增高，传输复杂度也提升

MQTT工作流程
客户端通过TCP/IP协议与MQTT Broker建立连接，可以选择TLS/SSL加密来实现，建立连接后既可以发布指定的主题也可以定于指定的主题，客户端发布时消息需要指定topic和QoS，Broker收到消息后会根据不同的QoS等级按照不通过的策略将消息分发给该主题的订阅者们

MQTT客户端
MQTTX是一款免费开源的全平台客户端，完全兼容MQTT协议MQTT官网

免费公共MQTT服务器
MQTT免费服务器
服务器信息

• 地址：broker.emqx.io
• mqtt端口：1883
• ws端口：8083
• mqtts端口：8883
• wss端口：8084
  

证书文件可以从官方手动下载


连接成功后可以设置订阅内容或者发送消息


代码编写
MQTT需要用到lwip和网络相关的组件，需要在Makefile配置文件中添加相关的依赖

1. COMPONENTS_BLSYS := bltime blfdt blmtd bl_os_adapter rfparam_adapter_tmp bl602_os_adapter bl602 bl602_std
   
2. COMPONENTS_LOOPS := bloop loopadc looport yloop loopset looprt
   
3. COMPONENTS_VFS := romfs
   
4. COMPONENTS_NETWORK := sntp dns_server lwip lwip_dhcpd
   
5. COMPONENTS_MQTT := axk_common http-parser axk_tls axk_mqtt wpa_supplicant tcp_transport blcrypto_suite
   
6. COMPONENTS_WIIF := wifi wifi_manager wifi_hosal
   
7. 
   
8. INCLUDE_COMPONENTS += freertos_riscv_ram newlibc
   
9. INCLUDE_COMPONENTS += mbedtls_lts vfs hosal coredump blog
   
10. INCLUDE_COMPONENTS += utils cli cjson
    
11. 
    
12. INCLUDE_COMPONENTS += $(COMPONENTS_LOOPS)
    
13. INCLUDE_COMPONENTS += $(COMPONENTS_BLSYS)
    
14. INCLUDE_COMPONENTS += $(COMPONENTS_VFS)
    
15. INCLUDE_COMPONENTS += $(COMPONENTS_NETWORK)
    
16. INCLUDE_COMPONENTS += $(COMPONENTS_WIIF)
    
17. INCLUDE_COMPONENTS += $(COMPONENTS_MQTT)
    
18. 
    
19. INCLUDE_COMPONENTS += $(PROJECT_NAME)

复制代码

在获取到IP地址的地方开始mqtt的连接，使用的是sdk里面提供的axk_mqtt组件，其配置连接代码如下
  

1. axk_mqtt_client_config_t mqtt_cfg = {
   
2.         .uri = "ws://broker.emqx.io:8083/mqtt",
   
3.         .event_handle = event_cb,
   
4. 
   
5.         //指定客户端名称
   
6.         .client_id = "axk mqtt",
   
7.         //设置服务器证书文件
   
8.         // .cert_pem = CA_CRT,
   
9.         // .cert_len = strlen(CA_CRT) + 1,
   
10.         // .username = "wb2",
    
11.         // .password = "wb2",
    
12.         
    
13.         //客户端证书
    
14.         // .client_cert_pem = CLIENT_CA,
    
15.         // .client_key_pem = CLIENT_KEY,
    
16. 
    
17.         
    
18.     };
    
19. 
    
20.     axk_mqtt_client_handle_t client = axk_mqtt_client_init(&mqtt_cfg);
    
21.     axk_mqtt_client_start(client);
    

复制代码

可以通过event->event_id来判断消息的类型，常用的的类型有一下几种

• MQTT_EVENT_CONNECTED：连接到mqtt服务器
• MQTT_EVENT_DISCONNECTED：断开连接
• MQTT_EVENT_SUBSCRIBED：订阅成功
• MQTT_EVENT_PUBLISHED：发送成功
• MQTT_EVENT_DATA：接收到消息
  

MQTT_EVENT_DATA用来接收消息，通过event->topic来判断是什么类型的消息，从而做出对应的操作，event->data是消息的内容

常用的函数

1. 
   
2. //初始化mqtt配置，并返回一个client对象
   
3. axk_mqtt_client_handle_t axk_mqtt_client_init(const axk_mqtt_client_config_t *config);
   
4. 
   
5. //重新连接
   
6. axk_err_t axk_mqtt_client_reconnect(axk_mqtt_client_handle_t client);
   
7. 
   
8. //断开连接
   
9. axk_err_t axk_mqtt_client_disconnect(axk_mqtt_client_handle_t client);
   
10. 
    
11. //设置client连接地址
    
12. axk_err_t axk_mqtt_client_set_uri(axk_mqtt_client_handle_t client, const char *uri);
    
13. 
    
14. //开始启动mqtt client
    
15. axk_err_t axk_mqtt_client_start(axk_mqtt_client_handle_t client);
    
16. 
    
17. //设置client配置
    
18. axk_err_t axk_mqtt_set_config(axk_mqtt_client_handle_t client, const axk_mqtt_client_config_t *config);
    
19. 
    
20. //订阅主题
    
21. int axk_mqtt_client_subscribe(axk_mqtt_client_handle_t client, const char *topic, int qos);
    
22. 
    
23. //取消订阅主题
    
24. int axk_mqtt_client_unsubscribe(axk_mqtt_client_handle_t client, const char *topic);
    
25. 
    
26. //发送主题消息
    
27. int axk_mqtt_client_publish(axk_mqtt_client_handle_t client, const char *topic, const char *data, int len, int qos, int retain);
    
28. 
    
29. //注册回调事件
    
30. axk_err_t axk_mqtt_client_register_event(axk_mqtt_client_handle_t client, axk_mqtt_event_id_t event, axk_event_handler_t event_handler, void *event_handler_arg);

复制代码

案例
通过mqtt协议来控制wb2板载led灯，首先创建wb2_led.h，代码如下

1. #ifndef _WB2_LED_H
   
2. #define _WB2_LED_H
   
3. 
   
4. #include <stdint.h>
   
5. 
   
6. #define PIN_R   14
   
7. #define PIN_G   17
   
8. #define PIN_B   3
   
9. 
   
10. #ifdef __cplusplus
    
11. extern "C" {
    
12. #endif
    
13. 
    
14.     /**
    
15.      * 初始化led
    
16.      * @param  pin
    
17.      */
    
18.     void wb2_led_init(uint8_t pin);
    
19.     /**
    
20.      * 打开指定颜色的灯
    
21.      * @param  led     r、g、b
    
22.      */
    
23.     void wb2_led_open(const char* led);
    
24.     /**
    
25.      * 关闭指定颜色的灯
    
26.      * @param  led     r、g、b
    
27.      */
    
28.     void wb2_led_close(const char* led);
    
29. 
    
30.     /**
    
31.      * 获取指定灯的状态
    
32.      * @param  led     r、g、b
    
33.      * @return int 0 close，1open
    
34.      */
    
35.     uint8_t wb2_led_status(const char* led);
    
36. 
    
37. #ifdef __cplusplus
    
38. }
    
39. #endif
    
40. 
    
41. #endif

复制代码

创建wb2_led.c，代码实现如下

1. #include "wb2_led.h"
   
2. #include <bl_gpio.h>
   
3. #include <string.h>
   
4. #include <stdint.h>
   
5. #include <bl602.h>
   
6. #include <glb_reg.h>
   
7. #include <stdio.h>
   
8. 
   
9. /**
   
10. * bit0: b
    
11. * bit1: g
    
12. * bit2: r
    
13. */
    
14. static uint8_t rgb = 0;
    
15. 
    
16. 
    
17. void update_status(uint8_t pin, uint8_t data)
    
18. {
    
19.     int mask = 0;
    
20.     switch (pin)
    
21.     {
    
22.         case PIN_R:
    
23.             mask = 2;
    
24.             break;
    
25.         case PIN_G:
    
26.             mask = 1;
    
27.             break;
    
28.         case PIN_B:
    
29.             mask = 0;
    
30.             break;
    
31.     }
    
32.     if (data) {
    
33.         rgb |= 1 << mask;
    
34.     }
    
35.     else {
    
36.         rgb &= ~(1 << mask);
    
37.     }
    
38. }
    
39. 
    
40. static int get_pin(const char* led)
    
41. {
    
42.     int pin = -1;
    
43.     if (strcmp(led, "r") == 0) {
    
44.         pin = PIN_R;
    
45.     }
    
46.     else if (strcmp(led, "g") == 0) {
    
47.         pin = PIN_G;
    
48.     }
    
49.     else if (strcmp(led, "b") == 0) {
    
50.         pin = PIN_B;
    
51.     }
    
52.     return pin;
    
53. }
    
54. 
    
55. /**
    
56. * 打开指定颜色的灯
    
57. * @param  led     r、g、b
    
58. */
    
59. void wb2_led_open(const char* led)
    
60. {
    
61.     if (get_pin(led) != -1) {
    
62.         update_status(get_pin(led), 1);
    
63.         bl_gpio_output_set(get_pin(led), 1);
    
64.     }
    
65. 
    
66. }
    
67. /**
    
68. * 关闭指定颜色的灯
    
69. * @param  led     r、g、b
    
70. */
    
71. void wb2_led_close(const char* led)
    
72. {
    
73.     if (get_pin(led) != -1) {
    
74.         update_status(get_pin(led), 0);
    
75.         bl_gpio_output_set(get_pin(led), 0);
    
76.     }
    
77. }
    
78. 
    
79. /**
    
80. * 获取指定灯的状态
    
81. * @param  led     r、g、b
    
82. * @return int 0 close，1open
    
83. */
    
84. uint8_t wb2_led_status(const char* led)
    
85. {
    
86.     int mask = 0;
    
87.     switch (get_pin(led))
    
88.     {
    
89.         case PIN_R:
    
90.             mask = 2;
    
91.             break;
    
92.         case PIN_G:
    
93.             mask = 1;
    
94.             break;
    
95.         case PIN_B:
    
96.             mask = 0;
    
97.             break;
    
98.     }
    
99.     return (rgb >> mask) & 0x1;
    
100. }
     
101. 
     
102. void wb2_led_init(uint8_t pin)
     
103. {
     
104.     bl_gpio_enable_output(pin, 0, 1);
     
105.     bl_gpio_output_set(pin, 0);
     
106. }

复制代码

main函数中调用wb2_led_init初始化led，并在mqtt连接后订阅/on/led/wb2、/off/led/wb2、以及/get/led/wb2 三个主题，然后再收到消息后去处理消息的内容，消息内容
例如：
发送主题/on/led/wb2的内容为r表示打开wb2板载的r led灯
发送主题/off/led/wb2的内容为r表示关闭wb2板载的r led灯
发送主题/get/led/wb2的内容为r表示获取wb2板载的r led灯的状态，并发送到/status/led/wb2主题，1为打开，0为关闭
main中代码如下

1. #define TOPIC_ON "/on/led/wb2"
   
2. #define TOPIC_OFF "/off/led/wb2"
   
3. #define TOPIC_GET "/get/led/wb2"
   
4. #define TOPIC_STATUS "/status/led/wb2"
   
5. 
   
6. /**
   
7. * 根据不同的tpioc处理事件
   
8. * @param  data
   
9. * @param  topic
   
10. */
    
11. void resolve_led_message(const char* data, const char* topic, axk_mqtt_client_handle_t client) {
    
12. 
    
13.     //打开灯
    
14.     if (strcmp(topic, TOPIC_ON) == 0) {
    
15.         wb2_led_open(data);
    
16.     }
    
17.     //关闭灯
    
18.     else if (strcmp(TOPIC_OFF, topic) == 0) {
    
19.         wb2_led_close(data);
    
20.     }
    
21.     //发送灯的状态
    
22.     else if (strcmp(TOPIC_GET, topic) == 0) {
    
23.         axk_mqtt_client_publish(client, wb2_led_status(data), TOPIC_STATUS, 1, 0, 0);
    
24.     }
    
25. 
    
26. }

复制代码

mqtt_data_task中创建一个queue队列接收消息并处理

1. /**
   
2. * 循环队列接收mqtt数据
   
3. * @param  argv            
   
4. */
   
5. void mqtt_data_task(void* argv)
   
6. {
   
7.     //创建queue
   
8.     mqtt_data_queue = xQueueCreate(10, sizeof(mqtt_data_t));
   
9.     mqtt_data_t data;
   
10.     while (pdTRUE == xQueueReceive(mqtt_data_queue, &data, portMAX_DELAY))
    
11.     {
    
12.         //接收处理消息
    
13.         resolve_led_message(data.data, data.topic, client);
    
14.     }
    
15. }

复制代码

mqtt订阅接收消息

1. switch ((axk_mqtt_event_id_t)event_id) {
   
2.         case MQTT_EVENT_CONNECTED:
   
3.             //订阅ON事件
   
4.             msg_id = axk_mqtt_client_subscribe(client, TOPIC_ON, 0);
   
5.             blog_info("sent subscribe successful, msg_id=%d", msg_id);
   
6. 
   
7.             //订阅OFF事件
   
8.             msg_id = axk_mqtt_client_subscribe(client, TOPIC_OFF, 0);
   
9.             blog_info("sent subscribe successful, msg_id=%d", msg_id);
   
10. 
    
11.             //订阅GET事件
    
12.             msg_id = axk_mqtt_client_subscribe(client, TOPIC_GET, 0);
    
13.             blog_info("sent subscribe successful, msg_id=%d", msg_id);
    
14.             break;
    
15.         case MQTT_EVENT_DATA:
    
16.             //event->data后面没有'\0'分割，所以需要手动处理
    
17.             char* data = pvPortMalloc(sizeof(char) * event->data_len + 1);
    
18.             char* topic = pvPortMalloc(sizeof(char) * event->topic_len + 1);
    
19.             mqtt_data_t* mqdata = (mqtt_data_t*)pvPortMalloc(sizeof(mqtt_data_t));
    
20. 
    
21.             mqdata->data = data;
    
22.             mqdata->topic = topic;
    
23. 
    
24.             //复制data和topic
    
25.             memset(data, '\0', event->data_len + 1);
    
26.             memset(topic, '\0', event->topic_len + 1);
    
27.             memcpy(data, event->data, event->data_len);
    
28.             memcpy(topic, event->topic, event->topic_len);
    
29. 
    
30.             //向queue中发送消息
    
31.             xQueueSend(mqtt_data_queue, mqdata, 0);
    
32.             
    
33.             break;
    
34.     }

复制代码

• 通过mqttx客户端设置并连接到mqtt服务器
• 添加/status/led/wb2订阅
• 发送/on/led/wb2主题消息，内容为g，可以看到开发板绿灯亮了
• 发送/get/led/wb2主题消息，内容为g，可以看到客户端收到了数据1
• 发送/off/led/wb2主题消息，内容为g，可以看到开发板绿灯灭了
• 发送/get/led/wb2主题消息，内容为g，可以看到客户端收到了数据0
  

限于本人能力有限难免有错误或遗漏的地方请多多包涵
具体的代码上传到gitee代码仓库

<iframe src="https://player.bilibili.com/player.html?isOutside=true&aid=113133834213894&bvid=BV1YqtNecEwe&cid=25652893931&p=1" scrolling="no" border="0" frameborder="no" framespacing="0" allowfullscreen="true"></iframe>

视频地址

学习学习

辛苦啦~

学习学习

学习了

666

666

很好的帖子，正好最近需要用到MQTT