ThingsBoard MQTT Transport 源码分析
正文与配图来自原稿。源码路径对齐分支
release-4.3。
相关：项目结构介绍 · application.yml 配置 · 自定义规则节点等
0、模块放在哪？（release-4.3 补充）
读 MQTT 传输建议分清三层：
| 路径 | 作用 |
|---|---|
common/transport/mqtt | 业务主体：Adaptor、Handler、Session、IP 限流、MqttTransportService（Netty 服务） |
netty-mqtt | MQTT 协议编解码 / 连接生命周期等底层 |
transport/mqtt | 独立微服务启动器（service.type=tb-transport 时单独进程） |
包名（common/transport/mqtt）大致为：
org.thingsboard.server.transport.mqtt
├── adaptors/ # Json / Proto / 兼容适配器
├── limits/ # IpFilter、ProxyIpFilter、限流统计
├── session/ # DeviceSessionCtx、Gateway 等
├── MqttTransportService.java
├── MqttTransportHandler.java
├── MqttTransportContext.java
└── MqttTransportServerInitializer.java下文分析的类，默认都在 common/transport/mqtt。
1、消息转换适配器 adaptors
首先看 mqtt-transport 的模块结构：
adaptors： 主要是抽象 MQTT 设备上传数据的转换方式。
类结构关系
MqttTransportAdaptor 接口里面定义了不同消息的转换方式：
release-4.3 中常见实现：
JsonMqttAdaptor：JSON 载荷ProtoMqttAdaptor：Protobuf 载荷BackwardCompatibilityAdaptor：向后兼容适配器
BackwardCompatibilityAdaptor 主要作用是适配 JsonMqttAdaptor 和 ProtoMqttAdaptor：默认先用 JsonMqttAdaptor 做数据转换，如果转换失败再用 ProtoMqttAdaptor。
适配器如何注入？
ThingsBoard 抽象了一个 TransportContext（传输层上下文），用于传输层需要使用的对象。MqttTransportContext 继承了 TransportContext，同时注入了 JsonMqttAdaptor 和 ProtoMqttAdaptor：
系统默认使用 JsonMqttAdaptor 来转换，体现在哪里？
需要看 MqttTransportService 在启动 Netty 服务端时注入了 MqttTransportContext：
然后将 MqttTransportContext 传到相关构造器，把数据转换对象存入 deviceSessionCtx：
2、IP 过滤
包含 IpFilter 和 ProxyIpFilter，里面实现的逻辑都一样。这两个类都是 InboundHandler，都继承了 ChannelInboundHandlerAdapter，所以都是 handler 处理器，需要在 channel 初始化时添加进去。详细可以搜索一下 AbstractRemoteAddressFilter 的作用。
进入 IpFilter 分析：当设备发送数据时，会触发这个 Handler，然后调用里面的 accept 做逻辑处理：
获取到客户端的 IP，进入 context.checkAddress(remoteAddress)，先判断配置（原稿写 application.yml，现多用环境变量，见 配置文）中是否开启 IP 速率限制（如 TB_TRANSPORT_IP_RATE_LIMITS_ENABLED）：
假如开启了速率限制，再看下面的逻辑：
var stats = ipMap.computeIfAbsent(address.getAddress(), a -> new InetAddressRateLimitStats());
return !stats.isBlocked() || (stats.getLastActivityTs() + ipBlockTimeout < System.currentTimeMillis());首先：
Map<InetAddress, InetAddressRateLimitStats> ipMap = new ConcurrentHashMap<>();意思是：设备第一次发数据时，先按 IP 在 ipMap 查是否有 InetAddressRateLimitStats；没有就创建一个放进去。InetAddressRateLimitStats 类如下：
第一次 !stats.isBlocked() 肯定为 true，再比较：
stats.getLastActivityTs() + ipBlockTimeout < System.currentTimeMillis()ipBlockTimeout 是配置值。若返回 true 就继续，否则不再接收。
如果设备达到速率限制，会把 isBlocked() 改为 true，再通过 Netty 底层的 AbstractRemoteAddressFilter 拒绝接收，从而实现限制。
3、如何放开限制
如果某一个设备的 IP 被限制了，如何放开？
在 DefaultTransportService 里启动了一个定时任务，一直去检查：
然后将 IP 限制移除，或修改 blocked 状态：
（配置上还可关注 TB_TRANSPORT_MAX_WRONG_CREDENTIALS_PER_IP、TB_TRANSPORT_IP_BLOCK_TIMEOUT 等，见 MQTT 配置页。）
4、设备 Session 管理
ThingsBoard 对设备 session 的抽象：
release-4.3 中常见类型包括：
DeviceSessionCtxMqttDeviceAwareSessionContext- Gateway / Sparkplug 相关 Session（网关、工业协议扩展）
deviceSessionCtx 的初始化：
Session 上会挂：设备身份、Adaptor、待发消息队列等，后续 CONNECT / PUBLISH 都围绕它展开。
5、MQTT 服务启动
根据条件注解判断是否启动 MQTT（原稿表达式，与版本细节可能略有出入，以源码为准）：
@ConditionalOnExpression(
"'${service.type:null}'=='tb-transport'
|| ('${service.type:null}'=='monolith'
&& '${transport.api_enabled:true}'=='true'
&& '${transport.mqtt.enabled}'=='true')"
)含义概要：
- 独立 Transport 微服务（
tb-transport），或 - 单体（
monolith）且开启 transport API 与 MQTT
满足则加载 MqttTransportService：
这是一个标准的 Netty 服务端启动：Boss/Worker、Pipeline（解码、IpFilter、Ssl、MqttTransportHandler 等）。MQTT 接入基于 Netty 实现；协议编解码细节还可下钻 netty-mqtt 模块。
6、设备连接
一切都从 channelRead 开始执行：
release-4.3 中逻辑概要：若是成功解码的 MqttMessage，则调用 processMqttMsg(ctx, message)。
进入 processMqttMsg 后：
CONNECT 流程会先验证设备 token（或其它凭证）：
进入 onValidateDeviceResponse(msg, ctx, connectMessage)，会与 tb-core 交互，需要把设备的 active 等状态置为在线：
鉴权通过后 Session 才算真正连上，后续才能处理 PUBLISH 等报文。
7、设备发送数据
原稿此节标题已列出，正文未展开。下面按 release-4.3 的 MqttTransportHandler 补一条阅读路径（辅）：
channelRead→processMqttMsg- 对 PUBLISH：进入
processPublish(ctx, mqttMsg) - 先
checkConnected：未完成连接则直接返回 - 取
topicName/packetId - 若 Topic 以网关 API 前缀开头（如
MqttTopics.BASE_GATEWAY_API_TOPIC），走GatewaySessionHandler；否则走普通设备上报 - 通过 Session 上的 Adaptor（JSON / Proto / 兼容）把 payload 转成平台内部消息
- 经 TransportService / 队列 交给 Core / Rule Engine（入库、规则链等）
读代码时建议断点顺序：
MqttTransportHandler#channelRead
→ processMqttMsg
→ processPublish
→ JsonMqttAdaptor / ProtoMqttAdaptor（转换）
→ DefaultTransportService（上行到 Core）设备侧 Topic 约定仍见官方 MQTT Device API（如遥测 v1/devices/me/telemetry）。
阅读路径小结
启动: MqttTransportService（Netty）+ ConditionalOnExpression
↓
接入: IpFilter / ProxyIpFilter（可选 IP 限流）
↓
处理: MqttTransportHandler#channelRead → processMqttMsg
↓
连接: CONNECT → 校验 Token → onValidateDeviceResponse → Session / 设备 active
↓
上报: PUBLISH → processPublish → Adaptor 转换 → 队列 / Core补充（辅）
- 配置项见 MQTT Transport 配置 与官网 MQTT 配置页（
MQTT_BIND_PORT、TB_TRANSPORT_IP_RATE_LIMITS_ENABLED等）。 - 网关设备、Sparkplug 在
session/gateway包，比单设备路径更复杂，可第二轮再读。 - 改 Adaptor 或 Handler 后需重新编译；打 jar / 镜像见专栏前文。