libhv教程19--MQTT的实现与使用

返回192.168.1.1

最后更新：2022-01-31 12:25:32 手机定位技术交流文章

MQTT(消息队列遥测传输)是ISO 标准(ISO/IEC PRF 20922)下基于发布/订阅范式的消息协议。它工作在TCP/IP协议族上，是为硬件性能低下的远程设备以及网络状况糟糕的情况下而设计的发布/订阅型消息协议。

MQTT是一个基于客户端-服务器的消息发布/订阅传输协议。MQTT协议是轻量、简单、开放和易于实现的，这些特点使它适用范围非常广泛。在很多情况下，包括受限的环境中，如：机器与机器（M2M）通信和物联网（IoT）。其在，通过卫星链路通信传感器、偶尔拨号的医疗设备、智能家居、及一些小型化设备中已广泛使用。

更多关于MQTT协议的介绍可自行百度，下文主要介绍使用libhv中MQTT的实现与使用。

文章目录

- MQTT实现
- - 协议定义
  - 接口定义
  - - mqtt_client_run
    - mqtt_client_stop
    - mqtt_client_connect 连接
    - mqtt_client_disconnect 断开连接
    - 连接回调 on_connect
    - 断链回调 on_close + 重连 mqtt_client_reconnect
    - 包回调 on_packet
- 使用示例
- - 编译
  - broker
  - 订阅端
  - 发布端
  - 代码流程图

MQTT实现

实现代码位于 mqtt目录
mqtt
整个实现只有不到700行，以往教程中我是很少讲源码解析的，但MQTT的实现非常精简且经典，非常适合用来学习，从而掌握使用libhv开发自定义应用层协议的技巧。

MQTT固定头格式（1字节flags+varint编码的长度字段）：

0	1-2	3	4-7
retain	qos	dup	type

最简单的MQTT消息（比如长度为0的PING消息），只需2个字节，所以MQTT协议是非常精简的协议。

协议定义

见 mqtt_protocol.h

头文件里主要定义了mqtt_type_e消息类型、mqtt_head_t头部、mqtt_message_t消息。

协议里稍微复杂的是MQTT_TYPE_CONNECT，即连接成功后发送的登录认证消息。
MQTT连接标示：

0	1	2	3-4	5	6	7
reserved	clean_session	has_will	will_qos	will_retain	has_password	has_username

接口定义

见 mqtt_client.h

接口列表：

mqtt_client_new：新建MQTT客户端结构体
mqtt_client_free：释放MQTT客户端结构体
mqtt_client_run：运行MQTT客户端
mqtt_client_stop：停止MQTT客户端
mqtt_client_set_id：设置客户端ID
mqtt_client_set_will：设置遗嘱
mqtt_client_set_auth：设置认证用户名密码
mqtt_client_set_callback：设置回调
mqtt_client_set_userdata：设置用户数据
mqtt_client_get_userdata：获取用户数据
mqtt_client_get_last_error：获取最后的错误码
mqtt_client_set_ssl_ctx：设置SSL_CTX（用于SSL/TLS加密通信）
mqtt_client_new_ssl_ctx：新建SSL_CTX
mqtt_client_set_reconnect：设置重连
mqtt_client_reconnect：重连
mqtt_client_connect：开始连接
mqtt_client_disconnect：断开连接
mqtt_client_publish：发布
mqtt_client_subscribe：订阅
mqtt_client_unsubscribe：取消订阅

下面我们介绍几个主要接口的实现：

mqtt_client_run

mqtt_client_run就是调用hloop_run，启动一个事件循环；

mqtt_client_stop

mqtt_client_stop就是调用hloop_stop，停止事件循环；

mqtt_client_connect 连接

hio_create_socket，创建一个套接字，返回一个hio_t对象；
hio_enable_ssl启用SSL/TLS，当然你可以调用mqtt_client_new_ssl_ctx去设置证书等选项；
hevent_set_userdata：设置用户数据（绑定上下文）；
hio_setcb_connect：设置连接回调函数；
hio_setcb_close：设置关闭回调函数；
hio_connect：开始连接；

mqtt_client_disconnect 断开连接

mqtt_client_set_reconnect：因为是主动断开连接，首先取消重连来避免触发断线重连；
mqtt_send_disconnect：发送一个MQTT主动断开连接的消息；
hio_close：关闭连接；

连接回调 on_connect

hio_set_unpack: 设置拆包规则，支持固定包长、分隔符、头部长度字段三种常见的拆包方式，调用该接口设置拆包规则后，内部会根据拆包规则处理粘包与分包，保证回调上来的是完整的一包数据，大大节省了上层处理粘包与分包的成本，MQTT协议对应头部长度字段 这种拆包规则，关于该接口的更详细介绍见 libhv教程14–200行实现一个纯C版jsonrpc框架
hio_setcb_read：设置读回调，因为上面设置了拆包规则，回调上来的就是完整的一包数据，所以叫on_packet;
hio_read：开始读；
mqtt_client_login：MQTT登录认证；

断链回调 on_close + 重连 mqtt_client_reconnect

断链回调里除了调用cli->cb通知上层掉线外，另外就是判断如果设置了重连，则启动一个定时器一段时间后再尝试重连；

reconn_setting_can_retry：判断是否还有剩余重连次数；
reconn_setting_calc_delay：计算重连延时；
htimer_add：添加一个定时器；
mqtt_client_reconnect：因为连接时已经记录了host、port、ssl等信息，重连就是再次调用mqtt_client_connect；

包回调 on_packet

包回调里，就是先调用mqtt_head_unpack解析MQTT头部，然后根据消息类型做对应处理，都是些协议相关的细节了，值得一提的是hio_set_heartbeat设置了一个心跳函数，每隔一段时间发送一个MQTT应用层心跳包来保活。