CQRS模式读写分离架构设计

可视化演示

展示CQRS核心架构：命令端写入与查询端读取分离、事件溯源重建状态、异步同步与最终一致性三种模式。

模式概述：一个柜台管两件事的混乱

老图书馆只有一个服务台，读者问“《三体》在哪个架”和采编员办“500 本新书入库”挤在一起。查一本书只要 3 秒，但读者必须等采编员把手头的登记做完；入库一件事只要 5 分钟，却总被读者的提问打断。传统软件系统里的单个模块也是这样：一个 User 模型既被下单接口写入，又被列表接口反复读取；写操作要加锁防并发，读操作又抢这把锁。读写挤在同一个模型里，两败俱伤。

CQRS 的主张非常朴素：把系统拆成两个模型。命令模型（Command Model）只接收改状态的操作，比如“创建订单”“扣减库存”；查询模型（Query Model）只回答读问题，比如“这个用户最近的 20 个订单是什么”。两者不再是同一个类、同一张表、同一套服务，各按各的节奏演化和扩容。

在动手拆分前，先要算清一笔账：系统的读写比到底是多少？定义读请求量为 \(Q_{\text{read}}\)，写请求量为 \(Q_{\text{write}}\)，读写比 \(R\) 为两者之比：

馆长开馆前做的第一件事不是买书架，而是翻上个月的台账：1000 次窗口请求里，940 次是查书，60 次是办业务。十次请求里有九次半在“读”，他这才下定决心拉红线分柜台。CQRS 的第一步永远是：先看清你的读写比。

命令模型：采编台的办事流程

采编台重新开张后，窗台上多了一叠印刷好的申请单：“新书入库单”上要填书名、ISBN、数量、入库时间。采编员只认单子不认人——单子没填全，直接退回；ISBN 重复了，直接驳回。每张单子办完，采编员在回执上盖章：“已入库”。

这些申请单就是命令（Command）。命令是“意图的载体”：它不描述怎么一步步做，只声明“我想让系统变成什么状态”。命令模型要做的只有三件事：接收命令、校验业务规则、应用状态变更。至于数据将来怎么被查询、被展示成什么样子，命令模型一概不管。

命令模型通常由命令处理器（Command Handler）负责接收。处理器先做业务校验（余额是否充足、库存是否够），再调用领域模型完成状态迁移。记当前状态为 \(S\)，命令为 \(C\)，应用函数 \(\text{apply}\) 负责校验并迁移状态，那么新状态 \(S'\) 可以写成：

采编台永远不会顺手帮你查书，也永远不会在登记簿之外另造一本“读者指南”——它只管把状态改对。改了状态之后的事，交给另一条柜台。这就是命令模型的本分。

查询模型：检索卡的秘密

目录厅里那一排检索卡，是整座图书馆最聪明的设计。卡片上只写三样东西：书名、作者、架位。读者按字母翻，三秒定位。注意，卡片上没有任何“图书分类法”“借阅历史”——那些信息对读者找书毫无帮助，只会让卡片又厚又乱。

查询模型（Query Model，也叫读模型）就是这套检索卡。它不复制业务系统的完整语义，而是按界面与查询场景，把数据“重新排版”一遍。这个过程叫投影（Projection）：从事件流或写侧数据中挑选字段，重组为利于查询的视图。写侧越严谨越好，读侧越“好查”越好。

投影可以形式化地写成：从一组事件 \(E_1, E_2, \ldots, E_n\) 中挑选所需字段并重组为视图 \(V\)：

读者不需要理解图书分类法，就像前端不需要理解领域模型。检索卡替读者做了“翻译”，把复杂的世界折叠成一张好用的卡片——这正是查询模型存在的全部意义。

读写分离与同步：两个柜台之间的传话

分柜台后第一个麻烦来了：新书凌晨 3 点入库，可目录厅的检索卡还是昨天下午的版本。最初馆长让采编员入库后立刻跑去目录厅插卡，结果采编台又堵了——同步变成了新的瓶颈。后来馆长加了条“传话通道”：采编员入库完毕，只要写一张“新书通知单”丢进传送篮，自有专人每 10 分钟收一次篮子，送去目录厅统一更新卡片。采编台不再跑腿，目录厅也不再干等。

技术上，这张“通知单”就是领域事件（Domain Event）。命令模型完成写入后发布事件，同步组件异步地把事件投递给查询模型，查询模型重新执行投影，刷新读数据。两个存储就此物理隔离：写库用关系型数据库保证事务与一致性，读库可以用 Elasticsearch、Redis 或文档数据库专攻查询。

异步带来一个关键指标——同步延迟。设写侧提交时刻为 \(t_{\text{write}}\)，读侧完成同步时刻为 \(t_{\text{read}}\)，则延迟为：

传话通道让两个柜台彻底解耦，代价是检索卡永远比真实书库慢半拍。馆长对读者的答复只有一句：“查到就是查到，查不到，过十分钟再来看。”这句话在技术圈有个正式的名字——最终一致性。

事件溯源：一本流水账本

有一年洪灾泡了目录厅一半卡片，馆长却一点不慌。因为他上任第一天就要求采编台写流水账：某年某月某日 14:30 入《三体》3 本，15:05 下架 1 本……每一条都按时间顺序记。对着一本没被水泡的流水账，馆长派两个实习生从开馆第一天逐条重放，三天后，一套全新的馆藏清单完整还原。事实不分对错，账本永远是对的。

事件溯源（Event Sourcing）就是这个流水账本。传统系统保存“当前状态”，比如余额 70 元；事件溯源保存的是一串“事实”：先入账 100，再支出 30。当前状态不再是事实，而是从事件流里计算出来的投影，随时可以丢弃、随时可以重建。

形式化地写，\(t\) 时刻的状态等于对初始状态 \(S_0\) 依次应用事件 \(e_1\) 到 \(e_t\) 做归约（fold）：

馆长从此不再担心卡片被烧。只要流水账在，任何时刻的馆藏都能“重放”出来。这就是事件溯源给 CQRS 的底气：状态会丢，事实不会。

最终一致性：迟到的检索卡

读者渐渐习惯了检索卡“慢半拍”，但馆长给自己定了三条规矩：第一，卡片迟到最多不能超过 10 分钟，而且 99% 的情况下要做到；第二，同一张新书通知单绝不重复插卡；第三，每周日晚闭馆后全员对账——把真实书库和检索卡逐本比对，发现差异当场修正。

这些规矩翻译成技术语言，就是最终一致性（Eventual Consistency）的三个配套机制：一致性 SLO、幂等消费、对账补偿。CQRS 把“一致性达成时间”从写入提交那一刻推迟到了同步完成那一刻，业务必须接受一个延迟窗口 \(\Delta t\)。

工程上可以给一致性定一个可量化的目标：延迟 \(\Delta t\) 不超过阈值 \(\varepsilon\) 的概率不低于 \(1-\delta\)，即：

馆长的三句话，成了 CQRS 落地团队的座右铭：迟到可以，错账不行；重放安全，幂等护航；平时不管，周日对账。最终一致性不是“没有一致性”，而是把一致性做成一个有界、可度量、可修复的工程承诺。

适用场景与权衡：什么时候该开分馆

消息传开后，隔壁社区小馆的馆长也来取经，照着图纸拉了一条红线。结果呢？全馆一天只有 30 次请求，两个柜台互相闲着，还要多养一个人。小馆长懊恼地拆掉红线。市馆馆长笑着说：分馆不是标准答案，是一道成本题。先算读写比，再算养两个柜台的成本，最后才动红线。

判断是否引入 CQRS，核心是权衡。经验判据有几条：读写比悬殊（R 远大于 1）、查询形态多样（列表、详情、报表、搜索）、写侧与读侧扩展方向不同、团队有精力维护两套模型和同步链路。反过来，简单 CRUD 系统、读写比接近 1、团队小或交付压力大时，CQRS 是负资产。

把权衡写成判据：设同步链路的额外维护成本为 \(C_{\text{sync}}\)，单体读写模型的基础成本为 \(C_{\text{base}}\)，当读写比满足以下条件时，分离的收益才值得付出：

半年后，市馆的年度报告写着：查询吞吐翻了三倍，入库效率提升两成，检索卡和书库的偏差率降到千分之一。而隔壁小馆，依然一个柜台，岁月静好。CQRS 从来不是最时髦的答案，而是最清醒的成本题。答案只属于读懂了自身读写比的人。

AI 应用场景

Python 演示

import timefrom collections import defaultdictfrom dataclasses import dataclass  @dataclassclass Command:    """A write-side intent: what to do and with which payload."""    account_id: str    kind: str  # 'deposit' or 'withdraw'    amount: float  @dataclassclass Event:    """An immutable fact appended to the event store."""    seq: int    account_id: str    kind: str    amount: float  class EventStore:    """Write-side: source of truth, an append-only event log."""     def __init__(self):        self._events = []        self._seq = 0     def append(self, cmd: Command) -> Event:        """Validate business rules, then persist the fact."""        if cmd.kind == 'withdraw' and cmd.amount > self.balance(cmd.account_id):            raise ValueError('insufficient balance')        self._seq += 1        event = Event(self._seq, cmd.account_id, cmd.kind, cmd.amount)        self._events.append(event)        return event     def balance(self, account_id: str) -> float:        """Rebuild current state from events (event sourcing replay)."""        total = 0.0        for e in self._events:            if e.account_id != account_id:                continue            total += e.amount if e.kind == 'deposit' else -e.amount        return total     def stream(self):        """Return all events in order for projection."""        return list(self._events)  class ReadModel:    """Query-side: a denormalized projection for fast reads."""     def __init__(self):        self.balances = defaultdict(float)        self.counts = defaultdict(int)        self.synced_seq = 0     def apply(self, e: Event) -> None:        """Apply one event into the projection view."""        if e.kind == 'deposit':            self.balances[e.account_id] += e.amount        else:            self.balances[e.account_id] -= e.amount        self.counts[e.account_id] += 1        self.synced_seq = e.seq  class Projector:    """Async sync: replay events into the read model."""    # 最终一致性: 读模型延迟一小段时间后才追上写侧     def __init__(self, store: EventStore, read_model: ReadModel, delay: float = 0.01):        self.store = store        self.read_model = read_model        self.delay = delay     def sync(self) -> None:        for e in self.store.stream():            if e.seq <= self.read_model.synced_seq:                continue            time.sleep(self.delay)  # simulate sync lag            self.read_model.apply(e)  def main():    store = EventStore()    read_model = ReadModel()    projector = Projector(store, read_model)     store.append(Command('alice', 'deposit', 100))    store.append(Command('alice', 'withdraw', 30))    store.append(Command('bob', 'deposit', 50))     print('before sync, alice balance =', read_model.balances['alice'])    projector.sync()    print('after sync,  alice balance =', read_model.balances['alice'])    print('event store size =', len(store.stream()))    print('read model synced_seq =', read_model.synced_seq)    print('replayed write-side balance =', store.balance('alice'))  if __name__ == '__main__':    main()

1. class EventStore:命令侧的“事实库”：只允许追加不可变事件，是系统的唯一事实来源，就像采编台的流水账本。
2. if cmd.kind == 'withdraw' and cmd.amount > self.balance(cmd.account_id):命令校验：余额不足直接拒绝，确保写侧业务规则不被破坏。
3. total += e.amount if e.kind == 'deposit' else -e.amount事件溯源重放：不保存余额，从每笔事件累加还原当前值。
4. class ReadModel:查询侧的投影：用最方便查询的字典结构保存余额与交易次数，结构与事件流完全不同。
5. if e.seq <= self.read_model.synced_seq:幂等消费：按事件序号跳过已处理事件，防止重复投影。
6. time.sleep(self.delay)模拟同步延迟，这正是最终一致性“慢半拍”的来源。
7. store.append(Command('alice', 'deposit', 100))写侧只接收命令对象，由事件存储完成校验与落库。

常见误区

面试题