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

Posted 2021-04-05 物联网技术指导

banner

前言

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

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

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

Fixed header	固定报头，所有控制报文都包含
Variable header	可变报文，部分控制报文包含
Payload	有效载荷，部分控制报文包含

固定报头 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/

下图为逻辑流程时序图：

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): 将引用计数减少。

分隔线君

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

物联网技术指导知识星球

物联网技术指导 发起了一个读者讨论 Thingsboard 通过Netty 实现MQTT Broker的性能和EMQX的性能，大家决定那个更高呢？