TBMQ 架构

简介
- 架构图
设计动机
TBMQ工作原理概览
单机模式与集群模式
编程语言

简介

本文介绍TBMQ的架构结构，说明数据在各组件间的流转及核心架构决策。 TBMQ经精心设计，具备以下特性：

可扩展性：采用先进开源技术构建的水平可扩展平台；
容错性：无单点故障，集群内每个broker（节点）在功能上相同；
健壮高效：可管理数百万客户端，每秒处理数百万条消息；
持久性：提供高消息持久性，确保数据不丢失。

架构图

下图展示broker的核心部分及消息传输路径。

设计动机

在ThingsBoard，我们积累了丰富的可扩展IoT应用构建经验，从而识别出基于MQTT解决方案的三大主要场景。

第一种场景中，大量设备产生海量消息，由特定应用消费，形成 扇入（Fan-In） 模式。通常由少数应用处理这些海量入站数据，必须确保不丢失任何一条消息。
第二种场景中，大量设备订阅需送达的特定更新或通知。少量入站请求引发大量出站数据。此场景称为 扇出（Fan-Out，广播） 模式。
第三种场景为点对点（P2P）通信，针对性强的一对一消息传递模型。适用于私信或命令-响应交互等用例，消息通过唯一定义的主题在单个发布者与特定订阅者之间路由。

在上述所有场景中，通常使用将服务质量（QoS）设为1或2的持久客户端，以确保可靠的消息投递，即使因重启或升级而暂时离线仍然成立。

基于上述场景，我们有意将TBMQ设计为对三者均表现优异。

我们的设计原则侧重于确保broker的容错性和高可用性。因此，设计上刻意避免了对主节点或协调流程的依赖。我们确保了集群内所有节点具有相同的功能。

优先支持分布式处理，支持随着操作增长而轻松实现水平可扩展性。broker应当支持高吞吐量，并保证向客户端低延迟交付消息。确保数据持久性和复制在我们的设计中至关重要。目标是构建一个系统，其中broker一旦确认接收到消息，该消息就是安全的，不会丢失。

为确保满足上述要求，并防止在客户端或部分broker实例故障时丢失消息，TBMQ采用 Kafka 作为底层基础设施。

TBMQ工作原理概览

Kafka在MQTT消息处理的各个阶段扮演关键角色。所有未处理的已发布消息、客户端会话和订阅均存储在专用Kafka主题中。TBMQ使用的Kafka主题完整列表见此处。所有broker节点均可通过这些主题访问最新的客户端会话和订阅状态，并在本地维护副本以高效处理与投递消息。当客户端与某broker节点断开时，其它节点可基于最新状态无缝继续运作。此外，新加入集群的broker节点在启动时即可获得这些关键信息。

客户端订阅在MQTT发布/订阅模式中具有重要意义。TBMQ采用 Trie 数据结构以优化性能，能够高效地在内存中保持客户端订阅的持久性，并便于快速访问相关主题模式。

当发布者客户端发送 PUBLISH 消息时，该消息被存储在初始Kafka主题 tbmq.msg.all 中。Kafka一旦确认消息的持久性，broker就会根据选定的QoS级别向发布者发送 PUBACK/PUBREC 消息或不发送任何响应。

随后，充当Kafka消费者的单独线程从上述Kafka主题中检索消息，并利用Subscription Trie数据结构识别预期接收者。根据客户端类型（DEVICE 或 APPLICATION）和下述持久化选项，broker要么将消息重定向到另一个特定Kafka主题，要么直接将消息传递给接收者。

非持久客户端

当 CONNECT 包满足以下条件时，客户端被归类为非持久客户端：

对于 MQTT v3.x：

将 clean_session 标志设为 true。

对于 MQTT v5：

将 clean_start 标志设为 true，且 sessionExpiryInterval 为 0 或未指定。

对于非持久客户端，发往它们的消息直接发布，无需额外持久化。须注意非持久客户端仅能为 DEVICE 类型。

集群模式下的非持久DEVICE处理

在集群模式下，多个TBMQ节点可协同运行，各自在 tbmq.msg.all 主题的同一消费者组中运行Kafka消费者。这样可以实现节点间的负载均衡和消息分发。但可能出现消息由某一TBMQ节点处理，而目标订阅者连接在另一节点的场景。为此，使用 downlink.basic Kafka主题在TBMQ节点间通信。确保处理消息的节点将消息转发给目标节点，再由其通过已建立连接投递给订阅者。

持久客户端

不满足上述非持久条件的MQTT客户端归类为持久客户端。持久客户端的条件如下：

对于 MQTT v3.x：

将 clean_session 标志设为 false。

对于 MQTT v5客户端：

sessionExpiryInterval 大于 0（无论 clean_start 为何）。
将 clean_start 标志设为 false，且 sessionExpiryInterval 为 0 或未指定。

基于我们在IoT生态中的实践及大量用例的成功落地，我们将MQTT客户端划分为两类：

DEVICE 客户端：主要大量发布消息，订阅的topic数量有限、消息速率较低，通常对应IoT设备或传感器。
APPLICATION 客户端：专注于订阅高消息速率的topic，通常需要离线时持久化消息并在稍后投递，确保关键数据不丢失，常用于实时分析、数据处理等应用层功能。

因此，我们通过为这两类客户端分离处理流程来优化性能。

持久DEVICE客户端

对于持久DEVICE客户端，我们使用 tbmq.msg.persisted Kafka主题处理从 tbmq.msg.all 提取的已发布消息。该设计将持久消息与其他类型消息分离处理，保证流程清晰高效。作为Kafka消费者的专用线程拉取这些消息并存入用于持久化的 Redis。此方法尤其适合DEVICE客户端，因其通常不需要接收大量消息，在客户端重连时可顺畅恢复已存消息，同时在中等到达速率场景下保持良好性能。

关于Redis作为DEVICE客户端持久消息存储的详细说明见此处。

集群模式下的持久DEVICE处理

与非持久客户端类似，集群模式下TBMQ节点协同运行，在 tbmq.msg.persisted 主题的同一消费者组中运行Kafka消费者。若消息由一节点处理而订阅者连接在另一节点，则通过 downlink.persisted Kafka主题将消息转发到目标节点，确保通过已建立连接将消息无缝投递给订阅者。

持久APPLICATION客户端

APPLICATION客户端数量与所用Kafka主题数量一一对应。最新版Kafka可支持数百万主题，使该设计即使面对最大规模企业用例仍适用。

从 tbmq.msg.all 主题中读取的、发往特定APPLICATION客户端的消息会存入对应Kafka主题。每个APPLICATION客户端分配一个独立线程（Kafka消费者），这些线程从对应Kafka主题拉取消息并投递给相应客户端。该方式通过高效投递显著提升性能。此外，Kafka消费者组的特性使 MQTT5 共享订阅对APPLICATION客户端极为高效。

APPLICATION客户端可处理大量接收消息，可达每秒数百万条。需注意，APPLICATION客户端仅能归类为持久客户端。

对于两种类型的客户端，我们提供了可配置的工具来控制每个客户端的消息是否持久化以及存储的时长。您可以参考以下环境变量来调整这些设置：

TB_KAFKA_APP_PERSISTED_MSG_TOPIC_PROPERTIES；

MQTT_PERSISTENT_SESSION_DEVICE_PERSISTED_MESSAGES_LIMIT；

MQTT_PERSISTENT_SESSION_DEVICE_PERSISTED_MESSAGES_TTL。

有关详细信息，请参阅以下文档。

集群模式下的持久APPLICATION处理

APPLICATION客户端在集群模式下的功能与独立设置相同，无需节点间通信。消息处理直接在目标节点进行，因为一旦客户端连接，就会为该APPLICATION客户端创建一个专属消费者。这种设计确保无缝快速的消息交付。通过避免额外的消息传输步骤，这种方法确保了显著改进的性能，即使系统水平扩展也不例外。

Kafka主题

以下是TBMQ使用的Kafka主题及说明。

tbmq.msg.all-存储来自MQTT客户端的全部已发布消息的主题。
tbmq.msg.app. + ${client_id}-存储APPLICATION客户端根据其订阅应接收的消息。
tbmq.msg.app.shared. + ${topic_filter}-存储APPLICATION客户端根据共同共享订阅应接收的消息。
tbmq.msg.persisted-存储DEVICE持久客户端根据订阅应接收的消息。
tbmq.msg.retained-存储所有保留消息，与MQTT保留消息功能相关。
tbmq.client.session-存储所有客户端会话。
tbmq.client.subscriptions-存储所有客户端订阅。
tbmq.client.session.event.request-存储客户端会话事件，如 CONNECTION_REQUEST、DISCONNECTION_REQUEST、CLEAR_SESSION_REQUEST 等。
tbmq.client.session.event.response. + ${service_id}-存储对上一主题事件的响应，发往目标客户端所连接的broker节点。
tbmq.client.disconnect. + ${service_id}-存储强制断开客户端连接事件（由UI/API管理员请求或会话冲突触发）。
tbmq.msg.downlink.basic. + ${service_id}-用于在broker节点间转发消息至DEVICE订阅者当前所连节点。
tbmq.msg.downlink.persisted. + ${service_id}-用于在broker节点间转发消息至DEVICE持久订阅者当前所连节点。
tbmq.sys.app.removed-处理APPLICATION客户端主题移除的事件，在客户端从APPLICATION转为DEVICE时使用。
tbmq.sys.historical.data-存储各broker节点发布的集群历史统计（如入站/出站消息数等）以计算集群总值。
tbmq.client.blocked-存储和分发已封禁客户端列表，阻止其连接broker。
tbmq.sys.internode.notifications + ${service_id}-broker节点间系统通知，用于同步认证提供方设置、认证管理员设置，并触发本地客户端会话缓存清理。

Redis

Redis 是强大的内存数据存储，在需要低延迟与高吞吐数据访问的场景中表现优异，是存储实时数据的理想选择。

TBMQ使用Redis存储DEVICE持久客户端的消息，为这些客户端实现高效的持久化与投递。Redis能以极快读写在内存中管理大数据集，结合Redis Cluster的可扩展性，确保持久消息在存储量增长、系统需求提升时仍能高效检索与投递。该可扩展性使Redis可通过多节点分发数据，无缝处理更大工作负载，保持高性能与可靠性。

PostgreSQL数据库

TBMQ使用 PostgreSQL 存储用户、用户凭据、MQTT客户端凭据、统计数据、WebSocket连接、WebSocket订阅等各类实体。

PostgreSQL以可靠、健壮、灵活著称，是确保数据完整性并支持高事务吞吐的强大开源关系型数据库。在TBMQ中，PostgreSQL作为系统关键元数据的主存储层。鉴于MQTT应用的高消息量与频繁读写特性，PostgreSQL的事务管理与ACID合规提供强一致性保证，确保重要数据在任何场景下都能安全存储与检索。

Web UI

TBMQ提供轻量友好的图形界面（GUI），以直观高效的方式简化broker管理。该GUI具备以下核心功能：

MQTT客户端凭据管理：用户可通过GUI创建、修改和删除MQTT客户端凭据。
客户端会话与订阅控制：管理员可监控和控制客户端会话状态，包括终止和管理活跃连接，并管理客户端订阅（添加、移除、修改）。
共享订阅管理：管理员可创建和管理Application共享订阅实体，实现向多个APPLICATION类型订阅客户端的消息高效分发。
保留消息管理：管理员可管理broker保存并投递给新订阅者的保留消息。
WebSocket客户端：GUI支持WebSocket客户端，管理员可建立、监控和管理WebSocket连接，用户可通过MQTT over WebSocket与TBMQ交互，实时调试和测试连接与消息流。

除上述管理功能外，GUI还提供监控仪表板，展示broker性能的全面统计与洞察，帮助管理员更好理解系统健康与性能。

Netty

为采用成熟高效的技术，我们选择Netty实现支持MQTT协议的TCP服务端。

Netty是高性能异步事件驱动网络框架，适用于构建可扩展、健壮的网络应用，非常适合TBMQ处理MQTT流量。选择Netty的重要原因之一是其高效处理大量并发连接的能力——在成千上万甚至数百万设备持续连接并交换数据的IoT环境中，Netty以较低资源消耗处理并发连接的能力非常关键。

Netty使用非阻塞I/O（NIO），无需为每个连接分配独立线程即可高效管理资源，显著降低开销。该方式在重负载下仍能保证高吞吐与低延迟，适合TBMQ这类MQTT broker的高要求场景。Netty具备高度灵活性，支持按协议需求定制网络栈；其模块化设计便于实现协议处理、消息解析和连接管理，并提供TLS加密选项，满足安全性与未来扩展需求。

Actor系统

TBMQ使用Actor系统作为底层机制，实现负责处理MQTT客户端的Actor。采用Actor模型可实现客户端消息的高效并发处理，确保高性能运行。

broker使用专为TBMQ定制的Actor系统实现。系统中存在两类Actor：

Client Actors：为每个连接的MQTT客户端创建对应的Client Actor，负责处理 CONNECT、SUBSCRIBE、UNSUBSCRIBE、PUBLISH 等主要消息类型，处理与MQTT客户端的交互并执行相关消息操作。
Persisted Device Actors：除Client Actor外，为所有持久类型的DEVICE客户端额外创建Persisted Device Actor，专门管理持久化相关操作及持久DEVICE客户端的消息存储与检索。

凭借Actor系统与多种Actor类型，TBMQ高效并发处理消息，在与客户端交互时保持最优性能与快速响应。更多关于Actor模型的介绍见链接。

消息分发服务

消息分发服务（Message dispatcher service）是TBMQ架构的另一核心组件，负责管理发布客户端与Kafka之间的消息流，确保安全处理、持久化及向正确订阅者的高效投递。

消息分发服务在通过Actor系统收到发布者的消息后开始工作。服务将消息发布到Kafka，保证其持久存储以安全处理与持久化。该步骤保证即便节点故障或临时断开，消息也不会丢失，使Kafka成为处理大量MQTT流量的可靠骨干。

Kafka确认消息已存储后，消息分发服务拉取消息并利用 Subscription Trie 分析哪些订阅者有资格接收。确定订阅者后，根据订阅者类型决定处理方式：

DEVICE非持久客户端：消息可立即投递给客户端。
DEVICE持久客户端：消息发布到DEVICE客户端Kafka主题，随后存入Redis。
APPLICATION持久客户端：消息发布到专用Application Kafka主题。

为每个订阅者确定并完成处理路由后，消息交给Netty进行实际网络传输，投递给相应的在线客户端。

订阅Trie

Trie数据结构因其层次化组织与基于前缀的匹配，非常利于快速查找。仅存储一次共同前缀即可快速匹配topic，减少搜索空间与内存占用。其时间复杂度取决于topic长度而非topic数量，即便在有数百万订阅的大规模环境中仍能保持稳定快速的查找。

在TBMQ中，所有客户端订阅从Kafka主题消费，并存入内存中的Trie数据结构。Trie按层级组织topic过滤器，每个节点代表topic过滤器的一级。

从Kafka读取 PUBLISH 消息时，broker需识别所有与消息topic匹配的订阅客户端并确保其收到消息。Trie可根据topic名称高效检索客户端订阅。一旦确定相关订阅，消息副本将转发给各对应客户端。

该方式可实现高性能消息处理，快速精确确定需接收特定消息的客户端。须注意因Trie存储在内存中，该方法会增加broker的内存占用。更多Trie数据结构信息见链接。

单机模式与集群模式

TBMQ设计为可水平扩展，支持自动向集群添加新的broker节点。集群内所有节点功能相同，共同承担整体负载，实现客户端连接与消息处理的均衡分布。

broker设计无需“主节点”或“协调进程”，无层级或中央节点管理其他节点。该去中心化设计消除了单点故障，提升系统整体健壮性与容错能力。

处理客户端连接请求时，可使用自选的负载均衡器，将入站连接分发到所有可用TBMQ节点，均衡工作负载并提高资源利用率。

若客户端与某broker节点断开（如节点关闭、移除或网络故障），可轻松重连至集群内任意健康节点。该无缝重连能力确保持续运行与不中断服务，客户端可与集群内任意可用节点建立连接。

通过水平扩展、负载均衡与新节点自动发现，TBMQ为大规模部署中的MQTT通信提供可扩展且具有容错能力的架构。

编程语言

TBMQ后端使用Java 17实现，前端以Angular 19框架开发为SPA。