在酒店做那个网站好,wordpress修改主题文件,外贸移动商城网站建设,商务网站开发实验报告MQTT背景应用
MQTT是机器对机器#xff08;M2M#xff09;/物联网#xff08;IoT#xff09;连接协议#xff0c;英文全名为“Message Queuing Telemetry Transport”#xff0c;“消息队列遥测传输”协议。它是专为受限设备和低带宽、高延迟或不可靠的网络而设计的M2M/物联网IoT连接协议英文全名为“Message Queuing Telemetry Transport”“消息队列遥测传输”协议。它是专为受限设备和低带宽、高延迟或不可靠的网络而设计的是一种基于发布/订阅publish/subscribe模式的“轻量级”通信协议该协议构建于TCP/IP协议之上由IBM在1999年发布。 名词释义
Publisher发布者Broker代理服务端Subscriber订阅者Topic发布/订阅的主题
流程概述上图中各类传感器的角色是发布者Publisher。譬如湿度传感器和温度传感器分别向接入的MQTT Broker中周期性发布两个主题名为“Moisture”湿度和“Temp”温度的主题伴随着这两个主题共同发布的还有湿度值和温度值被称为“消息”。几个客户端的角色是订阅者SubScriber如手机APP从Broker订阅了“Temp”主题便能在手机上获取到温度传感器Publish在Broker中的温度值。
发布者和订阅者的角色并非是固定的而是相对的。 发布者可以同时从Broker订阅主题同理订阅者也可以向Broker发布主题即发布者可以是订阅者订阅者也可以是发布者。
Broker可以是在线的云服务器也可以是本地搭建的局域网客户端。 按照需求实际上Broker自身也会保护一些订阅/发布主题的功能。
MQTT报文结构
任何通用/私有协议都是由事先规定好的、按某种规则约束的各种报文数据包组成的MQTT也不例外。 在MQTT协议中所有的数据包都是由最多三部分组成固定Header可变Header有效载荷 固定Header是必需的可变Header和有效载荷是非必需的。 因此理论上来说MQTT协议数据包的最小长度为2个字节造就了它身边占用的额外资源消耗最小化特色。
Fixed header 固定Header由至少两个字节组成如表2-2所示。 第一个字节的高4位描述了当前数据报文的类型低四位定义了与报文类型相关的标志位。 第二个及之后的至多4个字节代表着剩余数据的字节长度。 Remaining Length剩余长度表示当前报文剩余部分的字节数包括可变header和有效载荷的数据。
uMQTT的实现
uMQTT软件包是RT-Thread自主研发的基于MQTT3.1.1协议的客户端实现它提供了设备与MQTT Broker通讯的基本功能。
uMQTT软件包功能如下
实现基础的连接、订阅、发布功能具备多重心跳保活设备重连机制保证mqtt在线状态适应复杂情况支持Qos0QoS1QoS2三种发送信息质量支持多客户端使用用户端接口简便留有多种对外回调函数支持多种技术参数可配置易上手便于产品化开发功能强大资源占用率低支持功能可裁剪。
uMQTT的结构框架
uMQTT软件包主要用于在嵌入式设备上实现MQTT协议软件包的主要工作基于MQTT协议实现。 软件包实现过程中主要做了
根据MQTT3.1.1协议规定进行软件包数据协议的封包解包。传输层函数适配对接SALSocket Abstraction Layer层。uMQTT客户端层根据协议包层和传输层编写符合应用层的接口。实现基础连接、断连、订阅、取消订阅、发布消息等功能。支持qoS0/1/2三种发送信息质量。利用uplink timer定时器实现多重心跳保活机制和设备重连机制增加设备在线稳定性适应复杂情况。
uMQTT客户端
想要连接Broker嵌入式设备需要作为MQTT协议中的客户端来使用。 在uMQTT组件的umqtt.h文件中抽象出了初始化客户端用到的MQTT配置信息组成对应的数据结构体。
struct umqtt_info
{rt_size_t send_size,recv_size; //发送接收缓冲区大小const char *uri; //完整的URI包含URIURNconst char *client_id; //客户端IDconst char *lwt_topic; //遗嘱主题const char *lwt_message; //遗嘱消息const char *user_name; //用户名const char *password; //密码enum umqtt_qos lwt_qos; //遗嘱QoSumqtt_subsribe_cb lwt_cb; //遗嘱回调函数rt_uint8_t reconnect_max_num; //最大重连次数rt_uint32_t reconnect_interval; //最大重连时间间隔rt_uint8_t keepalive_max_num; /* 最大保活次数 */ rt_uint32_t keepalive_interval; /* 最大保活时间间隔 */ rt_uint32_t recv_time_ms; /* 接收超时时间 */ rt_uint32_t connect_time; /* 连接超时时间 */ rt_uint32_t send_timeout; /* 上行(发布/订阅/取消订阅)超时时间 */ rt_uint32_t thread_stack_size; /* 线程栈大小 */ rt_uint8_t thread_priority; /* 线程优先级 */
#ifdef PKG_UMQTT_TEST_SHORT_KEEPALIVE_TIMErt_uint16_t connect_keepalive_sec; /* 连接信息保活秒数 */
#endif
}这些配置信息一般在创建uMQTT客户端之前需要自行填写指定譬如Broker的“URI”、“用户名”或“密码”之类的关键信息。 其它的非关键信息如果没有指定那么会在创建客户端函数umqtt_create中调用umqtt_check_def_info函数来赋值为默认值。
static void umqtt_check_def_info(struct umqtt_info *info)
{if(info){if (info-send_size 0) { info-send_size PKG_UMQTT_INFO_DEF_SENDSIZE; }if (info-recv_size 0) { info-recv_size PKG_UMQTT_INFO_DEF_RECVSIZE; }if (info-reconnect_max_num 0) { info-reconnect_max_num PKG_UMQTT_INFO_DEF_RECONNECT_MAX_NUM; }if (info-reconnect_interval 0) { info-reconnect_interval PKG_UMQTT_INFO_DEF_RECONNECT_INTERVAL; }if (info-keepalive_max_num 0) { info-keepalive_max_num PKG_UMQTT_INFO_DEF_KEEPALIVE_MAX_NUM; }if (info-keepalive_interval 0) { info-keepalive_interval PKG_UMQTT_INFO_DEF_HEARTBEAT_INTERVAL; }if (info-connect_time 0) { info-connect_time PKG_UMQTT_INFO_DEF_CONNECT_TIMEOUT; }if (info-recv_time_ms 0) { info-recv_time_ms PKG_UMQTT_INFO_DEF_RECV_TIMEOUT_MS; }if (info-send_timeout 0) { info-send_timeout PKG_UMQTT_INFO_DEF_SEND_TIMEOUT; }if (info-thread_stack_size 0) { info-thread_stack_size PKG_UMQTT_INFO_DEF_THREAD_STACK_SIZE; }if (info-thread_priority 0) { info-thread_priority PKG_UMQTT_INFO_DEF_THREAD_PRIORITY; }}
}然而只有上述信息是无法运行起来一个MQTT客户端的。 故在umqtt.c中含有umqtt_info的umqtt_client结构体列出了初始化客户端用到的所有数据
struct umqtt_client
{int sock; /* 套接字 */ enum umqtt_client_state connect_state; /* mqtt客户端状态 */ struct umqtt_info mqtt_info; /* mqtt用户配置信息 */ rt_uint8_t reconnect_count; /* mqtt客户端重连计数 */ rt_uint8_t keepalive_count; /* mqtt保活计数 */ rt_uint32_t pingreq_last_tick; /* mqtt的PING请求上一次滴答值 */rt_uint32_t uplink_next_tick; /* 上行连接的下一次滴答值 */ rt_uint32_t uplink_last_tick; /* 上行连接的上一次滴答值 */ rt_uint32_t reconnect_next_tick; /* 客户端断开重连时的下一次滴答值 */ rt_uint32_t reconnect_last_tick; /* 客户端断开重连时的上一次滴答值 */ rt_uint8_t *send_buf, *recv_buf; /* 收发缓冲区指针 */ rt_size_t send_len, recv_len; /* 收发数据的长度 */ rt_uint16_t packet_id; /* mqtt报文标识符 */ rt_mutex_t lock_client; /* mqtt客户端互斥锁 */ rt_mq_t msg_queue; /* mqtt客户端消息队列 */ rt_timer_t uplink_timer; /* mqtt保活重连定时器 */ int sub_recv_list_len; /* 接收订阅信息的链表长度 */ rt_list_t sub_recv_list; /* 订阅消息的链表头 */ rt_list_t qos2_msg_list; /* QoS2的消息链表 */struct umqtt_pubrec_msg pubrec_msg[PKG_UMQTT_QOS2_QUE_MAX]; /* 发布收到消息数组(QoS2) */ umqtt_user_callback user_handler; /* 用户句柄 */ void *user_data; /* 用户数据 */ rt_thread_t task_handle; /* umqtt任务线程 */ rt_list_t list; /* umqtt链表头 */
};
部分成员的结构体和枚举类型定义可自行在umqtt.h文件中查看。 该结构体会在创建客户端函数umqtt_create中调用umqtt_check_def_info函数之后初始化
初始化遗嘱数据结构如果有的话为收发缓冲区申请内存创建互斥锁、消息队列和超时重连定时器超时回调实现重连保活初始化各链表创建umqtt_thread–mqtt数据收发线程返回mqtt_client结构体
当第6步返回的值不为空时即可调用umqtt_start函数来通过LWIP发送CONNECT报文连接Broker连接成功后便会启动umqtt_thread线程开启MQTT的通信。
LWIP
LWIP轻量级IP是一种开源、轻量且高效的Internet协议IP套件实现。 它主要设计用于嵌入式系统并经常用于具有有限资源的小型到中型设备如微控制器、物联网设备和实时操作系统。 LWIP是用C编程语言编写的旨在为资源受限的环境提供网络功能。
LWIP的主要特点和特性包括
小巧的占用空间LWIP旨在具有最小的内存和代码大小占用适用于资源受限的设备。支持常见网络协议LWIP支持标准的网络协议如IPv4、IPv6、TCP传输控制协议、UDP用户数据报协议、ICMPInternet控制消息协议等。可移植性LWIP非常可移植可以轻松适应各种嵌入式平台和操作系统。与实时操作系统RTOS集成LWIP通常与FreeRTOS等RTOS系统一起使用适用于实时和多任务应用。BSD套接字API使熟悉标准套接字编程的开发人员能够更轻松地使用该库。可扩展性LWIP设计为可扩展性允许开发人员根据需要为其特定应用程序添加自定义功能或协议。开源LWIP以开源许可证分发通常是MIT许可证允许免费使用。
uMQTT与LWIP
在umqtt_start函数中首先会将uMQTT客户端的状态置为UMQTT_CS_LINKING表示正在连接中接下来会调用umqtt_connect函数将本地客户端连接到Broker。
连接到Broker的过程分两步
创建套接字与Broker建立链路连接。发送CONNECT报文创建MQTT协议连接。
在umqtt_connect函数中通过调用umqtt_trans_connect函数来完成第一步
int umqtt_trans_connect(const char *uri, int *sock)
{int _ret 0;struct addrinfo *addr_res RT_NULL;*sock -1;// 域名解析_ret umqtt_resolve_uri(uri, addr_res);if((_ret 0) || (addr_res RT_NULL)){LOD_E(resolve uri err);_ret UMQTT_FAILED;goto exit;}//创建套接字if((*sock socket(addr_res-ai_family, SOCK_STREAM, UMQTT_SOCKET_PROTOCOL)) 0 ){LOG_E(create socket error!);_ret UMQTT_FAILED;goto exit;}//设置套接字工作在非阻塞模式下_ret ioctlsocket(*sock, FIONBIO, 0);if (_ret 0) {LOG_E( iocontrol socket error!);_ret UMQTT_FAILED;goto exit;}// 建立连接if( (_ret connect(*sock, addr_res-ai_addr, addr_res-ai_addrlen)) 0){LOG_E( connect err!);closesocket(*sock);*sock -1;_ret UMQTT_FAILED;goto exit;}exit:if (addr_res) {freeaddrinfo(addr_res);addr_res RT_NULL;}return _ret;
}这个函数就是uMQTT通过LWIP与Broker建立连接的核心函数。 该函数是通过SAL即套接字抽象层组件来调用相关接口访问LWIP的。 用到的部分SAL组件封装的函数getaddrinfo是在umqtt_resolve_uri函数中用来解析域名的
int getaddrinfo(const char *nodename,const char *servname,const struct addrinfo *hints,struct addrinfo **res)
{return sal_getaddrinfo(nodename, servname, hints, res);
}
---------------------------------------------------------------------------------------------
#define connect(s, name, namelen) sal_connect(s, name, namelen)
#define recvfrom(s, mem, len, flags, from, fromlen) sal_recvfrom(s, mem, len, flags, from, fromlen)
#define send(s, dataptr, size, flags) sal_sendto(s, dataptr, size, flags, NULL, NULL)
#define socket(domain, type, protocol) sal_socket(domain, type, protocol)
#define closesocket(s) sal_closesocket(s)
#define ioctlsocket(s, cmd, arg) sal_ioctlsocket(s, cmd, arg)
uMQTT发送组包
当uMQTT客户端与Broker成功建立链路层连接后就会立刻发送CONNECT报文建立MQTT的协议层连接。
uMQTT组件使用了巧妙的结构体共用体来管理所有的收发报文
union umqtt_pkgs_msg /* mqtt message packet type */
{struct umqtt_pkgs_connect connect; /* connect */ struct umqtt_pkgs_connack connack; /* connack */ struct umqtt_pkgs_publish publish; /* publish */ struct umqtt_pkgs_puback puback; /* puback */ struct umqtt_pkgs_pubrec pubrec; /* publish receive (QoS 2, step_1st) */ struct umqtt_pkgs_pubrel pubrel; /* publish release (QoS 2, step_2nd) */ struct umqtt_pkgs_pubcomp pubcomp; /* publish complete (QoS 2, step_3rd) */ struct umqtt_pkgs_subscribe subscribe; /* subscribe topic */ struct umqtt_pkgs_suback suback; /* subscribe ack */ struct umqtt_pkgs_unsubscribe unsubscribe; /* unsubscribe topic */ struct umqtt_pkgs_unsuback unsuback; /* unsubscribe ack */
};struct umqtt_msg
{union umqtt_pkgs_fix_header header; /* fix header */ rt_uint32_t msg_len; /* message length */ union umqtt_pkgs_msg msg; /* retain payload message */
};
union umqtt_pkgs_msg是一个联合体包含了多种不同类型的MQTT消息包。 每个成员都对应一种MQTT消息类型如connect、connack、publish等。 这个联合体的作用是可以容纳不同种类的MQTT消息包但在任何给定时刻只能包含一个有效的消息包。使得在处理MQTT消息时可以有效地共享内存空间减小内存占用。
struct umqtt_msg包含了三个成员
header是一个联合体’umqtt_pkgs_fix_header’表示MQTT消息的固定头部它包含了消息的控制标志和其他必要的元数据。msg_len是一个32位整数表示MQTT消息的长度。msg是一个联合体用于存储具体类型的MQTT消息包。通过header中的控制标志可以确定在’msg’中使用哪个成员。
通过umqtt_encode函数来调用不同的组包函数填充对应格式的结构体然后发送到Broker服务端。
/** * packaging the data according to the format** param type the input packaging type* param send_buf the output send buf, result of the package* param send_len the output send buffer length* param message the input message** return 0: failed or other error* 0: package data length*/
int umqtt_encode(enum umqtt_type type, rt_uint8_t *send_buf, size_t send_len, struct umqtt_msg *message)
{int _ret 0;switch (type){case UMQTT_TYPE_CONNECT:_ret umqtt_connect_encode(send_buf, send_len, (message-msg.connect));break;case UMQTT_TYPE_PUBLISH:_ret umqtt_publish_encode(send_buf, send_len, message-header.bits.dup, message-header.bits.qos, (message-msg.publish));break;case UMQTT_TYPE_PUBACK:_ret umqtt_puback_encode(send_buf, send_len, message-msg.puback.packet_id);break;case UMQTT_TYPE_PUBREC:// _ret umqtt_pubrec_encode();break;case UMQTT_TYPE_PUBREL:_ret umqtt_pubrel_encode(send_buf, send_len, message-header.bits.dup, message-msg.pubrel.packet_id);break;case UMQTT_TYPE_PUBCOMP:_ret umqtt_pubcomp_encode(send_buf, send_len, message-msg.pubcomp.packet_id);break;case UMQTT_TYPE_SUBSCRIBE:_ret umqtt_subscribe_encode(send_buf, send_len, (message-msg.subscribe));break;case UMQTT_TYPE_UNSUBSCRIBE:_ret umqtt_unsubscribe_encode(send_buf, send_len, (message-msg.unsubscribe));break;case UMQTT_TYPE_PINGREQ:_ret umqtt_pingreq_encode(send_buf, send_len);break;case UMQTT_TYPE_DISCONNECT:_ret umqtt_disconnect_encode(send_buf, send_len);break;default:break;}return _ret;
}
由于报文类型较多接下来仅以“CONNECT”报文可变header——“协议名称”、“协议等级”、“连接标志”、“保活间隔秒有效载荷——“客户端标识符”、“遗嘱主题”、“遗嘱消息”、“用户名”、“密码”为例”来简述uMQTT的组包过程 1.填充MQTT客户端的默认配置信息 2.调用umqtt_encode - umqtt_connect_encode编码函数组包
该函数首先调用MQTTSerialize_connectLength计算可变header和有效载荷的长度得到的len会被作为参数传递给umqtt_pkgs_len函数它的作用是计算固定header中的剩余长度字段的字节数并加上固定header第一个字节长度即1与buflen作比较判断该包数据的有效性。
if (umqtt_pkgs_len(len MQTTSerialize_connectLength(options)) buflen)希望得到的len长度就是固定header中的剩余长度值从而方便后面的组包过程而有效的报文长度buflen len 1 剩余长度字段的字节数。
组包过程完成之后会调用umqtt_trans_send函数通过LWIP将发送缓冲区数据发送到socket连接的Broker。
int umqtt_trans_send(int sock, const rt_uint8_t *send_buf, rt_uint32_t buf_len, int timeout)
{int _ret 0;rt_uint32_t offset 0U;while(offset buflen){_ret send(sock, send_buf offset, buf_len - offset, 0);if(_ret 0)return -errno;offset _ret;}return _ret;
}uMQTT接收解包
当uMQTT将CONNECT报文发送完成后就会调用umqtt_handle_readpacket函数完成CONNECT过程后该函数也会在umqtt_thread线程中被循环调用来接收数据读取Broker的回复对接收到的数据包进行解包处理
读Fixed header的第一个字节这里会调用umqtt_trans_recv函数读取socket数据作用就是从对应的sock中读取buf_len长度的数据到recv_buf。读Fixed header的Remaining length字段并解析剩余长度。读剩余数据——可变header有效载荷。解析数据包并根据不同报文类型做相应处理。UMQTT_TYPE_CONNACK调用set_uplink_recon_tick(client, UPLINK_NEXT_TICK)函数设置下一次重连滴答值调用set_connect_status(client, UMQTT_CS_LINKED)函数设置uMQTT客户端状态为已连接。
设置重连滴答值
设置下一次重连滴答值reconnect tick的目的是为了控制在失去连接后尝试重新建立连接的时间间隔。
网络部稳定性在物联网IoT或其他网络应用中设备可能会面临不稳定的网络条件如临时的断线或信号干扰。在这种情况下失去连接后立即进行重连可能会导致频繁的连接尝试造成网络资源浪费增加设备的电力消耗。避免过早的重连设置重连滴答值可以防止设备过早地尝试重新连接。重连的时间间隔可以根据设备和网络的特定需求进行调整以确保在网络稳定之前不会进行不必要的连接尝试。节约资源 预定重连滴答值有助于节约设备资源因为设备不必持续监视网络状态并进行连接尝试。它可以在预定的时间间隔后尝试连接以降低设备的功耗。逐渐增加重连频率 通常设置下一次重连滴答值的方式是逐渐增加重连频率。这意味着如果第一次重连尝试失败设备可能会等待一段较长的时间然后在下一次尝试时等待较短的时间。这种策略允许设备在网络稳定性恢复时更频繁地尝试连接同时避免了过于频繁的连接尝试。
总之设置下一次重连滴答值是为了在网络不稳定或断开连接的情况下以一种经济高效的方式管理设备的连接重试以提高设备的性能和资源利用率。
至此已经完成了CONNECT报文的收发过程下一步就是启动umqtt_thread线程调用umqtt_handle_readpacket函数来处理从Broker服务器端收到的数据报文。
