。

它是专为受限设备和低带宽、高延迟或不可靠的网络而设计的M2M/物联网IoT连接协议英文全名为“Message

Queuing

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

第一个字节的高4位描述了当前数据报文的类型低四位定义了与报文类型相关的标志位。

Remaining

Length剩余长度表示当前报文剩余部分的字节数包括可变header和有效载荷的数据。

uMQTT软件包是RT-Thread自主研发的基于MQTT3.1.1协议的客户端实现它提供了设备与MQTT

uMQTT软件包功能如下

实现基础的连接、订阅、发布功能具备多重心跳保活设备重连机制保证mqtt在线状态适应复杂情况支持Qos0QoS1QoS2三种发送信息质量支持多客户端使用用户端接口简便留有多种对外回调函数支持多种技术参数可配置易上手便于产品化开发功能强大资源占用率低支持功能可裁剪。

uMQTT软件包主要用于在嵌入式设备上实现MQTT协议软件包的主要工作基于MQTT协议实现。

根据MQTT3.1.1协议规定进行软件包数据协议的封包解包。

传输层函数适配对接SALSocket

Abstraction

Layer层。

uMQTT客户端层根据协议包层和传输层编写符合应用层的接口。

实现基础连接、断连、订阅、取消订阅、发布消息等功能。

支持qoS0/1/2三种发送信息质量。

利用uplink

timer定时器实现多重心跳保活机制和设备重连机制增加设备在线稳定性适应复杂情况。

uMQTT客户端

想要连接Broker嵌入式设备需要作为MQTT协议中的客户端来使用。

在uMQTT组件的umqtt.h文件中抽象出了初始化客户端用到的MQTT配置信息组成对应的数据结构体。

struct

PKG_UMQTT_TEST_SHORT_KEEPALIVE_TIMErt_uint16_t

连接信息保活秒数

}这些配置信息一般在创建uMQTT客户端之前需要自行填写指定譬如Broker的“URI”、“用户名”或“密码”之类的关键信息。

其它的非关键信息如果没有指定那么会在创建客户端函数umqtt_create中调用umqtt_check_def_info函数来赋值为默认值。

static

PKG_UMQTT_INFO_DEF_RECONNECT_MAX_NUM;

}if

PKG_UMQTT_INFO_DEF_RECONNECT_INTERVAL;

}if

PKG_UMQTT_INFO_DEF_KEEPALIVE_MAX_NUM;

}if

PKG_UMQTT_INFO_DEF_HEARTBEAT_INTERVAL;

}if

PKG_UMQTT_INFO_DEF_CONNECT_TIMEOUT;

}if

PKG_UMQTT_INFO_DEF_RECV_TIMEOUT_MS;

}if

PKG_UMQTT_INFO_DEF_SEND_TIMEOUT;

}if

PKG_UMQTT_INFO_DEF_THREAD_STACK_SIZE;

}if

PKG_UMQTT_INFO_DEF_THREAD_PRIORITY;

故在umqtt.c中含有umqtt_info的umqtt_client结构体列出了初始化客户端用到的所有数据

struct

pubrec_msg[PKG_UMQTT_QOS2_QUE_MAX];

发布收到消息数组(QoS2)

部分成员的结构体和枚举类型定义可自行在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支持标准的网络协议如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建立链路连接。

发送CONNECT报文创建MQTT协议连接。

在umqtt_connect函数中通过调用umqtt_trans_connect函数来完成第一步

int

{freeaddrinfo(addr_res);addr_res

RT_NULL;}return

}这个函数就是uMQTT通过LWIP与Broker建立连接的核心函数。

该函数是通过SAL即套接字抽象层组件来调用相关接口访问LWIP的。

用到的部分SAL组件封装的函数getaddrinfo是在umqtt_resolve_uri函数中用来解析域名的

int

---------------------------------------------------------------------------------------------

#define

当uMQTT客户端与Broker成功建立链路层连接后就会立刻发送CONNECT报文建立MQTT的协议层连接。

union

umqtt_pkgs_msg是一个联合体包含了多种不同类型的MQTT消息包。

每个成员都对应一种MQTT消息类型如connect、connack、publish等。

这个联合体的作用是可以容纳不同种类的MQTT消息包但在任何给定时刻只能包含一个有效的消息包。

使得在处理MQTT消息时可以有效地共享内存空间减小内存占用。

struct

header是一个联合体’umqtt_pkgs_fix_header’表示MQTT消息的固定头部它包含了消息的控制标志和其他必要的元数据。

msg_len是一个32位整数表示MQTT消息的长度。

msg是一个联合体用于存储具体类型的MQTT消息包。

通过header中的控制标志可以确定在’msg’中使用哪个成员。

通过umqtt_encode函数来调用不同的组包函数填充对应格式的结构体然后发送到Broker服务端。

/**

(message-msg.connect));break;case

send_len,

(message-msg.publish));break;case

send_len,

message-msg.puback.packet_id);break;case

UMQTT_TYPE_PUBREC://

umqtt_pubrec_encode();break;case

send_len,

message-msg.pubrel.packet_id);break;case

send_len,

message-msg.pubcomp.packet_id);break;case

umqtt_subscribe_encode(send_buf,

send_len,

(message-msg.subscribe));break;case

umqtt_unsubscribe_encode(send_buf,

send_len,

(message-msg.unsubscribe));break;case

send_len);break;case

umqtt_disconnect_encode(send_buf,

send_len);break;default:break;}return

_ret;

由于报文类型较多接下来仅以“CONNECT”报文可变header——“协议名称”、“协议等级”、“连接标志”、“保活间隔秒有效载荷——“客户端标识符”、“遗嘱主题”、“遗嘱消息”、“用户名”、“密码”为例”来简述uMQTT的组包过程

2.调用umqtt_encode

该函数首先调用MQTTSerialize_connectLength计算可变header和有效载荷的长度得到的len会被作为参数传递给umqtt_pkgs_len函数它的作用是计算固定header中的剩余长度字段的字节数并加上固定header第一个字节长度即1与buflen作比较判断该包数据的有效性。

(umqtt_pkgs_len(len

MQTTSerialize_connectLength(options))

buflen)希望得到的len长度就是固定header中的剩余长度值从而方便后面的组包过程而有效的报文长度buflen

len

组包过程完成之后会调用umqtt_trans_send函数通过LWIP将发送缓冲区数据发送到socket连接的Broker。

int

当uMQTT将CONNECT报文发送完成后就会调用umqtt_handle_readpacket函数完成CONNECT过程后该函数也会在umqtt_thread线程中被循环调用来接收数据读取Broker的回复对接收到的数据包进行解包处理

读Fixed

header的第一个字节这里会调用umqtt_trans_recv函数读取socket数据作用就是从对应的sock中读取buf_len长度的数据到recv_buf。

读Fixed

length字段并解析剩余长度。

读剩余数据——可变header有效载荷。

解析数据包并根据不同报文类型做相应处理。

UMQTT_TYPE_CONNACK调用set_uplink_recon_tick(client,

UPLINK_NEXT_TICK)函数设置下一次重连滴答值调用set_connect_status(client,

UMQTT_CS_LINKED)函数设置uMQTT客户端状态为已连接。

设置重连滴答值

tick的目的是为了控制在失去连接后尝试重新建立连接的时间间隔。

网络部稳定性在物联网IoT或其他网络应用中设备可能会面临不稳定的网络条件如临时的断线或信号干扰。

在这种情况下失去连接后立即进行重连可能会导致频繁的连接尝试造成网络资源浪费增加设备的电力消耗。

避免过早的重连设置重连滴答值可以防止设备过早地尝试重新连接。

重连的时间间隔可以根据设备和网络的特定需求进行调整以确保在网络稳定之前不会进行不必要的连接尝试。

节约资源

预定重连滴答值有助于节约设备资源因为设备不必持续监视网络状态并进行连接尝试。

它可以在预定的时间间隔后尝试连接以降低设备的功耗。

逐渐增加重连频率

通常设置下一次重连滴答值的方式是逐渐增加重连频率。

这意味着如果第一次重连尝试失败设备可能会等待一段较长的时间然后在下一次尝试时等待较短的时间。

这种策略允许设备在网络稳定性恢复时更频繁地尝试连接同时避免了过于频繁的连接尝试。

总之设置下一次重连滴答值是为了在网络不稳定或断开连接的情况下以一种经济高效的方式管理设备的连接重试以提高设备的性能和资源利用率。

至此已经完成了CONNECT报文的收发过程下一步就是启动umqtt_thread线程调用umqtt_handle_readpacket函数来处理从Broker服务器端收到的数据报文。