第二階段：核心組件深度解讀
模組 05：消息總線 — 系統的神經網絡理解組件間如何溝通 — 這是整個系統的通訊基礎
在模組 02 中我們提到，所有組件透過消息總線 (MessageBus) 通訊。現在讓我們深入理解它的運作機制。
一、為什麼需要消息總線？
❌
沒有消息總線
組件直接互相調用。每加一個新組件，就要修改所有相關組件。10 個組件就有 45 條連線。
結果：牽一髮動全身
✅
有消息總線
所有組件只和消息總線交互。新增組件只需要「訂閱」感興趣的主題。
結果：完全解耦，擴展自如
flowchart LR
subgraph before["❌ 直接連接 — 網狀結構 (TB式)"]
direction TB
A1["DataEngine"] <--> B1["ExecutionEngine"]
A1 <--> C1["RiskEngine"]
B1 <--> C1
end
subgraph after["✅ 消息總線 — 星型結構 (LR式)"]
direction LR
A2["組件 A"] --- MB["📡 MessageBus"] --- B2["組件 B"]
C2["組件 C"] --- MB --- D2["組件 D"]
end
圖 1：無消息總線（左）vs 有消息總線（右）的組件連接方式
二、三種通訊模式
消息總線支援三種完全不同的通訊模式，適用於不同場景：
2.1 發布/訂閱 (Pub/Sub)
📢 一對多廣播
發布者不知道誰在聽，訂閱者不知道誰在發。它們透過主題 (Topic) 鬆耦合連接。
sequenceDiagram
participant DE as 📊 DataEngine<br/>(發布者)
participant MB as 📡 MessageBus
participant S1 as 🎯 策略 A<br/>(訂閱者)
participant S2 as 🎯 策略 B<br/>(訂閱者)
participant Port as 💰 Portfolio<br/>(訂閱者)
Note over S1,Port: 啟動時各自訂閱
S1->>MB: subscribe("data.quotes.BTCUSDT.BINANCE")
S2->>MB: subscribe("data.quotes.BTCUSDT.BINANCE")
Port->>MB: subscribe("data.quotes.*")
Note over DE: 收到新報價
DE->>MB: publish("data.quotes.BTCUSDT.BINANCE", quote)
MB->>S1: 推送報價
MB->>S2: 推送報價
MB->>Port: 推送報價
圖 2：發布/訂閱模式 — DataEngine 發布報價，多個組件同時收到
2.2 請求/回應 (Request/Response)
🔄 一問一答
一個組件發出請求，期待收到一個回應。適用於需要「查詢」數據的場景。
sequenceDiagram
participant S as 🎯 你的策略
participant MB as 📡 MessageBus
participant DE as 📊 DataEngine
S->>MB: request(DataType.BAR, instrument_id)
MB->>DE: 轉發請求
DE->>DE: 查詢歷史 K 線
DE->>MB: response(bars_data)
MB->>S: 回傳結果
圖 3：請求/回應模式 — 策略查詢歷史數據
2.3 點對點 (Direct Messaging)
📬 指名送達
直接把消息發送給特定的組件。通常用於命令傳遞。
sequenceDiagram
participant RE as 🛡️ RiskEngine
participant MB as 📡 MessageBus
participant EE as ⚙️ ExecutionEngine
Note over RE: 風控通過，轉發訂單命令
RE->>MB: send("ExecutionEngine", submit_order_cmd)
MB->>EE: 直接送達
圖 4：點對點模式 — 風控引擎直接把命令發給執行引擎
三、主題路由系統
消息總線用主題字串來路由消息，類似於 MQTT 或 Redis 的頻道概念：
| 主題模式 | 說明 | 例子 |
|---|---|---|
data.quotes.{symbol}.{venue} | 特定品種的報價 | data.quotes.BTCUSDT.BINANCE |
data.trades.{symbol}.{venue} | 特定品種的逐筆成交 | data.trades.ETHUSDT.OKX |
data.bars.{bar_type} | 特定類型的 K 線 | data.bars.BTCUSDT.BINANCE-1-MINUTE-LAST |
events.order.{venue} | 特定交易所的訂單事件 | events.order.BINANCE |
events.position.* | 所有持倉事件（萬用字元） | 匹配所有持倉相關事件 |
💡 萬用字元訂閱
使用 * 可以匹配任何內容。例如 data.quotes.* 會接收所有品種的報價。這在你想要監控全部市場活動時非常有用。
四、消息流水線
了解一條消息是如何在系統中流動的：
flowchart LR
subgraph flow["消息流水線"]
direction LR
A["1️⃣ 產生"] --> B["2️⃣ 發布"] --> C["3️⃣ 匹配"] --> D["4️⃣ 同步調度"] --> E["5️⃣ 完成"]
end
F["⏱️ 關鍵特性：<br/>同步完成<br/>保證順序"]
F -.-> D
style flow fill:#0f1419,stroke:#4fc3f7,stroke-width:1px
圖 5：消息的同步調度流程
⚠️ 同步調度的意義
這意味著當 DataEngine 發布一個報價事件時，它會等待所有訂閱者（你的策略、Portfolio 等）都處理完畢後才繼續。好處是事件順序嚴格保證，壞處是如果你的策略處理太慢，會拖慢整個系統。
五、Redis 外部總線
在實盤環境中，消息總線可以選擇性地連接到 Redis，實現：
💾
消息持久化
所有消息寫入 Redis Streams，即使系統崩潰也不會丟失。重啟後可以從 Redis 恢復狀態。
🌐
跨進程通訊
多個 TradingNode 可以透過 Redis 共享消息。適用於分布式部署場景。
📊
外部監控
外部工具（如監控面板）可以訂閱 Redis 中的交易事件，即時監控系統狀態。
flowchart LR
subgraph node1["🖥️ TradingNode 1"]
s1["策略 A"]
mb1["消息總線"]
s1 --> mb1
end
subgraph node2["🖥️ TradingNode 2"]
s2["策略 B"]
mb2["消息總線"]
s2 --> mb2
end
subgraph redis["🔴 Redis"]
streams["Redis Streams<br/>消息持久化"]
end
mb1 <-->|"發布/訂閱"| streams
mb2 <-->|"發布/訂閱"| streams
圖 6：多節點透過 Redis 共享消息
六、節流器 (Throttler)
交易所通常有 API 頻率限制。消息總線內建節流器來防止超過限制：
🚦 節流器工作原理
假設 Binance 限制每秒最多 10 個訂單請求：
- 節流器設置為
limit=10, interval=1s - 前 10 個訂單命令立即通過
- 第 11 個命令進入緩衝隊列，等到下一秒再發送
- 這樣既保證了最大吞吐量，又不會被交易所拉黑
七、互動練習
🧠 練習：選擇通訊模式
以下場景應該使用哪種通訊模式？（發布/訂閱、請求/回應、點對點）
- DataEngine 收到新的 BTC 報價，需要通知所有策略 → ____
- 策略想查詢過去 100 根 K 線的歷史數據 → ____
- RiskEngine 風控通過後，需要把訂單送給 ExecutionEngine → ____
- 新的訂單成交了，需要通知 Portfolio 和策略更新 → ____
- 發布/訂閱 — 一個發布者，多個訂閱者
- 請求/回應 — 需要查詢特定數據並等待結果
- 點對點 — 特定命令發給特定組件
- 發布/訂閱 — 訂單事件需要通知多個組件
八、本模組重點回顧
📋 你應該記住的關鍵點
- 消息總線是所有組件間唯一的通訊管道，實現完全解耦
- 三種模式：發布/訂閱（一對多）、請求/回應（查詢）、點對點（命令）
- 主題路由：用字串主題和萬用字元匹配訂閱者
- 同步調度：在單線程上同步完成，保證事件順序
- Redis 擴展：支持消息持久化和跨進程通訊
- 節流器：自動限制 API 調用頻率，避免超限