## 1. 系统概述

coordinate-broadcast 是整个 Coordinate 消息系统中负责消息路由和分发的核心组件，基于 **rumqttd** 进行二次开发，扮演着 MQTT 消息代理（Broker）的角色。系统包含两个独立的 Broker 实例：

- **Broadcast Server（MQTT v5 Extended）**：内部服务器通信，使用纯 TCP 连接，实现完整 MQTT 协议并扩展 AddSubscribe/RemoveSubscribe 用于动态订阅管理
- **WebSocket Server（MQTT v5 Lite）**：外部客户端接入，使用 WebSocket 传输，仅支持连接管理相关消息

作为分布式消息架构的核心枢纽，coordinate-broadcast 接收来自 coordinate-server 的事件发布请求，并通过 MQTT 协议将消息推送至各个客户端（coordinate-connector 或 coordinate-sdk 接入的客户端）。

从功能定位角度来看，coordinate-broadcast 主要承担三类职责：第一是作为消息路由中枢，接收发布者的消息并根据主题过滤器进行智能路由；第二是作为连接管理器，管理大量并发 MQTT 客户端的连接生命周期；第三是作为消息存储器，通过内存段（Segment）结构维护消息历史，支持消息回溯和离线推送。系统设计遵循高性能、低延迟的原则，采用 Actor 架构模式实现消息处理与连接管理的解耦，整体架构能够支撑十万量级的并发连接。

## 2. 系统架构

### 2.1 整体架构图

```mermaid
flowchart TB
    subgraph coordinate-server["coordinate-server"]
        direction LR
        Server["server"]
    end

    subgraph coordinate-broadcast["coordinate-broadcast"]
        subgraph broadcast["Broadcast Server (MQTT v5 Extended) - 内部 TCP"]
            direction TB
            BServer["Server"]
        end

        subgraph router["Router - 核心路由"]
            direction TB
            Router["Router"]
            Subs["订阅映射"]
            Sched["调度器"]
        end

        subgraph link["Link - 网络I/O"]
            direction TB
            Link["Link"]
        end

        subgraph segments["Segments - 消息存储"]
            direction TB
            Seg["分段存储"]
        end

        subgraph ws["WebSocket Server (MQTT v5 Lite) - 外部客户端"]
            direction TB
            WSServer["Server"]
        end
    end

    subgraph clients["MQTT Clients"]
        direction LR
        Connector["connector"]
        SDK["coordinate-sdk"]
    end

    Server -->|"MQTT v5 Extended"| BServer
    BServer --> Router
    Router <--> Link
    Router <--> Seg
    Link --> WSServer
WSServer -->|"MQTT over WebSocket"| clients
```

### 2.2 模块职责划分

coordinate-broadcast 采用模块化的架构设计，每个模块承担独立的职责，通过清晰定义的接口进行模块间交互。这种设计使得各模块可以独立演进，同时便于测试和维护。系统主要包含以下六个核心模块：

**server 模块**负责网络监听和连接接受，是系统对外的入口点。它启动 TCP 监听器（支持 WebSocket 和纯 TCP 两种模式），接受客户端连接并进行 TLS 握手处理。在接受连接后，server 模块负责解析 MQTT Connect 包，验证客户端身份，并创建对应的 Link 实例来处理后续通信。该模块还负责 WebSocket 子协议协商，确保使用 "mqtt" 子协议进行通信。

**protocol 模块**实现了 MQTT 协议的解析与序列化功能，是系统进行协议级通信的基础。该模块实现了两个协议版本，用于不同的场景：
- **v5 Extended**：用于**内部服务器通信**（broadcast 模块），实现完整的 MQTT 协议并扩展 AddSubscribe/RemoveSubscribe 用于动态订阅管理
- **v5 Lite**：用于**外部 WebSocket 客户端**，仅支持连接管理相关消息（ Connect/ConnAck/Ping/Disconnect ）
协议实现采用 trait 抽象设计，提供了良好的扩展性，能够方便地添加新协议版本支持。

**router 模块**是系统的核心消息路由引擎，负责管理所有订阅关系、执行消息路由、调度消费者并维护连接状态。它维护订阅映射表（Subscription Map），将主题过滤器映射到对应的连接集合；当收到发布消息时，router 根据订阅关系查找订阅者并将消息推送给目标连接。该模块还实现了共享订阅（Shared Subscription）功能，支持消息在订阅同一主题的多个消费者之间进行负载均衡分发。

**link 模块**负责网络 I/O 操作和协议编解码，是连接生命周期管理的核心实现。每个 Link 实例对应一个 MQTT 客户端连接，负责处理入站和出站数据流。Link 模块使用异步通道（Channel）与 Router 进行通信，实现消息的接收和发送。该模块还负责处理 QoS（服务质量）确认、消息重传等可靠性机制。

**broadcast 模块**实现了专门的广播服务器功能，用于将消息分发至多个订阅者。与普通的 pub/sub 不同，广播服务器针对大规模分发场景进行了优化，能够高效地将同一条消息推送给数千个订阅者。该模块通常与 coordinate-server 配合使用，处理系统级别的事件广播。

**segments 模块**实现了消息的持久化存储功能，使用 CommitLog 结构管理消息历史。每个 Segment 是一个内存映射文件，包含固定大小的消息块。当消息进入时，会追加到当前的 Segment；当 Segment 满时，会创建新的 Segment 并对旧的进行归档。这种设计使得系统能够支持消息回溯和离线推送，同时通过定期清理旧 Segment 来控制内存使用。

### 2.3 数据流设计

系统的核心数据流遵循“接收→路由→分发”的三阶段设计，每一阶段都由独立的异步任务处理，保证系统的高吞吐量。以下是消息从接收到分发的完整数据流程：

在第一阶段连接建立时，server 模块接受客户端 TCP 连接，创建 Network 层进行底层 I/O 处理。客户端发送 MQTT Connect 包后，系统解析连接参数并验证身份认证信息。认证通过后，创建 RemoteLink 实例并将连接信息注册到 Router。Router 为该连接分配唯一的 ConnectionId，并创建对应的入站和出站队列用于消息缓冲。

在第二阶段消息接收时，客户端通过 Link 发送 Publish 数据包。Link 将数据包放入连接对应的入站队列，然后通知 Router 进行处理。Router 从队列中取出数据包，首先进行 QoS 确认处理（对于 QoS 1 需要回复 PubAck）。然后将消息追加到 CommitLog 进行持久化存储。最后根据消息主题查找订阅映射表，确定目标订阅者并生成数据请求。

在第三阶段消息分发时，Router 将数据请求加入调度队列。调度器（Scheduler）根据连接状态和背压情况决定分发时机。当连接可以接收新数据时，调度器从队列中取出请求并调用 Link 的发送方法。Link 将数据包序列化后通过 TCP 套接字发送给客户端。如果连接背压（Buffer Full），调度器会暂停向该连接分发数据，等待客户端确认接收完成后继续。

## 3. 核心组件设计

### 3.1 Broker 核心类

Broker 是 coordinate-broadcast 的主控制类，负责协调各模块启动和管理全局状态。以下是 Broker 的核心设计：

```rust
pub struct Broker {
    pub config: Config,
    pub router_tx: Sender<(ConnectionId, Event)>,
}

impl Broker {
    pub async fn start(&mut self) -> Result<(), Error> {
        // 1. 启动广播服务器
        if let Some(config) = self.config.broadcast.clone() {
            tokio::spawn(async move {
                let mut broadcast_server = BroadcastServer::new(config, router_tx, V0);
                broadcast_server.start().await.unwrap();
            });
        }

        // 2. 启动 WebSocket 服务器（MQTT over WebSocket）
        if let Some(mut config) = self.config.ws.clone() {
            let mut server = Server::new(config, self.router_tx.clone(), V5);
            server.start(LinkType::Websocket).await;
        }
    }
}
```

Broker 的启动流程，首先根据配置启动广播服务器（Broadcast Server），然后启动 WebSocket 服务器（WebSocket Server）。

### 3.2 Router 核心类

Router 是系统的核心消息路由引擎，采用 Actor 模式设计，通过事件驱动的方式处理各类消息。以下是 Router 的核心数据结构：

```rust
pub struct Router {
    pub id: RouterId,
    pub config: RouterConfig,
    pub subscription_map: HashMap<Topic, HashSet<ConnectionId>>,
    pub shared_subscriptions: HashMap<String, SharedGroup>,
    pub scheduler: Scheduler,
    pub cache: RingBuffer<IncomingPacket>,
}

impl Router {
    pub fn events(&mut self, id: ConnectionId, data: Event) {
        match data {
            Event::Connect { connection, incoming, outgoing } => {
                self.handle_new_connection(connection, incoming, outgoing);
            }
            Event::DeviceData => self.handle_device_payload(id),
            Event::Disconnect => self.handle_disconnection(id, None),
            Event::PublishWill((client_id, _)) => self.handle_last_will(client_id),
        }
    }
}
```

Router 维护几个关键的数据结构：subscription_map 用于将主题过滤器映射到订阅者连接集合；shared_subscriptions 用于管理共享订阅组，支持 RoundRobin、Random 和 Sticky 三种分发策略；scheduler 负责调度消息分发任务，实现背压控制；cache 用于缓存入站数据包，避免频繁分配。

### 3.3 连接管理

连接管理是系统高并发的关键，采用了以下设计来实现高效的连接处理：

```rust
pub struct ConnectionSettings {
    pub connection_timeout_ms: u16,       // 连接超时时间
    pub max_payload_size: usize,          // 最大 payload 大小
    pub max_inflight_count: usize,         // 最大飞行中消息数
    pub auth: Option<HashMap<String, String>>,  // 内部认证配置
    pub external_auth: Option<AuthHandler>,      // 外部认证处理器
    pub dynamic_filters: bool,             // 是否支持动态过滤器
}
```

连接设置提供了灵活的认证机制，支持内部认证（基于配置文件的简单认证）和外部认证（通过异步回调函数进行自定义认证）。max_inflight_count 用于控制每个连接飞行中消息的数量，实现背压控制，防止单个连接占用过多资源导致其他连接延迟。

## 4. 关键技术实现

### 4.1 MQTT 协议实现

系统实现了两个协议版本，用于不同的场景：
- **v5 Extended**：用于**内部服务器通信**，实现完整的 MQTT 协议，包含 Connect、ConnAck、Publish、PubAck、Subscribe、SubAck、Unsubscribe、UnsubAck、PingReq、PingResp、Disconnect，并扩展支持 AddSubscribe（15）、RemoveSubscribe（16）消息类型用于动态订阅管理
- **v5 Lite**：用于**外部 WebSocket 客户端**，仅支持连接管理相关消息（ Connect/ConnAck/Ping/Disconnect ），不支持 Publish/Subscribe

通过统一的 Packet 类型封装所有消息类型：

```rust
pub trait Protocol {
    fn read_mut(&mut self, stream: &mut BytesMut, max_size: usize) -> Result<Packet, Error>;
    fn write(&self, packet: Packet, write: &mut BytesMut) -> Result<usize, Error>;
}

pub enum Packet {
    Connect(Connect, Option<ConnectProperties>, Option<LastWill>, ...),
    ConnAck(ConnAck, Option<ConnAckProperties>),
    Publish(Publish, Option<PublishProperties>),
    PubAck(PubAck, Option<PubAckProperties>),
    Subscribe(Subscribe, Option<SubscribeProperties>),
    SubAck(SubAck, Option<SubAckProperties>),
    // ... 其他包类型
}
```

协议实现的关键设计要点包括：第一，使用统一的 Packet 类型封装所有 MQTT 消息类型，通过 serde 库实现序列化；第二，通过 Protocol trait 抽象协议实现，便于扩展新版本；第三，实现了 QoS 机制，支持 AtMostOnce（0）和 AtLeastOnce（1）两种服务质量级别。

v5 Lite 是 MQTT v5 的子集，仅保留连接管理相关消息。

### 4.2 WebSocket 传输

系统支持 MQTT over WebSocket，允许浏览器或 Web 环境通过 WebSocket 协议接入。以下是 WebSocket 握手的关键实现：

```rust
struct WSCallback;
impl Callback for WSCallback {
    fn on_request(self, request: &Request, mut response: Response) -> Result<Response, ErrorResponse> {
        // 协商 mqtt 子协议
        if request.headers().get("sec-websocket-protocol").is_some() {
            response.headers_mut().insert("sec-websocket-protocol", HeaderValue::from_static("mqtt"));
        }
        Ok(response)
    }
}

// WebSocket 连接处理
let stream = accept_hdr_async(network, WSCallback).await?;
let stream = Box::new(WsStream::new(s));
task::spawn(remote(config, router_tx, stream, protocol, will_handlers));
```

WebSocket 传输的关键设计是在握手阶段协商 "mqtt" 子协议，确保使用标准化的 MQTT 消息格式进行后续通信。这种设计充分利用了 WebSocket 的全双工通信能力和防火墙穿透能力，同时保持与原生 MQTT 的兼容性。

### 4.3 发布/订阅机制

发布/订阅是系统的核心功能，实现了灵活的主题匹配和高效的消息分发：

```rust
pub struct Filter {
    pub path: String,           // 主题路径
    pub qos: QoS,              // QoS 级别
    pub nolocal: bool,         // 不转发本地发布
    pub preserve_retain: bool,
    pub retain_forward_rule: RetainForwardRule,
}

impl Router {
    fn prepare_filter(&mut self, id: ConnectionId, cursor: Offset, filter_idx: FilterIdx, 
                      filter: &protocol::Filter, group: Option<String>, subscription_id: Option<usize>) {
        // 1. 添加到订阅映射
        match self.subscription_map.get_mut(&filter.path) {
            Some(connections) => { connections.insert(id); }
            None => {
                let mut connections = HashSet::new();
                connections.insert(id);
                self.subscription_map.insert(filter.path.clone(), connections);
            }
        }

        // 2. 处理共享订阅
        if let Some(group_name) = &group {
            let shared_group = self.shared_subscriptions
                .entry(group_name.to_string())
                .or_insert(SharedGroup::new(cursor, self.config.shared_subscriptions_strategy.clone()));
            shared_group.add_client(client_id);
        }

        self.scheduler.track(id, request);
    }
}
```

订阅机制支持精确匹配和通配符匹配两种模式：单层通配符（+）匹配单层路径，多层通配符（#）匹配多层路径。共享订阅允许多个消费者订阅同一主题，系统通过策略模式实现消息分发，支持 RoundRobin（轮询）、Random（随机）和 Sticky（粘性）三种策略。

### 4.4 消息持久化

消息持久化使用分段日志（CommitLog）结构实现，平衡性能和存储开销：

```rust
pub struct Segment {
    pub id: u64,
    pub path: PathBuf,
    pub file: File,
    pub size: usize,
    pub cursor: u64,
}

impl Segment {
    pub fn append(&mut self, data: &[u8]) -> Result<Offset, Error> {
        let offset = self.cursor;
        self.file.write_all(data)?;
        self.cursor += data.len() as u64;
        Ok((self.id, offset))
    }
}
```

分段存储的设计要点包括：每个 Segment 有固定的最大大小（默认 1GB），当 Segment 满时创建新 Segment；消息以追加模式写入，保证顺序性；通过定期清理旧 Segment 来控制磁盘空间使用。消息持久化支持离线消息和消息回溯，是 MQTT 消息代理的核心功能之一。

## 5. 配置设计

### 5.1 配置文件结构

系统通过 mqtt.toml 配置文件进行配置，支持灵活的系统调优。以下是完整的配置结构：

```toml
# 路由器配置
[router]
id = 0
max_connections = 100010
max_outgoing_packet_count = 2000
max_segment_size = 1048576000    # 1GB
max_segment_count = 100

# WebSocket 服务器配置
[ws]
name = "ws"
listen = "0.0.0.0:8000"
next_connection_delay_ms = 1
[ws.connections]
connection_timeout_ms = 60000
max_payload_size = 20480
max_inflight_count = 100
# auth = { broadcast = "password" }

# 广播服务器配置
[broadcast]
name = "broadcast"
listen = "0.0.0.0:21884"
next_connection_delay_ms = 1
[broadcast.connections]
connection_timeout_ms = 60000
max_payload_size = 20480
max_inflight_count = 100

# Prometheus 指标
[prometheus]
listen = "127.0.0.1:9043"
interval = 1

# 指标推送配置
[metrics]
[metrics.alerts]
push_interval = 10
[metrics.meters]
push_interval = 2
```

### 5.2 配置项说明

配置项按照功能分为以下几个组：

**路由器配置**控制消息路由的核心行为：max_connections 限制最大连接数，防止资源耗尽；max_outgoing_packet_count 控制每个连接出队缓冲大小；max_segment_size 和 max_segment_count 控制消息存储的磁盘空间。

**服务器配置**控制网络监听行为：listen 指定监听地址和端口；next_connection_delay_ms 控制新连接的接受速率，防止瞬时连接风暴；connection_timeout_ms 控制空闲连接的超时时间。

**连接配置**控制单个连接的行为：max_payload_size 限制单条消息的最大大小；max_inflight_count 控制飞行中消息的数量，实现背压控制；auth 用于配置简单的用户名密码认证。

## 6. 系统交互设计

### 6.1 与 coordinate-server 的交互

coordinate-server 通过 coordinate-connector 库使用 **MQTT v5 Extended** 连接到 Broadcast Server，发布系统事件并接收订阅状态更新：

```rust
// coordinate-server 侧的事件发布
pub async fn publish(topic: String, event: Event) -> Result<(), AppError> {
    if let Some(senders) = BROADCAST.get() {
        let idx = CLIENT_INDEX.fetch_add(1, Ordering::Relaxed) % senders.len();
        let tx = &senders[idx];
        tx.send(BroadcastEvent::Publish(topic, event)).await?;
    }
    Ok(())
}
```

coordinate-server 维护一个连接池（默认4个连接），通过 round-robin 方式分发发布请求，实现负载均衡。每个连接独立维护 MQTT 会话，支持 Clean Session 和持久会话两种模式。

### 6.2 消息流总结

系统中事件流转的完整路径如下：coordinate-server 产生业务事件，通过 connector 客户端发布到 coordinate-broadcast；broadcast 根据订阅关系将消息路由到订阅该主题的所有客户端；客户端接收消息后触发本地业务处理逻辑。这种架构实现了组件间的松耦合，便于横向扩展。

## 7. 设计模式总结

### 7.1 架构模式

系统采用以下架构模式实现高性能和高可用性：

**Actor 模式**：Router 作为核心 Actor，通过 Channel 与连接通信。所有的状态修改都在 Router 单线程中处理，避免了锁竞争和数据竞争。

**事件驱动模式**：系统基于 tokio 的异步事件循环，所有 I/O 操作都是非阻塞的，能够充分利用单线程处理大量并发连接。

**分段存储模式**：使用 Segment 管理消息历史，支持消息回溯和离线推送，同时通过定期清理控制存储开销。

### 7.2 扩展性设计

系统提供了良好的扩展性支持：通过 Protocol trait 可以添加新的协议版本支持；通过 AuthHandler 接口可以实现自定义认证逻辑；通过共享订阅策略可以扩展分发算法。这种设计使得系统能够适应不断变化的业务需求。

## 8. 技术规格

| 指标 | 规格 |
|------|------|
| 支持协议 | MQTT v5 Extended（内部服务器）, MQTT v5 Lite（外部 WebSocket） |
| 传输层 | TCP, WebSocket, TLS |
| 最大并发连接 | 100,000+ |
| 最大消息Payload | 20KB（可配置） |
| 消息持久化 | 分段 CommitLog |
| QoS 级别 | 0, 1 |
| 共享订阅策略 | RoundRobin, Random, Sticky |
| 监控接口 | Prometheus |


coordinate-broadcast 是整个 Coordinate 消息系统中负责消息路由和分发的核心组件，基于 rumqttd 进行二次开发，扮演着 MQTT 消息代理（Broker）的角色。系统包含两个独立的 Broker 实例：

coordinate-broadcast 架构设计

架构设计

网络与通信

后端

Coordinate-broadcast是MQTT消息系统的核心组件，采用高性能Actor架构，支持10万+并发连接。系统包含TCP和WebSocket双协议，实现智能消息路由、连接管理及消息存储功能。支持MQTTv5扩展协议，提供动态订阅管理，适用于大规模分布式消息场景。

消息代理

数据结构

内存映射

负载均衡

数据流

防火墙

服务器

路由器

路由

MQTT

腾讯云 IM，全球通信，1折起


4核4G3M云服务器 新用户低至38元/年！

2026采购季 | AI焕新·智启新局

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云AI代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

功能1上新10个字符

功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符。

功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符。

功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符

功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符

功能4上新

文章&问答评论现已支持表情

全新交互，全新视觉，新增快捷键、悬浮工具栏、高亮块等功能并同时优化现有功能，全面提升创作效率和体验

社区富文本编辑器全新改版！诚邀体验～ 

精选全网热门MCP server，让你的AI更好用 🚀

💥开发者 MCP广场重磅上线！

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

coordinate-broadcast 架构设计-腾讯云开发者社区-腾讯云

coordinate-broadcast 架构设计

coordinate-broadcast 架构设计

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐