MQTT libmosquitto源码分析

概述

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,主要有如下：

 

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

 

数据包发送队列

向服务器发送数据包（packet）时，首先将数据包放到改发送队列中，并不真实发送数据，而是发送数据就绪信号，其他线程根据网络连接情况来处理发送请求；

该队列为单链表存储结构，每次有新数据包需要发送时，将新数据包插入到链表尾部；

真正发送数据时从链表头部开始发送数据包。

消息(message)

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

 

消息队列

主要处理收发消息时的缓存队列

注：

l   该队列与数据包队列没有直接关系；

l   数据包队列为网络层发送数据策略；

l   该队列为协议层处理逻辑；

 

发送消息队列（out_message）

发送消息队列，保存发送的消息或者收到的消息；加入该队列主要是因为整个消息流程未完成，还有后续交互需要处理；

 

1. 发送消息时，当Qos>0时，将消息加入当前队列；消息状态为mosq_ms_wait_for_pubrec或者mosq_ms_wait_for_puback;
2. 当收到相应PUBACK,PUBCOMP消息时，将其从该队列中移除

 

接收消息队列（in_message）

接收消息队列，保存收到的消息；加入该队列主要是因为整个消息流程未完成，还有后续交互需要处理

1. 接受到服务器端Qos==2的消息时会将其加入队列，消息状态为：mosq_ms_wait_for_pubrel；
2. 当收到相应PUBREL消息时，将其从该队列中移除

 

模块层次与关系

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

模块层次

 

模块关系

 

模块说明

mosquitto_internal.h

定义各种数据结构；

 

mosquitto：

外部调用接口

 

memory_mosq：

内存分配处理，可记录内存用量

_mosquitto_calloc：分配内存，相当于calloc

_mosquitto_packet_alloc：分配附加内存，根据remaining_length来分配payload内存

net_mosq：

l   网络基础操作，tcp创建，关闭等

l   向打包/解包数据，向_mosquitto_packet中写入/读取各种数据

 

_mosquitto_packet_queue

 

_mosquitto_packet_write//发送out_packet队列中所有package

_mosquitto_net_write

 

send_mosq：

主要实现发送请求逻辑（协议组包），实际命令请求实现组包；

 

有如下接口：

int _mosquitto_send_simple_command(struct mosquitto *mosq, uint8_t command);

remaining_length=0：For DISCONNECT, PINGREQ and PINGRESP

 

int _mosquitto_send_command_with_mid(struct mosquitto *mosq, uint8_t command, uint16_t mid, bool dup);

int _mosquitto_send_real_publish(struct mosquitto *mosq, uint16_t mid, const char *topic, uint32_t payloadlen, const void *payload, int qos, bool retain, bool dup);

 

int _mosquitto_send_connect(struct mosquitto *mosq, uint16_t keepalive, bool clean_session);

int _mosquitto_send_disconnect(struct mosquitto *mosq);

int _mosquitto_send_pingreq(struct mosquitto *mosq);

int _mosquitto_send_pingresp(struct mosquitto *mosq);

int _mosquitto_send_puback(struct mosquitto *mosq, uint16_t mid);

int _mosquitto_send_pubcomp(struct mosquitto *mosq, uint16_t mid);

int _mosquitto_send_publish(struct mosquitto *mosq, uint16_t mid, const char *topic, uint32_t payloadlen, const void *payload, int qos, bool retain, bool dup);

int _mosquitto_send_pubrec(struct mosquitto *mosq, uint16_t mid);

int _mosquitto_send_pubrel(struct mosquitto *mosq, uint16_t mid);

int _mosquitto_send_subscribe(struct mosquitto *mosq, int *mid, const char *topic, uint8_t topic_qos);

int _mosquitto_send_unsubscribe(struct mosquitto *mosq, int *mid, const char *topic);

 

 

send_client_mosq：

与send_mosq类似，主要实现客户端常用高频使用接口；其他接口在send_mosq中

 

有如下函数：

_mosquitto_send_connect

_mosquitto_send_disconnect

_mosquitto_send_subscribe

_mosquitto_send_unsubscribe

 

messages_mosq：

主要针对消息的实现(PUBLISH,PUBACK,PUBREL..)；

 

有如下函数：

void _mosquitto_message_cleanup_all(struct mosquitto *mosq);

void _mosquitto_message_cleanup(struct mosquitto_message_all **message);

int _mosquitto_message_delete(struct mosquitto *mosq, uint16_t mid, enum mosquitto_msg_direction dir);

void _mosquitto_message_queue(struct mosquitto *mosq, struct mosquitto_message_all *message, enum mosquitto_msg_direction dir);

void _mosquitto_messages_reconnect_reset(struct mosquitto *mosq);

int _mosquitto_message_remove(struct mosquitto *mosq, uint16_t mid, enum mosquitto_msg_direction dir, struct mosquitto_message_all **message);

void _mosquitto_message_retry_check(struct mosquitto *mosq);

int _mosquitto_message_out_update(struct mosquitto *mosq, uint16_t mid, enum mosquitto_msg_state state);

 

read_handle：

处理收到的数据包，根据数据包类型做相应处理。

有如下函数：

int _mosquitto_packet_handle(struct mosquitto *mosq);

int _mosquitto_handle_connack(struct mosquitto *mosq);

int _mosquitto_handle_pingreq(struct mosquitto *mosq);

int _mosquitto_handle_pingresp(struct mosquitto *mosq);

int _mosquitto_handle_publish(struct mosquitto *mosq);

int _mosquitto_handle_pubrec(struct mosquitto *mosq);

int _mosquitto_handle_pubrel(struct mosquitto_db *db, struct mosquitto *mosq);

int _mosquitto_handle_suback(struct mosquitto *mosq);

int _mosquitto_handle_unsuback(struct mosquitto *mosq);

 

协议交互流程

CONNECT

说明：

1. 客户端连接成功后，如果在一段时间不发CONNECT请求，服务器端主动断掉socket；
2. 服务器端如果判定CONNECT请求不合法，将返回非0错误码；
3. 如果服务区端一段时间没有发送CONNACK，客户端需要主动断掉socket；

 

PUBLISH

注：

l   上图为C->S客户端主动发送PUBLISH消息流程；

l   若S->C ；由服务端主动发PUBLISH，流程一样，发送方向相反；

SUBSCRIBE

 

PINGREQ & DISCONNECT

 

主要数据结构

客户端状态

该状态为用户连接成功并通讯CONNECT之后结果；

 

enum mosquitto_client_state {

    mosq_cs_new = 0,

    mosq_cs_connected = 1,

    mosq_cs_disconnecting = 2,// mosquitto_disconnect时设置

    mosq_cs_connect_async = 3,// mosquitto_connect_bind_async，异步线程来connect _mosquitto_thread_main（需要WITH_THREADING）

    mosq_cs_connect_pending = 4//没用到

};

 

消息状态

消息发送与接收流程用，关注mosq_ms_wait_for_xxxx状态，客户端处理此类消息；

消息处理流程可参考协议处理流程部分；

 

enum mosquitto_msg_state {。

    mosq_ms_invalid = 0,

    mosq_ms_publish_qos0 = 1,

    mosq_ms_publish_qos1 = 2,

    mosq_ms_wait_for_puback = 3,//Oos==1时，发送PUBLISH后等待PUBACK返回

    mosq_ms_publish_qos2 = 4,

    mosq_ms_wait_for_pubrec = 5, //Oos==2时，发送PUBLISH后，等待PUBREC返回

    mosq_ms_resend_pubrel = 6,

    mosq_ms_wait_for_pubrel = 7, //Oos==2时，发送PUBREC后等待PUBREL返回

    mosq_ms_resend_pubcomp = 8,

    mosq_ms_wait_for_pubcomp = 9, //Oos==2时，发送PUBREL后等待PUBCOMP返回

    mosq_ms_send_pubrec = 10,

    mosq_ms_queued = 11

};

 

数据包、数据包队列

发送数据（组包后）或者接受数据后（解包前）状态

struct _mosquitto_packet{

    uint8_t *payload;

    struct _mosquitto_packet *next;

    uint32_t remaining_mult;

    uint32_t remaining_length;

    uint32_t packet_length;

    uint32_t to_process;//发送进度，记录还未发送多少字节，缺省为packet_length

    uint32_t pos;//组包或者发送时用到，发送时记录发送到什么位置

    uint16_t mid;//消息id，当Qos==0 时回调on_publish时用

    uint8_t command;

    int8_t remaining_count;

};

 

消息

struct mosquitto_message{

    int mid;

    char *topic;

    void *payload;

    int payloadlen;

    int qos;

    bool retain;

};

 

 

消息队列

struct mosquitto_message_all{

    struct mosquitto_message_all *next;

    time_t timestamp;//时间，记录本地软件tick时间

    //enum mosquitto_msg_direction direction;

    enum mosquitto_msg_state state;

    bool dup;

    struct mosquitto_message msg;

};

 

 

会话相关属性（上下文）

struct mosquitto {

    mosq_sock_t sock;

    mosq_sock_t sockpairR, sockpairW;// socket管道通知：非阻塞模式时，通知用，在mosquitto_loop 调用发送,

    enum _mosquitto_protocol protocol;

    char *address;

    char *id;//客户端ID

    char *username;

    char *password;

    uint16_t keepalive;

    uint16_t last_mid;  //最后一个消息id，发消息后++

    enum mosquitto_client_state state;

    time_t last_msg_in;

    time_t last_msg_out;

    time_t ping_t;

    struct _mosquitto_packet in_packet;//接收数据包用

    struct _mosquitto_packet *current_out_packet;

    struct _mosquitto_packet *out_packet;//发送数据包队列

    struct mosquitto_message *will;

#ifdef WITH_TLS

    SSL *ssl;

    SSL_CTX *ssl_ctx;

    char *tls_cafile;

    char *tls_capath;

    char *tls_certfile;

    char *tls_keyfile;

    int (*tls_pw_callback)(char *buf, int size, int rwflag, void *userdata);

    char *tls_version;

    char *tls_ciphers;

    char *tls_psk;

    char *tls_psk_identity;

    int tls_cert_reqs;

    bool tls_insecure;

#endif

    bool want_write;

    bool want_connect;

#if defined(WITH_THREADING) && !defined(WITH_BROKER)

    pthread_mutex_t callback_mutex;

    pthread_mutex_t log_callback_mutex;

    pthread_mutex_t msgtime_mutex;

    pthread_mutex_t out_packet_mutex;

    pthread_mutex_t current_out_packet_mutex;

    pthread_mutex_t state_mutex;

    pthread_mutex_t in_message_mutex;

    pthread_mutex_t out_message_mutex;

    pthread_mutex_t mid_mutex;

    pthread_t thread_id;

#endif

    bool clean_session;

 

    void *userdata;

    bool in_callback;

    unsigned int message_retry;

    time_t last_retry_check;

    struct mosquitto_message_all *in_messages;//收到消息队列

    struct mosquitto_message_all *in_messages_last;

    struct mosquitto_message_all *out_messages;发送消息队列

    struct mosquitto_message_all *out_messages_last;

 

    void (*on_connect)(struct mosquitto *, void *userdata, int rc);

    void (*on_disconnect)(struct mosquitto *, void *userdata, int rc);

    void (*on_publish)(struct mosquitto *, void *userdata, int mid);

    void (*on_message)(struct mosquitto *, void *userdata, const struct mosquitto_message *message);

    void (*on_subscribe)(struct mosquitto *, void *userdata, int mid, int qos_count, const int *granted_qos);

    void (*on_unsubscribe)(struct mosquitto *, void *userdata, int mid);

    void (*on_log)(struct mosquitto *, void *userdata, int level, const char *str);

    //void (*on_error)();

    char *host;

    int port;

    int in_queue_len;  //收到消息队列长度

    int out_queue_len;//发送消息队列长度

    char *bind_address;

    unsigned int reconnect_delay;

    unsigned int reconnect_delay_max;

    bool reconnect_exponential_backoff;

    bool threaded;

    int inflight_messages; //对于Qos>0的消息，记录没有完成交互记录

    int max_inflight_messages;

};

主要处理流程

连接建立

 

重连流程

 

数据发送

 

接收数据

注：此流程需要在另外线程预先监听socket状态（调用select）；

接受数据有以下方法：

1. 用户启动线程或者timer，调用函数mosquitto_loop；
2. 直接调用函数mosquitto_loop_start，该函数会自动创建线程；在线程中调用mosquitto_loop_forever来处理数据收发；

注意：需要打开预编译WITH_THREADING；

消息发送

 

消息接收

 

消息重发及心跳

心跳说明：

1. 超过一个keeplive没有任何数据，需要主动发送PINGREQ
2. 发送PINGREQ后一个keeplive没收到PINGRESP，主动断掉客户端DISCONNECT;

代码优缺点

优点

l   完整实现MQTT协议，

l   跨平台实现，可以在LINUX,WINDOWS,MAC下运行

l   结构相对清晰；

l   功能相对完善；

缺点与不足

l   TCP连接建立与发送CONNECT耦合在一起，导致协议层与网络层耦合度太高；

l   处理发送，收取数据及收取数据后续处理在一个线程，可能会有性能瓶颈；

l   代码臃肿，可读性差；

l   不能判定wifi/3G/4G,移动设备有时候需要准确判定其状态来做相关数据操作；

l   用户消息数据需要自定义（文本，语音，文件，图片，超级连接，名片…）

l   没有是否压缩字段，对消息数据很多情况需要进行智能压缩；

l   有些协议命令没有回复(disconnect)，packet发送出去之后不知道是否能到达；

l   消息无时间戳，排序严格按照收发消息先后次序，有问题；

l   无法获取历史消息和离线消息，无批量获取消息类型；

l   消息重发永久性重发，重连机制不是很完善；

最后

以上就是碧蓝荷花为你收集整理的MQTT libmosquitto源码分析的全部内容，希望文章能够帮你解决MQTT libmosquitto源码分析所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。