ThingsBoard MQTT 协议登录源码分析

正文与配图来自原稿。聚焦 MQTT CONNECT（登录/鉴权） 在 CE 中的调用链；Netty 基础不展开。可与 MQTT Transport 源码分析 对照。

1、概述

ThingsBoard 支持多种协议接入；MQTT Transport 基于 Netty。设备「登录」本质是发 CONNECT，平台校验凭证后建立会话。

2、入口：处理 MQTT 消息

消息进入后走 processMqttMsg(ctx, message)：

CONNECT 类型进入 processConnect(ctx, (MqttConnectMessage) msg)：

常见 Token 方式再进 processAuthTokenConnect(ctx, msg)：

3、Transport API：异步鉴权请求

进入 transportService.process() → doProcess(transportType, protoMsg, callback)：

核心是 transportApiRequestTemplate.send(protoMsg)：得到 ListenableFuture，再挂消费回调，最后在线程中收尾：

sendToRequestTemplate(...) 把请求发到配置的 Topic（如 queue.transport_api.requests_topic）：

单体常用 内存队列 实现 TbQueueProducer：

4、对端：ResponseTemplate 消费请求

发送侧是 DefaultTbQueueRequestTemplate，接收侧对应 DefaultTbQueueResponseTemplate：轮询 poll 取请求。

先看 TbQueueConsumer / 内存 poll：

处理消息、校验并查询设备：

鉴权结果写回响应 Topic：

回到 ResponseTemplate 发出响应：

5、回到 RequestTemplate：成功回调

Transport 侧 DefaultTbQueueRequestTemplate 收到响应后执行回调：

校验设备响应数据后，打开会话：

transportService.process(sessionInfo, SESSION_EVENT_MSG_OPEN, callback)

注意消息类型为 SessionEventMsg：

6、再进 Core：DeviceActor

sendToDeviceActor 再次入队（如 tb_core）。消费者在 DefaultTbCoreConsumerService，处理可跟到 forwardToDeviceActor：

进入 Actor：系统 Actor → AppActor → DeviceActor，消息进 MailBox；再处理 session_event：

7、链路小结

Netty 收 CONNECT
  → processMqttMsg → processConnect → processAuthTokenConnect
  → TransportApi RequestTemplate（requests_topic / 内存队列）
  → ResponseTemplate poll → 查库鉴权 → 响应 Topic
  → RequestTemplate onSuccess
  → SESSION_EVENT_MSG_OPEN
  → tb_core → DeviceActor 建立会话

集群下队列实现换成 Kafka 等，模板与 Actor 边界不变。联调若报 NO_ACTIVE_CONNECTION，多半是会话未建立或未订 RPC Topic，见 RPC NO_ACTIVE_CONNECTION。