CQRS模式：事件驱动架构实践

可视化演示

展示CQRS读写分离架构与事件溯源流程，动态演示命令到事件存储到投影读模型的完整链路，交互切换CQRS、事件溯源与Saga模式。

CQRS 基础

美味斋老板的痛点，你一定能共鸣：一本“总账本”既要记订单，又要被翻来翻去查菜色。写的人要排队，读的人也要排队，双方互相拖累。老板的解法很朴素——既然记和查是两件本质不同的事，为什么非要共用一本账本？他把账本一分为二：“订单册”只负责写下顾客要什么，“菜单册”只负责回答今天有什么。

这个“分家”的思路，就是 CQRS（Command Query Responsibility Segregation，命令查询职责分离）的核心。它告诉我们：写操作（命令）和读操作（查询）应该使用各自独立的模型，而不是挤在一个模型里将就。

在传统单体系统中，读写共用同一张数据库表，查询复杂了会影响写入性能，写入频繁了又会拖慢查询。CQRS 让写模型专注于业务规则与状态变更，让读模型专注于查询效率与展示形态。两个模型可以放在不同的数据库，比如写库用关系型数据库保证事务，读库用 Elasticsearch 或 Redis 加速检索。两者还能独立伸缩——饭点下单暴增时，只需扩写模型的机器；客人频繁翻菜单时，只需扩读模型的机器。当然，天下没有免费的午餐：读写分离增加了一套同步机制，因此它更适合读写负载差异大、查询模式复杂、需要弹性伸缩的系统。小饭店强行分两本账，反而浪费纸。

老板满意地看着两本册子：点单高峰时，记单的小二忙得满头汗，但查菜的服务员依然笑容满面——写和读，终于各忙各的了。

事件驱动结合

美味斋的后厨有一个默契：厨师做好菜，不会端着盘子满大堂找顾客，而是“叮”地敲一下铃铛。铃铛一响，闲着的服务员自然会来端菜。这声铃铛，就是事件。它不指定“必须由谁来端”，只陈述一个事实——“有一道菜出锅了”。

在事件驱动架构中，“事件”是对已发生事实的描述：订单已创建、菜品已上桌、顾客已结账。它和“命令”有本质区别：命令表达了“我想要做什么”的意图，是尚未发生的请求；事件则表达“已经发生了什么”，是不可改变的事实。CQRS 的写端接收命令、执行业务逻辑，然后把结果发布为事件，供读端和其他系统消费。

这种异步解耦带来的好处立竿见影：生产者（后厨）不需要等消费者（服务员）响应，两者通过消息中间件（如 Kafka、RabbitMQ）连接。生产者往管道里丢一个事件就继续干自己的活，消费者按自己的节奏处理。即便某个消费者暂时故障，事件也会在管道中积压等待，不会丢失。事件流因此具备严格的先后顺序——先下单后上菜，这个顺序必须被保持，否则业务就乱套了。

铃铛声此起彼伏，却井然有序。厨师、服务员、洗碗工各听各的铃，谁也不会堵在谁门口——这就是事件驱动解耦后的节奏感。

事件存储

美味斋后厨角落挂着一本牛皮封面的“流水账”。每一道菜出锅，账房先生就添一行：“12:03 红烧肉出锅”“12:17 清蒸鱼出锅”。这本账只许往上写，不许拿橡皮擦改。谁要问“3 号桌到底上了几个菜”，翻账便知。老板立下规矩：账本不许涂改，这就是餐厅的“铁面流水账”。

事件存储（Event Store）就是系统的这本流水账。它与普通数据库最大的不同：只追加（append-only），不修改、不删除。每一行记录一个不可变的事件事实，并按发生顺序排列。这个特性让事件存储成为事件驱动系统的“真相之源”——它是所有业务事实的唯一权威记录。

基于事件存储，我们可以实现事件溯源（Event Sourcing）：不再保存对象的最新状态，而是保存产生这个状态的全部事件。想要当前状态？把事件从头到尾重放一遍即可。这带来惊人的能力：可以回退到任意历史时刻、可以诊断“当初是怎么变成这样的”、可以随时生成新的视角。代价则是事件会无限增长，需要定期做快照来压缩（把历史事件折叠成一个状态快照，事件继续保留）。

账本越写越厚，但老板心里反而踏实了——哪怕客人投诉“菜上错了”，翻一翻流水账，谁的责任一目了然，谁也改不了历史。

投影重建

服务员手里的“菜单册”是怎么来的？不是后厨现问的，而是账房先生每天早晨根据流水账整理出来的：看见“红烧肉出锅”就记一道红烧肉，看见标记“红烧肉售罄”就把菜单上的红烧肉划掉。菜单册是对流水账的一种“整理视角”——它只关心“今天能点什么菜”，不关心做菜的过程细节。

这个“从事件流整理出查询视图”的过程，就是投影（Projection）。读模型（Read Model）是投影的产物：一张为查询量身定制的表，上面的每一行都来自对事件流的折叠计算。比如电商系统里，“用户订单列表”视图就是由“OrderCreated 订单已创建”“OrderShipped 订单已发货”等事件投影而来的。

投影重建（Rebuild Projection）是 CQRS 的一把利器：既然事件流永久保留，读模型就可以随时推倒重建。菜单册被咖啡弄脏了？没关系，照流水账重新抄一份，几分钟后焕然一新。代码升级后想加一个字段？同样重放事件，生成包含新字段的读模型。读模型还可以按需构建多种形态——同一条事件流，既能投影成“今日菜单”，也能投影成“食材消耗统计”，互不影响。

菜单册旧了、脏了、被改乱了，都不怕——账房先生闭着眼都能重抄一本。读模型永远是“可抛弃、可重建”的临时产物，这份底气来自那本永不磨灭的流水账。

最终一致性

顾客催菜时，老板常说一句话：“菜正在做呢，马上到！”下单和上菜之间，天然存在一段等待时间。如果顾客下单后立刻要求后厨显示“菜已上桌”，那显然不现实。CQRS 系统同样如此：命令写入事件存储后，投影需要一点点时间才会更新读模型。这个“时间差”，就是最终一致性（Eventual Consistency）的延迟窗口。

一致性并不是只有“强”和“弱”两级。强一致性要求任何时刻读到的都是最新值；弱一致性允许读到旧值；最终一致性介于两者之间——它允许短时间内读到旧数据，但保证在没有新写入的情况下，所有读取最终会收敛到最新状态。CQRS 绝大多数场景选择最终一致性，换来的是更低的写入延迟和更高的系统吞吐。

要让最终一致性真正可靠，必须做好两件事：第一，幂等设计（Idempotency），消费者可能收到重复事件（比如消息重投），处理逻辑必须保证同一个事件执行两遍和执行一遍效果相同——就像服务员对同一个铃铛不应两次端同一道菜。第二，补偿机制，当某个事件处理失败时，通过反向操作抵消已经发生的影响。延迟窗口不是系统缺陷，而是分布式世界为了性能和可用性付出的合理代价。

老顾客都懂美味斋的脾气：菜是等出来的，急不得。系统设计者也要管理好这份预期——客户要的是“最终能吃到菜”，而不是“点单的瞬间菜就摆在桌上”。

Saga 模式

有个顾客点了一份“全家福套餐”：红烧肉、清蒸鱼、酸辣汤。这三道菜分别由三位厨师负责。厨师 A 红烧肉做好了，厨师 B 清蒸鱼也上蒸锅了，可厨师 C 一看冰箱——酸辣汤的料没备齐！此时怎么办？老板一声令下：红烧肉撤回来放凉菜间，清蒸鱼也先别上，整套订单取消。这道“撤菜”的动作，就是补偿（Compensation）。

在分布式系统中，一个业务操作往往跨越多个独立服务，每个服务有自己独立的数据库，无法用传统数据库事务保证原子性。Saga 模式应运而生：把一个大型事务拆分成一串本地小事务，每个小事务都有对应的补偿操作。正常流程依次执行；一旦某一步失败，就沿着相反方向依次触发补偿，把已完成的步骤“撤销”回滚。

Saga 有两种组织方式：编排式（Choreography）让各服务通过事件自行接力，没有中央指挥，好比后厨里各厨师听见铃声自行配合，灵活但难追踪；协同式（Orchestration）则由一个 Saga 协调者发号施令，明确告诉每一步该做什么、失败后启动哪条补偿链，好比老板亲自盯流程，清晰可控但多了一个中心节点。无论哪种方式，Saga 都不能保证“绝对同时成功”，而是保证“要么全部成功，要么通过补偿回到原点”。

套餐最终没有上成，顾客有点失望，但绝不会吃到“只有红烧肉没有汤”的残缺套餐。Saga 用一系列“撤菜”动作，守住了分布式世界里那份难得的整体正确。

AI 应用场景

Python 演示

#!/usr/bin/env python3"""CQRS + 事件溯源 + 投影 + Saga 的最小模拟""" class EventStore:    """事件存储：只追加，不可变"""    def __init__(self):        self.events = []     def append(self, event):        self.events.append(event)     def replay(self):        return list(self.events)  class OrderProjection:    """读模型投影：把事件折叠成订单视图"""    def __init__(self):        self.orders = {}     def apply(self, event):        etype = event["type"]        if etype == "OrderCreated":            self.orders[event["order_id"]] = {                "items": event["items"], "user": event["user"], "status": "CREATED"            }        elif etype == "OrderApproved":            self.orders[event["order_id"]]["status"] = "APPROVED"        elif etype == "OrderCancelled":            self.orders[event["order_id"]]["status"] = "CANCELLED"     def get(self, order_id):        return self.orders.get(order_id)  class Saga:    """Saga 编排：下单 -> 扣库存 -> 扣款，失败则补偿取消"""    def __init__(self, store, projection):        self.store = store        self.projection = projection     def start(self, order_id, items, user):        self.store.append({"type": "OrderCreated", "order_id": order_id,                           "items": items, "user": user})        self.projection.apply(self.store.events[-1])        try:            self._reserve_stock(order_id, items)            self._charge(user, 100)        except RuntimeError as e:            print("[失败]", e)            self._cancel(order_id)            return        self.store.append({"type": "OrderApproved", "order_id": order_id})        self.projection.apply(self.store.events[-1])        print("[成功]", order_id, "approved")     def _reserve_stock(self, order_id, items):        if order_id == "bad":            raise RuntimeError("库存不足")     def _charge(self, user, amount):        if user == "bad":            raise RuntimeError("余额不足")     def _cancel(self, order_id):        self.store.append({"type": "OrderCancelled", "order_id": order_id})        self.projection.apply(self.store.events[-1])        print("[补偿] 已取消订单并释放库存", order_id)  if __name__ == "__main__":    store = EventStore()    proj = OrderProjection()    saga = Saga(store, proj)     saga.start("A001", ["红烧肉", "清蒸鱼"], "小明")    print("读模型查询:", proj.get("A001"))     saga.start("bad", ["酸辣汤"], "小红")    print("读模型查询:", proj.get("bad"))    print("事件日志条数:", len(store.replay()))

1. self.store.append(event)事件存储只允许追加，天然留下完整审计日志。
2. self.projection.apply(self.store.events[-1])投影消费每个新事件，折叠进读模型。
3. self.orders[event["order_id"]]["status"] = "APPROVED"读模型只维护查询友好的现状。
4. try ... except RuntimeError捕获本地事务失败，进入 Saga 补偿分支。
5. self._cancel(order_id)补偿操作追加“订单已取消”事件，让状态回到业务可接受的终点。

常见误区

面试题