RT-Thread中使用Mqtt

环境：

开发板：Panduola（stm32L475）
KEIL5 开发环境
rtthread 4.0.3内核
使用ENV 配置Rtt

MQTT

1.MQTT介绍
​ 客户端 Client

  • 使用MQTT的程序或设备。客户端总是通过网络连接到服务端。
  • 它可以发布应用消息给其它相关的客户端。
  • 订阅以请求接受相关的应用消息。
  • 取消订阅以移除接受应用消息的请求。
  • 从服务端断开连接。服务端 Server一个程序或设备，作为发送消息的客户端和请求订阅的客户端之间的中介。

服务端

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

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

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

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

订阅 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数据包中，表示客户端收到的具体内容。

操作：

添加AT指令部分，使用ESP8266连接网络：

RT-Thread online packages  --->IoT - internet of things  --->[*] AT DEVICE: RT-Thread AT component porting or samples for different device[*]   Espressif ESP8266  --->

在这里插入图片描述

1. 添加RTT组件包：

MQTT：
在这里插入图片描述
CJSON：
在这里插入图片描述
AHT10:使用AHT10还需要打开I2c的驱动部分，使用旧版本1.0的AHT驱动，可以避免使用Sensor的框架
在这里插入图片描述
启用I2c总线：
在这里插入图片描述

2.使用scons --target=mdk5生成mdk工程

在这里插入图片描述

3.使用MQtt客户端：

修改连接参数：

#define MQTT_URI “tcp://192.168.1.110:1883”
#define MQTT_USERNAME “panduola”
#define MQTT_PASSWORD “panduola”
#define MQTT_SUBTOPIC “/test/topic2”
#define MQTT_PUBTOPIC “/test/topic1”

可在rtconfig.h中修改wifi连接参数：
在这里插入图片描述

  1. 创建一个客户端：
    static MQTTClient client;

  2. 初始化客户端：


/* 创建与配置 mqtt 客户端 */
static void mq_start(void)
{/* 初始 condata 参数 */MQTTPacket_connectData condata = MQTTPacket_connectData_initializer;static char cid[20] = {0};static int is_started = 0;if (is_started){return;}/* 配置 MQTT 文本参数 */{client.isconnected = 0;client.uri = MQTT_URI;/* 生成随机客户端 ID */rt_snprintf(cid, sizeof(cid), "rtthread%d", rt_tick_get());// rt_snprintf(sup_pub_topic, sizeof(sup_pub_topic), "%s%s", MQTT_PUBTOPIC, cid);/* 配置连接参数 */memcpy(sup_pub_topic, MQTT_SUBTOPIC, sizeof(MQTT_SUBTOPIC));memcpy(&client.condata, &condata, sizeof(condata));client.condata.clientID.cstring = cid;client.condata.keepAliveInterval = 60;client.condata.cleansession = 1;client.condata.username.cstring = MQTT_USERNAME;client.condata.password.cstring = MQTT_PASSWORD;/* 配置 mqtt 参数 */client.condata.willFlag = 0;client.condata.will.qos = 1;client.condata.will.retained = 0;client.condata.will.topicName.cstring = sup_pub_topic;client.buf_size = client.readbuf_size = 1024;client.buf = malloc(client.buf_size);client.readbuf = malloc(client.readbuf_size);if (!(client.buf && client.readbuf)){LOG_E("no memory for MQTT client buffer!");goto _exit;}/* 设置事件回调 */client.connect_callback = mqtt_connect_callback;client.online_callback = mqtt_online_callback;client.offline_callback = mqtt_offline_callback;/* 设置要订阅的 topic 和 topic 对应的回调函数 */client.messageHandlers[0].topicFilter = MQTT_SUBTOPIC;client.messageHandlers[0].callback = mqtt_sub_callback;client.messageHandlers[0].qos = QOS1;/* 设置默认订阅回调函数 */client.defaultMessageHandler = mqtt_sub_default_callback;}/* 启动 MQTT 客户端 */LOG_D("Start mqtt client and subscribe topic:%s", sup_pub_topic);paho_mqtt_start(&client);is_started = 1;_exit:return;
}
  1. 设置收到信息后的回调：
static void mqtt_sub_callback(MQTTClient *c, MessageData *msg_data)
{printf("Receive message ,Topic: %.*s,payload:\n", msg_data->topicName->lenstring.len,msg_data->topicName->lenstring.data);// 解析JSON消息cJSON *json_obj = cJSON_Parse((const char *)msg_data->message->payload);if (json_obj == NULL){printf("Failed to parse JSON message\n");return;}cJSON *object = RT_NULL;object = cJSON_GetObjectItem(json_obj, "location");if (object != NULL){printf("Location: %s\n", object->valuestring);}object = cJSON_GetObjectItem(json_obj, "led");if (object != NULL){if(object->type == cJSON_True)printf("led: ture\n");elseprintf("led: false\n");}cJSON_Delete(json_obj);return;
}
  1. 发布消息：

/* MQTT 消息发布函数 */
static void mq_publish(const char *send_str)
{MQTTMessage message;const char *msg_str = send_str;const char *topic = MQTT_PUBTOPIC;message.qos = QOS1;message.retained = 0;message.payload = (void *)msg_str;message.payloadlen = strlen(message.payload);MQTTPublish(&client, topic, &message);return;
}

常用的cjson函数：

void cJSON_Delete(cJSON *c)

删除 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

常用的mqtt的API：

/*** This function send an MQTT subscribe packet and wait for suback before returning.** @param client the pointer of MQTT context structure* @param qos MQTT Qos type, only support QOS1* @param topic topic filter name* @param callback the pointer of subscribe topic receive data function** @return the error code, 0 on start successfully.*//*订阅主题，最后一个形参是void (*subscribe_cb)(MQTTClient *client, MessageData *data);类型的函数指针*/
int paho_mqtt_subscribe(MQTTClient *client, enum QoS qos, const char *topic, subscribe_cb callback);

/*** This function publish message to specified mqtt topic.* @note it will be discarded, recommend to use "paho_mqtt_publish"** @param c the pointer of MQTT context structure* @param topicFilter topic filter name* @param message the pointer of MQTTMessage structure** @return the error code, 0 on subscribe successfully.*/
int MQTTPublish(MQTTClient *client, const char *topic, MQTTMessage *message);

source:

#include <rtthread.h>
#include <rtdevice.h>
#include <board.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>
#include "paho_mqtt.h"
// #include "wifi_config.h"#include "aht10.h"
#define DBG_TAG "main"
#define DBG_LVL DBG_LOG
#include <rtdbg.h>#include <cJSON.h>/*** MQTT URI farmat:* domain mode* tcp://iot.eclipse.org:1883** ipv4 mode* tcp://192.168.10.1:1883* ssl://192.168.10.1:1884** ipv6 mode* tcp://[fe80::20c:29ff:fe9a:a07e]:1883* ssl://[fe80::20c:29ff:fe9a:a07e]:1884*/
#define MQTT_URI "tcp://192.168.1.110:1883"
#define MQTT_USERNAME "panduola"
#define MQTT_PASSWORD "panduola"
#define MQTT_SUBTOPIC "/test/topic2"
#define MQTT_PUBTOPIC "/test/topic1"#define LED_PIN GET_PIN(E, 8)/* define MQTT client context */
static MQTTClient client;
static void mq_start(void);
static void mq_publish(const char *send_str);char sup_pub_topic[48] = MQTT_PUBTOPIC;
char sup_sub_topic[48] = MQTT_SUBTOPIC;static void mqtt_sub_callback(MQTTClient *c, MessageData *msg_data)
{printf("Receive message ,Topic: %.*s,payload:\n", msg_data->topicName->lenstring.len, msg_data->topicName->lenstring.data);// 解析JSON消息cJSON *json_obj = cJSON_Parse((const char *)msg_data->message->payload);if (json_obj == NULL){printf("Failed to parse JSON message\n");return;}cJSON *object = RT_NULL;object = cJSON_GetObjectItem(json_obj, "location");if (object != NULL){printf("Location: %s\n", object->valuestring);}object = cJSON_GetObjectItem(json_obj, "led");if (object != NULL){if (object->type == cJSON_True){printf("led: ture\n");rt_pin_write(LED_PIN, PIN_LOW);}else{printf("led: false\n");rt_pin_write(LED_PIN, PIN_HIGH);}}cJSON_Delete(json_obj);return;
}static void mqtt_sub_default_callback(MQTTClient *c, MessageData *msg_data)
{*((char *)msg_data->message->payload + msg_data->message->payloadlen) = '\0';LOG_D("mqtt sub default callback: %.*s %.*s",msg_data->topicName->lenstring.len,msg_data->topicName->lenstring.data,msg_data->message->payloadlen,(char *)msg_data->message->payload);return;
}static void mqtt_connect_callback(MQTTClient *c)
{LOG_I("Start to connect mqtt server");
}static void mqtt_online_callback(MQTTClient *c)
{LOG_D("Connect mqtt server success");LOG_D("Publish message: Hello,RT-Thread! to topic: %s", sup_pub_topic);mq_publish("Hello,RT-Thread!");
}static void mqtt_offline_callback(MQTTClient *c)
{LOG_I("Disconnect from mqtt server");
}/* 创建与配置 mqtt 客户端 */
static void mq_start(void)
{/* 初始 condata 参数 */MQTTPacket_connectData condata = MQTTPacket_connectData_initializer;static char cid[20] = {0};static int is_started = 0;if (is_started){return;}/* 配置 MQTT 文本参数 */{client.isconnected = 0;client.uri = MQTT_URI;/* 生成随机客户端 ID */rt_snprintf(cid, sizeof(cid), "rtthread%d", rt_tick_get());// rt_snprintf(sup_pub_topic, sizeof(sup_pub_topic), "%s%s", MQTT_PUBTOPIC, cid);/* 配置连接参数 */memcpy(sup_pub_topic, MQTT_SUBTOPIC, sizeof(MQTT_SUBTOPIC));memcpy(&client.condata, &condata, sizeof(condata));client.condata.clientID.cstring = cid;client.condata.keepAliveInterval = 60;client.condata.cleansession = 1;client.condata.username.cstring = MQTT_USERNAME;client.condata.password.cstring = MQTT_PASSWORD;/* 配置 mqtt 参数 */client.condata.willFlag = 0;client.condata.will.qos = 1;client.condata.will.retained = 0;client.condata.will.topicName.cstring = sup_pub_topic;client.buf_size = client.readbuf_size = 1024;client.buf = malloc(client.buf_size);client.readbuf = malloc(client.readbuf_size);if (!(client.buf && client.readbuf)){LOG_E("no memory for MQTT client buffer!");goto _exit;}/* 设置事件回调 */client.connect_callback = mqtt_connect_callback;client.online_callback = mqtt_online_callback;client.offline_callback = mqtt_offline_callback;/* 设置要订阅的 topic 和 topic 对应的回调函数 */client.messageHandlers[0].topicFilter = MQTT_SUBTOPIC;client.messageHandlers[0].callback = mqtt_sub_callback;client.messageHandlers[0].qos = QOS1;/* 设置默认订阅回调函数 */client.defaultMessageHandler = mqtt_sub_default_callback;}/* 启动 MQTT 客户端 */LOG_D("Start mqtt client and subscribe topic:%s", sup_pub_topic);paho_mqtt_start(&client);is_started = 1;_exit:return;
}/* MQTT 消息发布函数 */
static void mq_publish(const char *send_str)
{MQTTMessage message;const char *msg_str = send_str;const char *topic = MQTT_PUBTOPIC;message.qos = QOS1;message.retained = 0;message.payload = (void *)msg_str;message.payloadlen = strlen(message.payload);MQTTPublish(&client, topic, &message);return;
}rt_thread_t TH_Get_HU;
rt_thread_t Publish_value;
float humidity, temperature;void get_humi_temp(void *parameter)
{aht10_device_t dev;const char *i2c_bus_name = "i2c2";int count = 0;rt_thread_mdelay(2000);dev = aht10_init(i2c_bus_name);if (dev == RT_NULL){LOG_E(" The sensor initializes failure");}while (count++ < 100){humidity = aht10_read_humidity(dev);// LOG_D("humidity   : %d.%d %%", (int)humidity, (int)(humidity * 10) % 10);temperature = aht10_read_temperature(dev);// LOG_D("temperature: %d.%d", (int)temperature, (int)(temperature * 10) % 10);rt_thread_mdelay(1000);}
}void Publish_Date(void *parameter)
{char send_str[128];while (1){// sprintf(send_str, "{\"temperature\":%d.%d,\"humidity\":%d.%d}", (int)temperature, (int)(temperature * 10) % 10, (int)humidity, (int)(humidity * 10) % 10);sprintf(send_str, "{\"location\":\"10#A401\",\"led\":true,\"environment\":{\"temperature\":%d.%d,\"humidity\":%d.%d}}", (int)temperature, (int)(temperature * 10) % 10, (int)humidity, (int)(humidity * 10) % 10);mq_publish(send_str);rt_thread_mdelay(1000);}
}int main(void)
{mq_start();TH_Get_HU = rt_thread_create("get_humi_temp", get_humi_temp, RT_NULL, 1024, 20, 10);Publish_value = rt_thread_create("Publish_value", Publish_Date, RT_NULL, 1024, 20, 10);rt_pin_mode(LED_PIN, PIN_MODE_OUTPUT);rt_thread_startup(TH_Get_HU);rt_thread_mdelay(1000);rt_thread_startup(Publish_value);
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/704332.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

音视频捕捉技术：LCC382 SDI采集卡深度解析

音视频捕捉技术：LCC382 SDI采集卡深度解析

在日新月异的多媒体时代&#xff0c;高质量的音视频采集已成为众多领域不可或缺的一环。为此&#xff0c;灵卡科技精心打造了LCC382 —— 一款集高效性、灵活性与前沿技术于一身的SDI输入与环出、HDMI输出音视频采集卡&#xff0c;旨在满足从专业直播、视频会议到医疗影像、安防…
阅读更多...
Java微信小程序订阅消息提醒的实现与对接

Java微信小程序订阅消息提醒的实现与对接

文章目录 一、准备工作1. 注册微信小程序&#xff0c;并开通订阅消息功能。2. 获取小程序的AppID和AppSecret。3. 在微信小程序管理后台&#xff0c;设置提醒模板&#xff0c;并获取模板ID。4. 小程序端需要获取用户订阅允许提醒的权限&#xff08;1&#xff09;引导用户触发订…
阅读更多...
Google Ads被暂停的原因，如何防范？

Google Ads被暂停的原因，如何防范？

跨境出海业务少不了需要做Google Ads推广业务&#xff1b;其中让投手们闻风丧胆的消息就是帐户被暂停。当 Google 检测到任何违反其政策且可能损害用户在线体验的行为时&#xff0c;就会发生这种情况。那么如何在做广告推广的同时&#xff0c;保证账号不被封禁呢&#xff1f;看…
阅读更多...
前馈神经网络FNN、多层感知机MLP和反向传播推导

前馈神经网络FNN、多层感知机MLP和反向传播推导

目录 一、前馈神经网络FNN 激活函数的使用 二、多层感知机MLP MLP的典型结构 多层感知机MLP的特点 和前馈神经网络FNN的区别 三、传播推导 1、前向传播(Forward propagation) &#xff08;1&#xff09;输入层到隐藏层 &#xff08;2&#xff09;隐藏层到输出层 2、…
阅读更多...
Vue3知识总结-4

Vue3知识总结-4

Vue3知识总结-4 文章目录 Vue3知识总结-4插槽Slots渲染作用域默认内容具名插槽插槽中的数据传递具名插槽传递数据 组件声明周期声明周期示意图 组件生命周期的应用动态组件组件保持存活组件被卸载 异步组件依赖注入 插槽Slots 在某些场景中&#xff0c;可能想要为子组件传递一…
阅读更多...
easyx快速入门1

easyx快速入门1

1.基本说明 EasyX 是针对 C 的图形库&#xff0c;可以帮助 C/C 初学者快速上手图形和游戏编程。 比如&#xff0c;可以基于 EasyX 图形库很快的用几何图形画一个房子&#xff0c;或者一辆移动的小车&#xff0c;可以编写俄罗斯方块、贪吃蛇、黑白棋等小游戏&#xff0c;可以练…
阅读更多...
SDL系列（四）—— 事件机制

SDL系列（四）—— 事件机制

事件循环 大多数多媒体程序依靠 事件系统 来处理输入。 SDL 为处理输入事件提供了灵活的 API 。 本质上&#xff0c; SDL 将来自设备&#xff08;如键盘&#xff0c;鼠标或控制器&#xff09;的输入记录为 事件 &#xff0c;将它们存储在 “ 事件队列 ”中。 您可以将此…
阅读更多...
硬盘坏道如何检测和修复?

硬盘坏道如何检测和修复?

硬盘是我们储存数据的重要设备&#xff0c;然而在使用过程中&#xff0c;我们可能会遇到一些困扰&#xff0c;比如硬盘出现坏道的问题。那么&#xff0c;什么是坏道呢&#xff1f;硬盘出现坏道会对我们的性能和数据安全产生影响吗&#xff1f;如何去检测和修复这些坏道呢&#…
阅读更多...
【STM32-MX_GPIO_Init分析】

【STM32-MX_GPIO_Init分析】

MX_GPIO_Init分析源码如下&#xff1a; __HAL_RCC_GPIOE_CLK_ENABLE源码如下&#xff1a; #define RCC ((RCC_TypeDef *) RCC_BASE) #define RCC_BASE (AHB1PERIPH_BASE 0x3800UL) #define AHB1PERIPH_BASE (PERIPH_BASE 0x00020000U…
阅读更多...
失业焦虑如何缓解心情？流静冥想

失业焦虑如何缓解心情？流静冥想

失业焦虑如何缓解心情&#xff1f;人生旅途&#xff0c;失业犹如山重水复&#xff0c;焦虑似迷雾遮望眼。古语云&#xff1a;“山不厌高&#xff0c;海不厌深。”心之向往&#xff0c;冥想便是那披荆斩棘之斧&#xff0c;如何带你走出困境&#xff1f; “静以修身”&#xff0c…
阅读更多...
你是学会了还是学废了：Elasticsearch 7 集群拷贝到其它环境如何重置密码

你是学会了还是学废了：Elasticsearch 7 集群拷贝到其它环境如何重置密码

欢迎您关注我的公众号【尚雷的驿站】 公众号&#xff1a;尚雷的驿站 CSDN &#xff1a;https://blog.csdn.net/shlei5580 墨天轮&#xff1a;https://www.modb.pro/u/2436 PGFans&#xff1a;https://www.pgfans.cn/user/home?userId4159 前言 本文描述了将生产ES集群打包拷贝…
阅读更多...
实时美颜技术揭秘：直播美颜SDK的架构与优化

实时美颜技术揭秘：直播美颜SDK的架构与优化

当下&#xff0c;美颜技术成为直播平台吸引用户和提升用户体验的重要手段。本文将揭秘实时美颜技术&#xff0c;详细介绍直播美颜SDK的架构&#xff0c;并探讨其优化方法。 一、实时美颜技术概述 1、发展历程 随着图像处理算法的进步&#xff0c;逐渐发展到实时视频处理领域…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023