4、消息路由层：基于主题的消息分发、请求-响应模式与发布-订阅模式

好，咱们继续往下走。消息路由层，说白了就是网关的「神经中枢」。

你想想看，一个网关要同时对接十几个交易所，每个交易所又有行情、交易、账户、风控各种数据流。如果所有消息都挤在一起，那不乱成一锅粥了？我早年做第一个版本的时候，就吃过这个亏——所有消息走同一个队列，结果行情数据把交易指令给堵死了，那叫一个惨。

所以，消息路由层要解决的核心问题就两个：谁的消息该去哪？谁该收到什么消息？

4.1 两种消息模式：你选哪个？

在实际项目中，我们主要用两种模式：请求-响应和发布-订阅。它们各有各的适用场景，我分别说说。

4.1.1 请求-响应模式

这个模式最直观。客户端发一个请求，服务端处理完返回一个响应。就像你打电话问别人「现在BTC价格多少？」，对方回答你「50000」。一对一，有来有回。

在网关里，哪些场景用这个？

  • 下单指令：客户端提交一个限价单，网关转发给交易所，然后等待交易所返回订单ID和状态。
  • 查询账户：客户端问「我账户里还有多少USDT？」，网关去交易所查了再回复。
  • 撤单操作：客户端说「把订单号12345撤掉」，网关执行后告诉结果。

这里有个关键点：请求和响应必须能对上号。我见过不少新手直接把请求ID丢了，结果响应回来不知道对应哪个请求。嗯，这里要注意，每个请求必须带一个唯一的request_id，响应里也要原样带回来。

核心设计：请求-响应模式在网关里通常用「异步+回调」实现。网关收到请求后，先把请求ID和回调函数存到一个Map里，然后异步发出去。等响应回来，根据request_id找到对应的回调，执行它。这样就不会阻塞主线程。

4.1.2 发布-订阅模式

这个模式更像「广播」。一个消息源发布消息，所有订阅了该主题的客户端都能收到。就像你关注了一个公众号，它一发文章你就能看到。一对多，不需要你主动去问。

在网关里，典型场景：

  • 行情推送：交易所的ticker、orderbook深度、K线数据，这些是持续产生的，需要推给所有订阅者。
  • 成交回报：你的订单被成交了，交易所会推送成交明细。所有关心这个订单的模块（比如风控、清算）都要收到。
  • 账户变动通知：余额变化、持仓变化，这些事件也需要广播出去。

我曾经踩过一个坑：行情数据量太大，直接把消息队列撑爆了。后来加了限流和降级策略，才稳住。所以，发布-订阅模式虽然好用，但一定要考虑背压问题。

4.2 基于主题的消息分发

不管是请求-响应还是发布-订阅，消息都得有个「主题」来分类。主题就是消息的标签，告诉路由层「这条消息属于哪一类」。

我习惯用层级主题，类似文件路径。比如：

  • exchange.binance.spot.ticker.BTCUSDT —— 币安现货的BTC/USDT行情
  • exchange.okx.futures.orderbook.ETHUSDT —— OKX合约的ETH/USDT深度
  • trade.order.create —— 创建订单的指令
  • trade.order.filled —— 订单成交事件

这样做的好处是：支持通配符订阅。比如客户端可以订阅 exchange.binance.*.ticker.*，就能收到币安所有交易对的行情。或者订阅 trade.order.#，就能收到所有订单相关的事件（#表示匹配任意层级）。

小技巧：主题命名要遵循「从大到小」的原则。先按交易所分，再按产品类型分，再按数据类型分，最后按交易对分。这样路由效率最高，也方便做权限控制。

4.3 消息路由的核心数据结构

路由层内部，其实就是一个主题-订阅者映射表。我通常用字典树（Trie）来实现。为什么？因为主题是层级结构的，字典树天然适合做前缀匹配和通配符匹配。

举个例子：

class TopicTrie:
    def __init__(self):
        self.children = {}  # 子节点
        self.subscribers = []  # 订阅者列表
    
    def subscribe(self, topic, subscriber):
        # 按层级插入主题
        parts = topic.split('.')
        node = self
        for part in parts:
            if part not in node.children:
                node.children[part] = TopicTrie()
            node = node.children[part]
        node.subscribers.append(subscriber)
    
    def match(self, topic):
        # 匹配所有可能的订阅者
        parts = topic.split('.')
        results = []
        self._match_recursive(parts, 0, self, results)
        return results
    
    def _match_recursive(self, parts, idx, node, results):
        if idx == len(parts):
            results.extend(node.subscribers)
            return
        part = parts[idx]
        # 精确匹配
        if part in node.children:
            self._match_recursive(parts, idx+1, node.children[part], results)
        # 通配符 * 匹配单层
        if '*' in node.children:
            self._match_recursive(parts, idx+1, node.children['*'], results)
        # 通配符 # 匹配剩余所有层
        if '#' in node.children:
            results.extend(node.children['#'].subscribers)

这段代码看着简单，但实际项目里要考虑并发安全。我一般用读写锁来保护这个字典树，因为订阅操作很少（写），但匹配操作非常频繁（读）。

4.4 请求-响应模式的实现细节

刚才说了，请求-响应模式的核心是请求ID与回调的映射。我直接给个简化版的实现：

import asyncio
from typing import Dict, Callable, Awaitable

class RequestResponseRouter:
    def __init__(self):
        self._pending: Dict[str, asyncio.Future] = {}
    
    async def send_request(self, request_id: str, message: dict, timeout: float = 5.0) -> dict:
        # 创建一个Future，用于等待响应
        future = asyncio.get_event_loop().create_future()
        self._pending[request_id] = future
        
        try:
            # 发送消息到交易所（伪代码）
            await self._transport.send(message)
            # 等待响应，带超时
            response = await asyncio.wait_for(future, timeout=timeout)
            return response
        except asyncio.TimeoutError:
            # 超时处理
            self._pending.pop(request_id, None)
            raise TimeoutError(f"Request {request_id} timed out")
    
    def on_response(self, request_id: str, response: dict):
        # 收到响应时，找到对应的Future并设置结果
        future = self._pending.pop(request_id, None)
        if future and not future.done():
            future.set_result(response)

注意：这个Future一定要设置超时。我见过有人忘了设超时，结果交易所那边挂了，所有请求都卡在pending里，内存直接爆了。超时时间建议根据交易所的响应速度动态调整，比如币安一般100ms以内，但极端行情下可能到1秒。

4.5 发布-订阅模式的实现细节

发布-订阅模式相对简单，核心就是主题匹配+广播。但要注意性能问题，尤其是行情数据量大的时候。

我常用的优化手段：

  • 批量推送：把多个小消息合并成一个批次，减少网络开销。
  • 增量更新：比如orderbook深度，只推送变化的部分，而不是全量数据。
  • 订阅者分级：把订阅者分成不同优先级，行情数据可以降级处理，但交易指令必须优先。

来看一个简单的发布-订阅实现：

class PubSubRouter:
    def __init__(self):
        self._trie = TopicTrie()  # 用刚才的字典树
    
    def subscribe(self, topic: str, subscriber):
        self._trie.subscribe(topic, subscriber)
    
    def unsubscribe(self, topic: str, subscriber):
        # 从字典树中移除订阅者（实现略）
        pass
    
    async def publish(self, topic: str, message: dict):
        # 找到所有匹配的订阅者
        subscribers = self._trie.match(topic)
        # 并发推送
        tasks = [sub.on_message(topic, message) for sub in subscribers]
        await asyncio.gather(*tasks, return_exceptions=True)

避坑指南：我曾经用asyncio.gather不加return_exceptions=True，结果一个订阅者挂了，所有订阅者都收不到消息。加上这个参数后，异常会被捕获，不影响其他订阅者。

4.6 两种模式的融合

在实际网关里，这两种模式不是孤立的。它们经常混用。比如：

  • 客户端通过请求-响应发送下单指令，网关处理后，通过发布-订阅把成交回报广播给所有相关模块。
  • 行情服务通过发布-订阅推送数据，但客户端第一次连接时，需要通过请求-响应获取全量快照。

我习惯在网关内部维护一个统一的消息总线，把两种模式都封装进去。对外暴露的接口很简单：send_request() 和 subscribe()。内部自动判断走哪种模式。

总结一下：消息路由层是网关的「大脑」。请求-响应模式适合有明确返回的操作，发布-订阅模式适合持续推送的数据流。基于主题的分发让消息管理变得清晰，字典树是实现高效匹配的好工具。记住，性能、可靠性、超时处理，这三个点一个都不能少。

消息路由层核心架构 客户端层（交易终端、风控系统、清算模块） 消息路由层 请求-响应模式 下单指令 · 账户查询 · 撤单 request_id + Future 映射 发布-订阅模式 行情推送 · 成交回报 · 账户变动 主题匹配 + 广播 基于主题的分发引擎（字典树实现） 交易所适配层（币安 · OKX · 火币 · Bybit · 等） 图：消息路由层在网关中的位置

好了，这一章的内容就这些。消息路由层是网关里最考验设计功底的部分，既要快，又要稳，还要灵活。希望这些实战经验能帮你少走弯路。

← 上一章 📖 返回目录 下一章 →