Thingsboard源码分析MQTT设备连接协议(下)

前言

MQTT协议通过预定义的MQTT控制报文来通信，这一节按照MQTT控制报文类型，通过跟踪Thingsboard处理这些协议报文来解读程序如何处理协议的。

MQTT控制报文由三部分组成，按照 表格1 –MQTT控制报文的结构` 描述的顺序：

表格1 –MQTT控制报文的结构

固定报头 Fixed header: 1. MQTT控制报文类型，2. 标志Flags(DUP1=控制报文的重复分发标志，PUBLISH报文的服务质量等级，RETAIN3=PUBLISH报文的保留标志)，3.剩余长度 Remaining Length

可变报文: 报文标识符Packet Identifier

有效载荷: 某些MQTT控制报文载报文的最后部分包含一个有效载荷，对于PUBLISH来说有效载荷就是应用消息。

更多MQTT信息参见:

• MQTT3.1 英文文档 http://public.dhe.ibm.com/software/dw/webservices/ws-mqtt/mqtt-v3r1.html

• MQTT3.1中文文档 https://mcxiaoke.gitbook.io/mqtt/

下图为逻辑流程时序图：

处理入口

代码块1-1

1@Override
2    public void channelRead(ChannelHandlerContext ctx, Object msg) {
3        log.trace("[{}] Processing msg: {}", sessionId, msg);
4        try {
5            if (msg instanceof MqttMessage) {
6                processMqttMsg(ctx, (MqttMessage) msg);
7            } else {
8                ctx.close();
9            }
10        } finally {
11            ReferenceCountUtil.safeRelease(msg);
12        }
13    }


第5-9行代码，通过判断读取到的数据是否为MqttMessage，如果是则进入processMqttMsg(cox, (MqttMessage) msg)类; 如果不是则主动关闭连接。

第11行代码，自从Netty 4开始，对象的生命周期由它们的引用计数(reference counts)管理，而不是由垃圾收集器(garbage collector)管理了。Bytebuf是最值得注意的，它使用了引用计数来改进分配内存和释放内存的性能。

ReferenceCountUtil.safeRelease(msg): 将引用计数减少。

试看到此啦! 更多内容请加知识星球。