libMosquitto 源码分析之一：概述

Posted 2021-04-05 思源互联科技

概述

libmosquitto 作为 mosquitto 开源代码的一部分,主要用来实现 MQTT 协议 栈和数据包通讯功能。

本文主要 述 libmosquitto 部分代码架构,实现原理,部分重要代码解析; 另外还有针对该代码库的不足和问题分析。

阅读条件
阅读此文,需要了解 MQTT 协议结构和部分实现。

MQTT 简述

MQTT(Message Queuing Telemetry Transport,消息队列遥测传输)是 IBM开发的一个即时通讯协议,有可能成为物联网的重要组成部分。该协议支持所有 平台,几乎可以把所有联网物品和外部连接起来,被用来当做传感器和致动器(比 如通过 Twitter 让房屋联网)的通信协议。

MQTT协议是为大量计算能力有限,且工作在低带宽、不可靠的网络的远程 传感器和控制设备通讯而设计的协议,它具有以下主要的几项特性:1、使用发布/订阅消息模式, 供一对多的消息发布,解除应用程序耦合;2、对负载内容屏蔽的消息传输;

3、使用 TCP/IP 供网络连接;

4、有三种消息发布服务质量:

“至多一次”,消息发布完全依赖底层 TCP/IP 网络。会发生消息丢 失或重复。这一级别可用于如下情况,环境传感器数据,丢失一次 读记录无所谓,因为不久后还会有第二次发送。“至少一次”,确保消息到达,但消息重复可能会发生。“只有一次”,确保消息到达一次。这一级别可用于如下情况,在计 费系统中,消息重复或丢失会导致不正确的结果。

5、小型传输,开销很小(固定长度的头部是 2 字节),协议交换最小化,以 降低网络流量;
6、使用 Last Will 和 Testament 特性通知有关各方客户端异常中断的机制;

MQTT 协议参考:http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html

http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.pdf

名词约束

数据包(packet)客户端发送给服务器端(或者服务器发给客户端)指令的整体数据,在程序中用

结构体 _mosquitto_packet 来表示;

数据包 ID(Packet Identifier field)

该ID为16位整形数,在Variant Header中;有些包类型需要带Packet Identifier field,主要有如下:

Control Packet	Packet Identifier field
CONNECT	NO
CONNACK	NO
PUBLISH	YES (If QoS > 0)
PUBACK	YES
PUBREC	YES

PUBREL	YES
PUBCOMP	YES
SUBSCRIBE	YES
SUBACK	YES
UNSUBSCRI BE	YES
UNSUBACK	YES
PINGREQ	NO
PINGRESP	NO
DISCONNEC T	NO

PS:没搞明白为啥不有 5 个没有 ID,难道是为了节省 2 字节流量?

数据包发送队列

向服务器发送数据包(packet)时,首先将数据包放到改发送队列中,并不真实 发送数据,而是发送数据就绪信号,其他线程根据网络连接情况来处理发送请求; 该队列为单链表存储结构,每次有新数据包需要发送时,将新数据包插入到链表 尾部;

真正发送数据时从链表头部开始发送数据包。消息(message)

专指用户消息(包含 PUBLISH,PUBACK,PUBREC,PUBREL,PUBCOMP),在程序中 用结构体 mosquitto_message_all 来表示;

消息队列

主要处理收发消息时的缓存队列
注:
 该队列与数据包队列没有直接关系;  数据包队列为网络层发送数据策略;  该队列为协议层处理逻辑;

发送消息队列(out_message)发送消息队列,保存发送的消息或者收到的消息;加入该队列主要是因为整个消

息流程未完成,还有后续交互需要处理;

1. 
   

2. 
   

   发送消息时,当 Qos>0 时,将消息加入当前队列;消息状态为mosq_ms_wait_for_pubrec 或者 mosq_ms_wait_for_puback;

3. 
   

4. 
   

   当收到相应PUBACK,PUBCOMP消息时,将其从该队列中移除

5. 
   

接收消息队列(in_message)接收消息队列,保存收到的消息;加入该队列主要是因为整个消息流程未完成,

还有后续交互需要处理

1. 接受到服务器端 Qos==2 的消息时会将其加入队列,消息状态为:mosq_ms_wait_for_pubrel;

2. 当收到相应PUBREL消息时,将其从该队列中移除

模块层次与关系

该代码主要有对外接口,协议流程实现,消息队列,MQTT 协议栈实现(组包, 解包),发送数据队列,TCP/IP 网络连接 6 部分。

模块层次

/

MQTT

TCP/IP

模块关系