Skip to content

Latest commit

 

History

History
567 lines (400 loc) · 20.3 KB

mqtt-design.md

File metadata and controls

567 lines (400 loc) · 20.3 KB

mqttclient设计与实现方式

设计思想

  • 整体采用分层式设计,代码实现采用异步设计方式,降低耦合。

  • 消息的处理使用回调的方式处理:用户指定订阅的主题与指定消息的处理函数

  • 不依赖外部任何文件。

API接口

mqttclient拥有非常简洁的API接口,参数都是非常简单的。

API 说明 示例
mqtt_lease() 申请一个mqtt客户端 mqtt_client_t *client = mqtt_lease();
mqtt_release() 释放已申请的mqtt客户端 mqtt_release(client);
mqtt_connect() 与服务器建立连接 mqtt_connect(client);
mqtt_disconnect() 与服务器断开连接 mqtt_disconnect(client);
mqtt_subscribe() 订阅主题,参数:主题名字、服务质量、指定当收到主题数据时的处理函数。 mqtt_subscribe(client, "topic", QOS0, sub_topic_handle);
mqtt_unsubscribe() 取消订阅指定主题,参数:主题名字 mqtt_unsubscribe(client,
mqtt_publish() 向指定主题发布数据,参数:主题名字,mqtt_message_t类型的数据内容 mqtt_publish(client, "topic", &msg);
mqtt_list_subscribe_topic() 列出客户端已订阅的主题 mqtt_list_subscribe_topic(client);
mqtt_set_host() 设置要连接的MQTT服务器地址,参数:域名 / 点分十进制的IP地址 mqtt_set_host(client, "www.jiejie01.top");
mqtt_set_port() 设置要连接的MQTT服务器端口号 mqtt_set_port(client, "1883");
mqtt_set_ca() 设置要连接的MQTT服务器ca证书 mqtt_set_ca(client, "ca ...");
mqtt_set_user_name() 设置客户端的用户名 mqtt_set_user_name(client, "any");
mqtt_set_password() 设置客户端的密码 mqtt_set_password(client, "any");
mqtt_set_client_id() 设置客户端的ID mqtt_set_client_id(client, "any");
mqtt_set_clean_session() 设置在断开连接后清除会话 mqtt_set_clean_session(client, 1);
mqtt_set_keep_alive_interval() 设置心跳间隔时间(秒) mqtt_set_keep_alive_interval(client, 50);
mqtt_set_cmd_timeout() 设置命令超时时间(毫秒),主要用于socket读写超时 mqtt_set_cmd_timeout(client, 5000);
mqtt_set_reconnect_try_duration() 设置重连的时间间隔(毫秒) mqtt_set_reconnect_try_duration(client, 1024);
mqtt_set_read_buf_size() 设置读数据缓冲区的大小 mqtt_set_read_buf_size(client, 1024);
mqtt_set_write_buf_size() 设置写数据缓冲区的大小 mqtt_set_write_buf_size(client, 1024);
mqtt_set_will_flag() 设置遗嘱标记 mqtt_set_will_flag(client, 1);
mqtt_set_will_options() 设置遗嘱的配置信息,指定遗嘱主题,服务质量,遗嘱保留标记,遗嘱内容 mqtt_set_will_options(client, "will_topic", QOS0, 0, "will_message");
mqtt_set_version() 设置MQTT协议的版本,默认值是4,MQTT版本为3.1.1 mqtt_set_version(client, 4);
mqtt_set_reconnect_handler() 设置重连时的回调函数 mqtt_set_reconnect_handler(client, reconnect_handler);
mqtt_set_interceptor_handler() 设置拦截器处理函数,将所有底层数据上报给用户 mqtt_set_interceptor_handler(client, interceptor_handler);

MQTT客户端的核心结构

mqtt_client_t 结构

typedef struct mqtt_client {
    char                        *mqtt_client_id;
    char                        *mqtt_user_name;
    char                        *mqtt_password;
    char                        *mqtt_read_buf;
    char                        *mqtt_write_buf;
    char                        *mqtt_host;
    char                        *mqtt_port;
    char                        *mqtt_ca;
    void                        *mqtt_reconnect_data;
    uint16_t                    mqtt_keep_alive_interval;
    uint16_t                    mqtt_packet_id;
    uint32_t                    mqtt_will_flag          : 1;
    uint32_t                    mqtt_clean_session      : 1;
    uint32_t                    mqtt_ping_outstanding   : 2;
    uint32_t                    mqtt_version            : 4;
    uint32_t                    mqtt_ack_handler_number : 24;
    uint32_t                    mqtt_cmd_timeout;
    uint32_t                    mqtt_read_buf_size;
    uint32_t                    mqtt_write_buf_size;
    uint32_t                    mqtt_reconnect_try_duration;
    size_t                      mqtt_client_id_len;
    size_t                      mqtt_user_name_len;
    size_t                      mqtt_password_len;
    mqtt_will_options_t         *mqtt_will_options;
    client_state_t              mqtt_client_state;
    platform_mutex_t            mqtt_write_lock;
    platform_mutex_t            mqtt_global_lock;
    mqtt_list_t                 mqtt_msg_handler_list;
    mqtt_list_t                 mqtt_ack_handler_list;
    network_t                   *mqtt_network;
    platform_thread_t           *mqtt_thread;
    platform_timer_t            mqtt_reconnect_timer;
    platform_timer_t            mqtt_last_sent;
    platform_timer_t            mqtt_last_received;
    reconnect_handler_t         mqtt_reconnect_handler;
    interceptor_handler_t       mqtt_interceptor_handler;
} mqtt_client_t;

该结构主要维护以下内容:

  1. MQTT客户端连接服务器必要的参数,如客户端ID mqtt_client_id、用户名mqtt_user_name、密码mqtt_password以及客户端ID长度mqtt_client_id_len、用户名长度mqtt_user_name_len、密码长度mqtt_password_len等。

  2. 读写数据缓冲区mqtt_read_buf、mqtt_write_buf及其大小的配置mqtt_read_buf_size、mqtt_write_buf_size

  3. 服务器相关的配置信息,如服务器地址mqtt_host、服务器端口号mqtt_port、服务器CA证书mqtt_ca

  4. 一些MQTT客户端的配置信息:如心跳时间间隔mqtt_keep_alive_interval、MQTT报文标识符mqtt_packet_id、遗嘱标记位mqtt_will_flag、清除会话标记mqtt_clean_session、MQTT协议版本mqtt_version、等待应答列表的最大记录个数mqtt_ack_handler_number等。

  5. 一些其他的配置,如遗嘱消息相关的配置mqtt_will_options、客户端的状态mqtt_client_state、写缓冲区的互斥锁mqtt_write_lock、全局的互斥锁mqtt_global_lock等。

  6. 命令超时时间mqtt_cmd_timeout(主要是读写阻塞时间、等待响应的时间、重连等待时间等)。

  7. 维护消息处理列表mqtt_msg_handler_list,这是mqtt协议必须实现的内容,所有来自服务器的publish报文都会被处理(前提是订阅了对应的消息,或者设置了拦截器)。

  8. 维护ack链表mqtt_ack_handler_list,这是异步实现的核心,所有等待响应的报文都会被挂载到这个链表上。

  9. 维护一个网络组件层mqtt_network,它可以自动选择数据通道。

  10. 维护一个内部线程mqtt_thread,所有来自服务器的mqtt包都会在内部线程这里被处理!

  11. 两个定时器,分别是掉线重连定时器与保活定时器mqtt_reconnect_timer、mqtt_last_sent、mqtt_last_received

  12. 设置掉线重连后告知应用层的回调函数mqtt_reconnect_handler与参数mqtt_reconnect_data

  13. 设置底层的拦截器的回调函数mqtt_interceptor_handler,将所有底层数据上报给应用层。

mqttclient实现

以下是整个框架的实现方式,方便大家更容易理解mqttclient的代码与设计思想,让大家能够修改源码与使用,还可以提交pr或者issues,开源的世界期待各位大神的参与,感谢!

除此之外以下代码的记录机制超时处理机制是非常好的编程思想,大家有兴趣一定要看源代码!

申请一个mqtt客户端

mqtt_client_t *mqtt_lease(void);
  1. 这个函数的内部通过动态申请内存的方式申请了一个MQTT客户端结构mqtt_client_t

  2. 调用**_mqtt_init()**函数将其内部的进行了默认的初始化,如申请网络组件的内存空间、初始化相关的互斥锁、链表等。

释放已申请的mqtt客户端

mqtt_release()

回收MQTT客户端结构mqtt_client_t的内存空间、网络组件的内存空间、与服务器断开连接。

设置MQTT客户端的信息

通过宏定义去统一设置MQTT客户端结构mqtt_client_t的信息,定义如下:

#define MQTT_CLIENT_SET_DEFINE(name, type, res)         \
    type mqtt_set_##name(mqtt_client_t *c, type t) {    \
        MQTT_ROBUSTNESS_CHECK((c), res);                \
        c->mqtt_##name = t;                             \
        return c->mqtt_##name;                          \
    }

由编译器预处理得到相关的函数:mqtt_set_xxx()

MQTT_CLIENT_SET_DEFINE(client_id, char*, NULL)
MQTT_CLIENT_SET_DEFINE(user_name, char*, NULL)
MQTT_CLIENT_SET_DEFINE(password, char*, NULL)
MQTT_CLIENT_SET_DEFINE(host, char*, NULL)
MQTT_CLIENT_SET_DEFINE(port, char*, NULL)
MQTT_CLIENT_SET_DEFINE(ca, char*, NULL)
MQTT_CLIENT_SET_DEFINE(reconnect_data, void*, NULL)
MQTT_CLIENT_SET_DEFINE(keep_alive_interval, uint16_t, 0)
MQTT_CLIENT_SET_DEFINE(will_flag, uint32_t, 0)
MQTT_CLIENT_SET_DEFINE(clean_session, uint32_t, 0)
MQTT_CLIENT_SET_DEFINE(version, uint32_t, 0)
MQTT_CLIENT_SET_DEFINE(cmd_timeout, uint32_t, 0)
MQTT_CLIENT_SET_DEFINE(read_buf_size, uint32_t, 0)
MQTT_CLIENT_SET_DEFINE(write_buf_size, uint32_t, 0)
MQTT_CLIENT_SET_DEFINE(reconnect_try_duration, uint32_t, 0)
MQTT_CLIENT_SET_DEFINE(reconnect_handler, reconnect_handler_t, NULL)
MQTT_CLIENT_SET_DEFINE(interceptor_handler, interceptor_handler_t, NULL)

连接服务器

int mqtt_connect(mqtt_client_t* c);

参数只有 mqtt_client_t 类型的指针,连接服务器则是使用非异步的方式设计,因为必须等待连接上服务器才能进行下一步操作。

过程如下:

  1. 调用底层的连接函数连接上服务器:

    network_connect(c->network);
  2. 序列化mqttCONNECT报文并且发送。

    MQTTSerialize_connect(c->write_buf, c->write_buf_size, &connect_data)
    mqtt_send_packet(c, len, &connect_timer)
  3. 等待来自服务器的CONNACK报文

    mqtt_wait_packet(c, CONNACK, &connect_timer)
  4. 连接成功后创建一个内部线程mqtt_yield_thread,并在合适的时候启动它:

    /* connect success, and need init mqtt thread */
    c->mqtt_thread= platform_thread_init("mqtt_yield_thread", mqtt_yield_thread, c, MQTT_THREAD_STACK_SIZE, MQTT_THREAD_PRIO, MQTT_THREAD_TICK);
    
    if (NULL != c->mqtt_thread) {
        mqtt_set_client_state(c, CLIENT_STATE_CONNECTED);
        platform_thread_startup(c->mqtt_thread);
        platform_thread_start(c->mqtt_thread);       /* start run mqtt thread */
    }
  5. 而对于重连来说则不会重新创建线程,直接改变客户端状态为连接状态即可:

    mqtt_set_client_state(c, CLIENT_STATE_CONNECTED);

订阅报文

int mqtt_subscribe(mqtt_client_t* c, const char* topic_filter, mqtt_qos_t qos, message_handler_t handler)

订阅报文使用异步设计来实现的,参数有字符串类型的主题(支持通配符"#" "+"),主题的服务质量,以及收到报文的处理函数`,如不指定则有默认处理函数。

过程如下:

  1. 序列化订阅报文并且发送给服务器

    MQTTSerialize_subscribe(c->write_buf, c->write_buf_size, 0, mqtt_get_next_packet_id(c), 1, &topic, (int*)&qos)
    mqtt_send_packet(c, len, &timer)
  2. 创建对应的消息处理节点,这个消息节点在收到服务器的SUBACK订阅应答报文后会挂载到消息处理列表mqtt_msg_handler_list

    mqtt_msg_handler_create(topic_filter, qos, handler)
  3. 在发送了报文给服务器那就要等待服务器的响应了,先记录这个等待SUBACK

    mqtt_ack_list_record(c, SUBACK, mqtt_get_next_packet_id(c), len, msg_handler)

取消订阅

int mqtt_unsubscribe(mqtt_client_t* c, const char* topic_filter);

与订阅报文的逻辑基本差不多的,指定了取消订阅的主题。

实现过程如下:

  1. 序列化订阅报文并且发送给服务器

    MQTTSerialize_unsubscribe(c->write_buf, c->write_buf_size, 0, packet_id, 1, &topic)
    mqtt_send_packet(c, len, &timer)
  2. 创建对应的消息处理节点,这个消息节点在收到服务器的UNSUBACK取消订阅应答报文后将消息处理列表mqtt_msg_handler_list上的已经订阅的主题消息节点销毁

    mqtt_msg_handler_create((const char*)topic_filter, QOS0, NULL)
  3. 在发送了报文给服务器那就要等待服务器的响应了,先记录这个等待UNSUBACK

    mqtt_ack_list_record(c, UNSUBACK, packet_id, len, msg_handler)

发布报文

int mqtt_publish(mqtt_client_t* c, const char* topic_filter, mqtt_message_t* msg)

向指定主题发布一个MQTT报文。参数只有mqtt_client_t 类型的指针,字符串类型的主题(支持通配符),要发布的消息(包括服务质量消息主体)。

使用如下:

mqtt_message_t msg;

msg.qos = 2;
msg.payload = (void *) buf;

mqtt_publish(&client, "testtopic1", &msg);

代码的实现核心思想都差不多,过程如下:

  1. 先序列化发布报文,然后发送到服务器

    MQTTSerialize_publish(c->write_buf, c->write_buf_size, 0, msg->qos, msg->retained, msg->id,topic, (unsigned char*)msg->payload, msg->payloadlen);
    mqtt_send_packet(c, len, &timer)
  2. 对于QOS0的逻辑,不做任何处理,对于QOS1和QOS2的报文则需要记录下来,在没收到服务器应答的时候进行重发

    if (QOS1 == msg->qos) {
        rc = mqtt_ack_list_record(c, PUBACK, mqtt_get_next_packet_id(c), len, NULL);
    } else if (QOS2 == msg->qos) {
        rc = mqtt_ack_list_record(c, PUBREC, mqtt_get_next_packet_id(c), len, NULL);
    }
  3. 还有非常重要的一点,重发报文的MQTT报文头部需要设置DUP标志位,这是MQTT协议的标准,因此,在重发的时候作者直接操作了报文的DUP标志位,因为修改DUP标志位的函数我没有从MQTT库中找到,所以我封装了一个函数,这与LwIP中的交叉存取思想是一个道理,它假设我知道MQTT报文的所有操作,所以我可以操作它,这样子可以提高很多效率:

    mqtt_set_publish_dup(c,1);  /* may resend this data, set the udp flag in advance */

内部线程

static void mqtt_yield_thread(void *arg)

主要是对mqtt_yield函数的返回值做处理,比如在disconnect的时候销毁这个线程。

核心的处理函数

  1. 数据包的处理mqtt_packet_handle

    static int mqtt_packet_handle(mqtt_client_t* c, platform_timer_t* timer)

    对不同的包使用不一样的处理:

    switch (packet_type) {
        case 0: /* timed out reading packet */
            break;
    
        case CONNACK:
            break;
    
        case PUBACK:
        case PUBCOMP:
            rc = mqtt_puback_and_pubcomp_packet_handle(c, timer);
            break;
    
        case SUBACK:
            rc = mqtt_suback_packet_handle(c, timer);
            break;
            
        case UNSUBACK:
            rc = mqtt_unsuback_packet_handle(c, timer);
            break;
    
        case PUBLISH:
            rc = mqtt_publish_packet_handle(c, timer);
            break;
    
        case PUBREC:
        case PUBREL:
            rc = mqtt_pubrec_and_pubrel_packet_handle(c, timer);
            break;
    
        case PINGRESP:
            c->ping_outstanding = 0;
            break;
    
        default:
            goto exit;
    }

    并且做保活的处理:

    mqtt_keep_alive(c)

    当发生超时后的处理:

    if (platform_timer_is_expired(&c->last_sent) || platform_timer_is_expired(&c->last_received)) 

    序列化一个心跳包并且发送给服务器

    MQTTSerialize_pingreq(c->write_buf, c->write_buf_size);
    mqtt_send_packet(c, len, &timer);

    当再次发生超时后,表示与服务器的连接已断开,需要重连的操作,设置客户端状态为断开连接

    mqtt_set_client_state(c, CLIENT_STATE_DISCONNECTED);
  2. ack链表的扫描,当收到服务器的报文时,对ack列表进行扫描操作

    mqtt_ack_list_scan(c);

    当超时后就销毁ack链表节点:

    mqtt_ack_handler_destroy(ack_handler);

    当然下面这几种报文则需要重发操作:(PUBACK 、PUBREC、 PUBREL 、PUBCOMP,保证QOS1 QOS2的服务质量)

    if ((ack_handler->type ==  PUBACK) || (ack_handler->type ==  PUBREC) || (ack_handler->type ==  PUBREL) || (ack_handler->type ==  PUBCOMP))
        mqtt_ack_handler_resend(c, ack_handler);
  3. 保持活性的时间过去了,可能掉线了,需要重连操作

    mqtt_try_reconnect(c);

    重连成功后尝试重新订阅报文,保证恢复原始状态~

    mqtt_try_resubscribe(c)

发布应答与发布完成报文的处理

static int mqtt_puback_and_pubcomp_packet_handle(mqtt_client_t *c, platform_timer_t *timer)
  1. 反序列化报文

    MQTTDeserialize_ack(&packet_type, &dup, &packet_id, c->read_buf, c->read_buf_size)
  2. 取消对应的ack记录

    mqtt_ack_list_unrecord(c, packet_type, packet_id, NULL);

订阅应答报文的处理

```c
static int mqtt_suback_packet_handle(mqtt_client_t *c, platform_timer_t *timer)
```
  1. 反序列化报文

    MQTTDeserialize_suback(&packet_id, 1, &count, (int*)&granted_qos, c->read_buf, c->read_buf_size)
  2. 取消对应的ack记录

    mqtt_ack_list_unrecord(c, packet_type, packet_id, NULL);
  3. 安装对应的订阅消息处理函数,如果是已存在的则不会安装

    mqtt_msg_handlers_install(c, msg_handler);

取消订阅应答报文的处理

```c
static int mqtt_unsuback_packet_handle(mqtt_client_t *c, platform_timer_t *timer)
```
  1. 反序列化报文

    MQTTDeserialize_unsuback(&packet_id, c->read_buf, c->read_buf_size)
  2. 取消对应的ack记录,并且获取到已经订阅的消息处理节点

    mqtt_ack_list_unrecord(c, UNSUBACK, packet_id, &msg_handler)
  3. 销毁对应的订阅消息处理函数

    mqtt_msg_handler_destory(msg_handler);

来自服务器的发布报文的处理

```c
static int mqtt_publish_packet_handle(mqtt_client_t *c, platform_timer_t *timer)
```
  1. 反序列化报文

    MQTTDeserialize_publish(&msg.dup, &qos, &msg.retained, &msg.id, &topic_name,
            (unsigned char**)&msg.payload, (int*)&msg.payloadlen, c->read_buf, c->read_buf_size)
  2. 对于QOS0、QOS1的报文,直接去处理消息

    mqtt_deliver_message(c, &topic_name, &msg);
  3. 对于QOS1的报文,还需要发送一个PUBACK应答报文给服务器

    MQTTSerialize_ack(c->write_buf, c->write_buf_size, PUBACK, 0, msg.id);
  4. 而对于QOS2的报文则需要发送PUBREC报文给服务器,除此之外还需要记录PUBREL到ack链表上,等待服务器的发布释放报文,最后再去处理这个消息

    MQTTSerialize_ack(c->write_buf, c->write_buf_size, PUBREC, 0, msg.id);
    mqtt_ack_list_record(c, PUBREL, msg.id + 1, len, NULL)
    mqtt_deliver_message(c, &topic_name, &msg);

说明:一旦注册到ack列表上的报文,当具有重复的报文是不会重新被注册的,它会通过**mqtt_ack_list_node_is_exist()**函数判断这个节点是否存在,主要是依赖等待响应的消息类型与msgid。

发布收到与发布释放报文的处理

static int mqtt_pubrec_and_pubrel_packet_handle(mqtt_client_t *c, platform_timer_t *timer)
  1. 反序列化报文

    MQTTDeserialize_ack(&packet_type, &dup, &packet_id, c->read_buf, c->read_buf_size)
  2. 产生一个对应的应答报文

    mqtt_publish_ack_packet(c, packet_id, packet_type);
  3. 取消对应的ack记录

    mqtt_ack_list_unrecord(c, UNSUBACK, packet_id, &msg_handler)

上一篇mqttclient配置及裁剪工具

下一篇mqttclient连接到百度天工物接入