mosquitto源码分析报告.docx
《mosquitto源码分析报告.docx》由会员分享，可在线阅读，更多相关《mosquitto源码分析报告.docx（25页珍藏版）》请在冰点文库上搜索。
mosquitto源码分析报告
mosquitto源码分析
本文由逍遥子撰写，转发请标注原址：
一、 Mosquitto简介
mosquitto是一款实现了消息推送协议MQTTv3.1的开源消息代理软件，提供轻量级的，支持可发布/可订阅的的消息推送模式，使设备对设备之间的短消息通信变得简单，例如现在应用广泛的低功耗传感器，手机、嵌入式计算机、微型控制器等移动设备。
Mosquitto采用出版/订阅的模式实现MQTT协议，这种设计模式将通信终端之间的关系统一到服务程序中进展管理，可极大减轻客户端的开发和维护工作。
1.1、 mqtt协议简介
MQTT〔MessageQueuingTelemetryTransport，消息队列遥测传输〕是IBM开发的一个即时通讯协议，有可能成为物联网的重要组成局部。
在某些应用场合中，可通过该协议维持与客户端的长连接。
关于mqtt协议更详细的介绍，请参考其官方：
1.2、 出版/订阅模式简介
出版/订阅模式定义了如何向一个节点发布和订阅消息，这些节点被称作主题（topic）。
主题可以被认为是消息的传输中介，发布者（publisher）发布消息到主题，订阅者（subscriber）从主题订阅消息。
这种模式使得消息订阅者和消息发布者保持互相独立，不需要接触即可保证消息的传送。
Tcp协议中，tcp连接只提供一对一的可靠传输，例如：
主机A与B进展通信，如此发起tcp连接的一端只需要知道对方的ip地址和端口号即可，如如下图1-1所示：
图1-1 一对一通信
每一个tcp连接都是由下面的五个元素确定：
<源ip地址，源端口号，目的ip地址，目的端口号，通信协议>
在实际程序的开发过程中一条连接建立之后，它可能需要在一段时间内一直被通信双方所保持，以备下次数据传输使用。
另外，通信的终端数目可能是多个，这就需要每个终端都要维持它所有的通信关系，如如下图1-2所示
图1-2 多对多通信
此时，每个参与通信的客户端所需维持的连接数量将非常庞大，这非常不利于程序的开发和实现。
出版/订阅模式即是一种解决这种问题的方法，它通过增加一个中间层的方式，让中间层来维护这种多对多的关系，这个中间层通常称之为服务器，如如下图1-3所示：
图1-3增加中间层的多对多通信
通过增加中间层服务器，每个客户端都只需要维护自己同服务器之间的连接即可，而客户端之间的关系如此交由中间服务器来维护，这种设计模式将复杂的通信关系维护工作从客户端剥离出来，非常方便客户端的开发和维护。
Mosquito程序即是通过这种方式进展工作，在mosquitto程序内部，将客户端之间的关系通过一棵订阅树来维持。
1.3、 Mosquito
Mosquito源码目录结构介绍。
|----client
|----examples
|----mysql_log
|----temperature_conversion
|----installer
|----lib
|----cpp
|----jsws
|----python
|----logo
|----man
|----po
|----libmosquitto
|----mosquitto
|----mosquitto_pub
|----mosquitto_sub
|----mosquitto-tls
|----mqtt
|----misc
|----currentcost
|----gnome-panel
|----security
|----service
|----monit
|----svscan
|----upstart
|----src
|----db_dump
|----test
|----broker
|----c
|----lib
|----c
|----cpp
|----python
|----python3
|----ssl
|----demoCA
|----rootCA
|----signingCA
所需关注的目录有/mosquitto-1.2/src、/mosquitto-1.2/lib、/mosquitto-1.2/client三个目录，其中src和lib目录下主要放置mosquitto的实现代码以与局部底层与网络相关的操作，client目录主要为两个客户端程序的实现源码。
Mosquito的源码与其相关文档可从其官方获取，其官方为：
mosquitto客户端和服务器运行命令
[1]发布者客户端运行命令示例：
./mosquitto_pub-h192.168.6.243-p1883-t"111"-m"thisisjason.hou"-u111-P111[2]订阅者客户端运行命令示例:
./mosquitto_sub-h192.168.6.243-i111-p1883-t111-k60-d-c-uhjx-Phjx[3]mosquitto服务器端运行命令示例：
./mosquitto
二、 Mosquito的数据结构
1〕 structmosquito
结构体structmosquito主要用于保存一个客户端连接的所有信息，例如用户名、密码、用户ID、向该客户端发送的消息等，其定义为：
structmosquitto{
intsock;
char*address;
char*id;
char*username;
char*password;
uint16_tkeepalive;
time_tlast_msg_in;
time_tlast_msg_out;
structmosquitto_client_msg*msgs;
…
}
上面列举了该结构体局部重要成员，其中sock表示mosquitto服务器程序与该客户端连接通信所用的socket描述符；address表示该客户端的IP地址；id是该客户端登陆mosquitto程序时所提供的ID值，该值与其他的客户端不能重复；成员username和password用于记录客户端登陆时所提供的用户名和密码；keepalive是该客户端需在此时间内向mosquitto服务器程序发送一条ping/pong消息。
参数last_msg_in和last_msg_out用于记录上次收发消息的时间；参数structmosquitto_client_msg*msgs用于暂时存储发往该context的消息。
2〕 structmosquitto_db
结构体structmosquitto_db是mosquitto对所有内部数据的统一管理结构，可以认为是其内部的一个内存数据库。
它保存了所有的客户端，所有客户端的订阅关系等等，其定义形式为：
structmosquitto_db{
dbid_tlast_db_id;
struct_mosquitto_subhiersubs;
structmosquitto**contexts;
struct_clientid_index_hash*clientid_index_hash;
intcontext_count;
structmosquitto_msg_store*msg_store;
intmsg_store_count;
structmqtt3_config*config;
intsubscription_count;
……
};
上述结构体声明中，结构体成员struct_mosquitto_subhiersubs保存了订阅树的总树根，mosquitto中对所有的topic都在该订阅树中维护，客户端的订阅关系也在该订阅树中维护；结构体成员structmosquitto**contexts可理解为一个用于存放所有客户端变量〔类型为structmosquitto〕地址的数组，mosquitto程序中，所有的客户端都在此数组中保存；成员intcontext_count用于保存数组contexts的大小，该值也是当前mosquitto程序中维持的所有客户端的数目；成员结构体struct_clientid_index_hash*clientid_index_hash用于保存一个hash表，该hash表可通过客户端的ID快速找到该客户端在数组contexts中的索引；结构体成员structmqtt3_config*config用于保存mosquitto的所有配置信息；
3〕struct_mosquitto_subhier
数据结构struct_mosquitto_subhier是用于保存订阅树的节点〔包括叶子节点和中间节点〕，mosquitto中对订阅树采用孩子-兄弟链表法的方式进展存储，该存储方式主要借助与数据结构struct_mosquitto_subhier来完成，该数据结构的定义为：
struct_mosquitto_subhier{
struct_mosquitto_subhier*children;
struct_mosquitto_subhier*next;
struct_mosquitto_subleaf*subs;
char*topic;
structmosquitto_msg_store*retained;
};
成员说明：
children：
该成员指针指向同结构的第一个孩子节点；
next：
该成员指针指向该节点的下一个兄弟节点；
subs：
该成员指向订阅列表；
topic：
该节点对应的topic片段；
4〕 struct_mosquitto_subleaf
在mosquitto程序中，对某一topic的所有订阅者被组织成一个订阅列表，该订阅列表是一个双向链表，链表的每个节点都保存有一个订阅者，该链表的节点即是：
struct_mosquitto_subleaf，定义形式为：
struct_mosquitto_subleaf{
struct_mosquitto_subleaf*prev;
struct_mosquitto_subleaf*next;
structmosquitto*context;
intqos;
};
其中成员structmosquitto*context表示一个订阅客户端，prev和next表示其前一个节点和后一个节点。
5〕structmqtt3_config
结构体structmqtt3_config用于保存mosquitto的所有配置信息，mosquitto程序在启动时将初始化该结构体并从配置文件中读取配置信息保存于该结构体变量内。
三、 Mosquito的核心功能分析
3.1、订阅树
Mosquitto通过订阅树的方式来管理所有的topic以与客户端的订阅关系，它首先将所有的topic按照/分割并组织成一棵树结构，从根节点到树中的每个节点即组成该节点所对应的一个topic，每个topic都保存一个订阅列表，该订阅列表中保存了所有订阅当前topic的客户端信息。
例如有如下订阅关系：
客户端a1,a2,a3订阅了topic：
A1/B1/C1
客户端b1,b2订阅了topic：
A2/B2/C2
客户端c1,c2订阅了topic：
A1/B1/C3
客户端d1订阅了topic：
A2/B3
如此在mosquitto程序中需要先将topic按照/进展分割，然后将分割后的topic片段组织成订阅树，上述订阅树的示意图为图3-1：
图3-1 订阅树示意图
Mosquitto程序在实现中根据topic消息的性质将订阅树分为两颗子树：
业务子树和系统子树；mosquitto程序中将topic分为两种类型来处理：
系统topic和业务topic，前者主要用于发布和维护mosquitto内部的系统消息，后者的topic是用户订阅的业务topic，做这种区分的原因是因为这两种的类型的topic性质和实现方式上有许多差异，这种差异主要表现在以下4点：
1〕生存周期不同，系统topic无论是否有用户订阅都会存在与订阅树中，而业务topic必须有客户端订阅才能存在〔除非其消息字段retain设置为1〕；
2〕创建方式不同，系统topic在消息发布时进展创建，业务topic即可以在订阅时创建也可以在消息发布时创建〔此时需要该消息retain字段设置为1〕；
3〕消息保存方式不同，但凡发布到系统topic的消息都会被保存下来，业务消息将直接挂到订阅列表的各context的消息队列中，如果没有连接订阅或未设置其retain字段，消息将不会被保存下来；
消息的retain字段是否被设置在函数mqtt3_handle_publish进展检查，在该函数中有如下代码：
retain=（header&0x01）;
该代码可获取消息头部的第一个bit位，在mqtt3.1协议中，该为用于表示消息的类型是否为retain。
订阅树在程序中的采用孩子—兄弟链表法来表示。
其主要涉与的数据结构是：
struct_mosquitto_subhier
struct_mosquitto_subleaf
3.1.1、订阅树的搭建
1、创建订阅树
mosquitto程序启动时将创建订阅树，该过程将创建三个节点：
订阅树总根节点、业务子树根节点和系统子树根节点，这两个子树根节点作为订阅树总根节点的两个子节点，其中订阅树总根节点和业务子树的根节点中topic成员的值为空字符串，而系统子树根节点中保存的值为“$SYS〞，如图3-2所示。
图3-2 订阅树的创建
订阅树的创建主要在文件database.c中mqtt3_db_open函数里实现。
订阅树中节点的数据结构为struct_mosquitto_subhier，订阅树采用“孩子—兄弟〞链表法保存。
2、搭建订阅树
在订阅树中，系统子树与业务子树的搭建过程不一样，系统子树是在系统消息发布时创建，而业务子树创建过程即可以在消息发布时创建也可以在客户端订阅时才创建。
1）系统子树搭建过程
Mosquitto中，系统子树在发布系统消息时，自动检测topic片段是否存在，如果不存在如此在系统topic上创建节点以搭建订阅树。
例如，mosquitto程序启动时，将首先向系统topic：
$SYS/broker/version发送一条版本消息“〞，此时订阅树的系统子树只有一个根节点，如图3.2所示，其搭建过程如下：
（1）将topic按照〞/〞分成topic片段，系统Topic：
$SYS/broker/version将被分割为$SYS、broker、version三局部。
（2）根据第一个topic片段“$SYS〞遍历订阅树的子节点找到系统子树的根节点。
（3）根据topic下一个片段“broker〞查找系统子树；此时系统子树中不存在topic片段“broker〞的节点，如此为订阅树产生一个节点，其数据结构为：
struct_mosquitto_subhier。
此时订阅树由图3-2变为图3-3所示：
图3-3添加broker节点之后的订阅树
上述过程在函数mqtt3_db_messages_queue中调用函数_sub_add来完成。
（4）依此处理topic剩下的片段，在系统子树中添加topic片段“version〞，该过程通过递归调用函数_sub_add来完成。
添加完“version〞片段之后的订阅树如图3-4所示。
图3-4添加version之后的订阅树
系统子树搭建过程中，所用到的函数调用关系如如下图3-5所示
图3-5系统子树搭建的函数调用关系
2）业务子树搭建过程
业务子树的搭建分为两种类型：
一种在订阅时创建；一种是在消息发布时创建，这种方式与系统topic的创建过程一样，因此下面的内容将主要描述前一种方式。
在mosquitto程序运行期间，收到一条客户端的订阅请求后将调用函数mqtt3_sub_add将该客户端挂到对应的业务子树节点的订阅列表中，此时，如果所订阅树中不存在客户端所订阅的topic，如此会自动为之添加相应的节点，此过程即为订阅树的业务子树搭建过程。
例如，在mosquitto程序启动时〔此时订阅树如图3-2所示〕客户端订阅了topic：
year/month/events,业务子树的搭建过程为：
（1）将topic按照〞/〞分成topic片段，topic：
year/month/event将被分割成：
year、month、events三个topic片段；此过程在函数mqtt3_sub_add中完成。
（2）遍历订阅树的第一级子节点，以查找业务子树的根节点，找到业务子树的根节点之后进展后续的处理。
此过程在函数mqtt3_sub_add中完成。
（3）依此处理三级topic片段：
year、month、events，此过程与系统topic的添加过程相似，如果订阅树中不某个存在topic片段，如此为订阅树添加此节点，添加完成之后的订阅树如图3-6：
本条功能在函数_sub_add中完成。
图3-6添加业务topic之后的订阅树
业务子树搭建过程中，所用到的函数调用关系如如下图3-7所示
图3-7业务子树搭建的函数调用关系
3.1.2、使用订阅树发布消息
在Mosquito程序中，消息发送过程主要通过遍历订阅树来完成，具体为：
递归遍历订阅树找到指定的订阅列表，并将消息挂到订阅列表中的每个contextg的消息队列中，如果消息的retain字段被设置为1，如此mosquitto还需要保存此消息，以备新订阅的客户端可以立即收到上次发送的消息；另外，发往系统topic的消息也会被mosquitto保存起来。
上述消息发送过程主要通过函数mqtt3_db_messages_queue来完成。
另外，根据mqtt协议，mosquitto还对#和+两个通配符提供支持，通配符#可放在topic的第一级之外的其他topic片段中，用以表示匹配所有后续topic片段，例如：
a/#可以匹配a/b、a/b/c、a等，但是#需要独立为一个分级才能起作用，即a/b#c/d这种情形不被支持。
通配符用“+〞用以匹配一级topic，例如a/+/c可以匹配a/b/c、a/d/c但是不能匹配a/d/f。
另外消息的qos等级不同，mosquitto对消息的发送过程也不一样。
例如，在如图3-8所示的订阅树中，以qos为0时发布消息为例，消息发布时通过mqtt3_db_messages_queue完成。
图3-8订阅树的遍历
向topic：
year/month/events1发送消息，其发送过程为：
1）将topic按照〞/〞分成topic片段：
year、month、event1s三个topic片段；该topic分片过程在函数mqtt3_db_messages_queue中完成。
2）遍历订阅树根的子节点，并根据第一级topic分片的类型是否为“$SYS〞选择业务子树还是系统子树进展递归遍历。
同时还需判断消息中retain标识位是否设置为1，如果是，如此需将该topic的所有后续分片参加到订阅树中，经过该添加之后，订阅树中就存在了该topic，这一添加topic片段的过程在函数_sub_add中完成。
3）递归遍历子树，每次选取一个topic分片，将之与当前的节点的所有子节点进展比拟，以找到匹配的节点，然后继续递归匹配下去，递归处理过程主要由函数_sub_search完成；
4）当匹配完所有的topic分片或者遇到topic分片为#时，进展如下处理：
5）当消息的retain字段被设置为1之后，需要将该消息持久化，即保存下来，以备有新订阅者订阅该topic的消息时可以收到上一条消息。
6）遍历当前订阅树节点的订阅列表，将消息挂载所有订阅者的消息队列中，此过程通过函数mqtt3_db_message_insert完成。
mosquitto消息发送过程中所调用函数的关系如如下图3-9所示：
图3-9 消息发布函数调用关系图
3.1.3、从订阅树中删除订阅客户端
在Mosquitto程序中，客户端可以向mosquitto服务器发送取消对某个topic的订阅请求，服务器收到请求之后，将从订阅树中删除该客户端，该过程需要遍历订阅树以找该topic在订阅树中的位置，进而获得到订阅该topic的订阅列表，从而将其从订阅树中删掉。
该过程与发送消息类似，首先将topic进展分片，然后根据topic片段遍历订阅树找到该topic的订阅列表；然后遍历订阅列表，将该客户端从订阅列表中删除。
取消订阅的函数调用关系如图3-10所示。
图3-10取消订阅函数调用关系
3.1.4、订阅树机制的优缺点分析
Mosquito程序采用订阅树形式维护客户端之间的订阅与发布消息，这种方式优点是逻辑清晰，便于开发和维护。
缺点是其遍历过程效率较低。
同时，程序中存在很多对订阅树的遍历过程：
订阅、发布消息、取消订阅等，在客户端数量增加时，该功能对效率的影响将更为明显。
因此，在mosquitto的实际应用中很难支持5万以上的客户端，尤其在客户端网络状态不好时，其断开重练操作将非常频繁，这样也造成大量对订阅树的遍历操作，从而严重影响mosquitto的效率。
3.2、mosquitto的消息收发机制
Mosquito最主要的功能就是对消息的正确接收、维护和转发，因此其消息的接、收功能非常重要。
Mosquitto的消息收发主要通过poll机制来完成。
3.2.1、poll机制简介
Poll机制是一种I/O多路转接〔I/Omultiplexing〕技术，这种技术先构造一个有关描述符的表，然后调用一个函数，知道这些描述符中的一个已准备就绪好进展I/O时，该函数才返回，该函数返回之后它告诉进程那些描述符已经准备好进展I/O）。
其工作过程为：
〔1〕首先构造一个pollfd结构数组，每个数组元素指定一个描述符编号以与对其所关心的状态。
〔2〕为每一个需要监听状态的设备描述符创建一个pollfd结构体，并将之放入pollfd结构数组中。
〔3〕调用函数poll监控pollfd结构数组中每个结构体的状态。
Poll函数的原型如下：
intpoll（structpollfdfdarray[],nfds_tnfds,inttimeout）
其中参数fdarray[]就是pollfd结构体数组，nfds是数组的元素数，timeout是等待超时限定，如果超过此时间还没有描述符准备好I/O，就返回；pollfd结构体的定义为：
structpollfd{
intfd;
shortevent;
shortrevent;
};
其中，intfd是文件描述符，在mosquitto的中主要指socket描述符
升级会员 