CQRS 读写分离与事件投影

可视化演示

将命令与查询分离到独立模型，经事件投影构建读模型，实现读写独立扩展与最终一致。

起源与定义：从混沌单间到双子影城

老式电影院的困境，正是传统软件架构的日常。一个订单模块里，下单、改状态、校验库存这些“写操作”，和查列表、看详情、跑报表这些“读操作”共用同一个实体模型、同一张表、同一把锁。写流量大时，慢查询把数据库拖垮；读流量大时，事务提交被阻塞。就像一个剧组既要拍戏又要现场售票，两边都做不好。

传统 CRUD 系统使用同一个领域模型同时承担读写职责。但读与写有着截然不同的诉求：写关注业务规则和不变量，读关注查询速度和形态。Greg Young 在 2010 年前后将 CQRS 系统化：把系统拆成两条通道——写侧用命令（Command）表达“希望发生的意图”，读侧用查询（Query）读取投影后的视图，两者使用不同的模型、不同的存储、不同的优化策略。

导演从此专心在片场拍戏，观众在放映厅舒服观影。两条通道各自演进、各自扩容，这就是 CQRS 的起点。

命令端设计：导演的拍摄指挥部

导演身边有一位尽职的场记助理。助理收到“演员就位”的指令后，先核对剧本（校验规则），再喊“开拍”，演员按剧本执行，摄像机记录下真实发生的一切。这里，助理就是命令处理器，演员就是聚合对象，摄像机记录下来的素材就是领域事件。

命令端是 CQRS 的写半边。命令对象是不可变的意图描述，比如 PlaceOrderCommand(userId, items)。命令处理器（Command Handler）负责从仓库加载聚合、调用业务方法、校验业务不变量，最终让聚合产生一个领域事件。这个事件被追加到事件存储中，代表“已经发生的事实”——不是命令的希望，而是岁月的见证。

现场拍完的镜头被完整归档，场记在台本上写下“本镜已拍”。没有人需要立刻跑到放映厅去手动改银幕——那是投影器的工作。

读模型构建：观众看到的银幕

观众不会去片场看摄像机，他们看的是售票台、场次屏、座位图——每一个都是为特定查询精心准备的信息形态。同一部电影可以投影成海报、预告片、影评榜，它们来自同一份原片，却以各自适合的样子呈现在不同入口。

读模型是专门为查询而设计的模型，通常以反规范化（denormalized）的形式存在。查询“我的订单”页面需要用户名、订单号、商品列表、状态、金额，传统架构要 JOIN 三张表才能查全；而读模型在事件投影时就把这些字段物化成一张宽表。CQRS 允许甚至鼓励读模型与写模型使用完全不同的存储：写库用 MySQL 保证事务，读库用 Elasticsearch 或 Redis 追求性能。

影院的每个入口都挂着不同的信息牌：给观众看排片表，给财务看票房日报。读模型的“多形态”，正是 CQRS 应对复杂查询需求的优雅姿势。

事件投影机制：放映机的魔法

真正的魔法藏在放映室。放映员把母带装进放映机，一帧帧投到银幕上。如果银幕布旧了要换新的，放映员只需要把母带从头再放一遍——投影可以无限重放，不必重新拍电影。CQRS 的读模型也是从事件流里“投影”出来的，那个负责投影的放映员，叫投影器（Projector）。

投影器订阅事件流，每收到一个事件（比如 OrderPlaced），就更新一次读模型。它需要维护一个进度位置（position/checkpoint），记录自己已经处理到第几个事件。当读模型需要重建时，投影器从位置 0 开始重放全部历史事件。为了提升效率，系统每隔一段时间打一个快照——相当于把胶片某几卷的“剧情摘要”存档，重建时从最近的快照开始，而不必从头逐帧放映。

放映机不会为了换幕布重拍电影，它只需要重放胶片。投影器同理——事件流就是那卷永不失真的母带。

最终一致性：胶片需要冲洗时间

导演喊完“卡”，观众不可能瞬间看到这个镜头——胶片还要冲洗、剪辑、拷贝、配送。观众看到的画面总是“几分钟前的事实”，但故事不会因此出错。CQRS 中的读模型也有类似的“冲洗时间”，这就是最终一致性。

命令提交成功后，事件进入日志，但投影器可能还没处理到那条事件，读模型短暂滞后。如果用户刚下单就立刻刷新“我的订单”，可能会看不到新订单——这就是“读己之写”问题。解决思路包括：写后重定向到详情页并等待、对关键路径采用同步投影、客户端做短暂重试或轮询。更重要的保障是幂等（idempotent）：同一事件被重复应用时不能重复加余额，投影函数必须以事件 ID 或版本号做去重。

观众永远不会看到“穿帮的瞬间”——导演喊停后几分钟，银幕上才出现刚才那个镜头。银幕反映的是真实发生过的事，只是略微迟到，不是编造的事。

同步与重放：影院调度与断点续播

影院会根据影片热度安排上映策略：首映夜也许用卫星“实时直播”（同步投影），日常则靠实体拷贝配送（异步投影）。放映员还要在每本胶片上做标记，一旦停电，换好灯泡后从标记处继续放——这就是检查点（checkpoint）的妙处。

投影器的同步策略分为两类：同步投影在命令事务内同时更新读模型，读侧立即可见，但写延迟增加，适合极小规模的低延迟场景；异步投影让事件发布后由投影器在后台消费，读写彻底解耦，是主流选择。异步模式下，消息中间件的持久化订阅（Persistent Subscription）会为每个投影器记录消费位置。投影器崩溃重启后，从断点位置继续消费，而不是从头再来。

放映员在换片时会记住“放到第几本盒”。哪怕放映机烧了灯泡，换好之后接着那盒继续放——不会傻到把整卷胶片从头再放一遍。

工程落地实践：从单厅影院到院线集团

从单厅影院到院线集团，并不是每个厅都需要上 IMAX。爆米花柜台用普通 POS 机就好，旗舰厅才值得上 CQRS——落地要看场景。

CQRS 适合：读多写少、查询形态复杂、报表分析拖垮交易库、需要审计回放的系统。不适合：简单 CRUD、对强一致要求苛刻且无法接受异步延迟的系统。它与 Event Sourcing（事件溯源，简称 ES）是正交概念——CQRS 可以不依赖 ES，直接命令后更新读库；ES 也可以不配 CQRS。两者搭配最顺：ES 提供不可变事件流，CQRS 把事件流投影成多样读模型。此外，监控的核心指标是投影延迟（Lag）：最新事件序号减去投影器已处理位置，这个差值就是运维大屏上最该跳动的数字。

院线集团的中控室墙上挂着所有影厅的放映状态大屏，投影延迟就是那块跳动的数字——它告诉你哪块银幕掉队了，以及整个放映体系是否健康。

AI 应用场景

Python 演示

# -*- coding: utf-8 -*-"""CQRS 读写分离与事件投影的最小演示"""  class BankAccount:    """写模型:聚合根"""     def __init__(self, account_id):        self.account_id = account_id        self.balance = 0        self.handled = 0     def apply(self, event):        if event["type"] == "deposited":            self.balance += event["amount"]        elif event["type"] == "withdrawn":            self.balance -= event["amount"]        self.handled += 1     def handle(self, command, event_store):        if command["op"] == "deposit" and command["amount"] > 0:            event = {"type": "deposited", "account": self.account_id,                     "amount": command["amount"]}            event_store.append(event)            self.apply(event)        elif command["op"] == "withdraw" and command["amount"] <= self.balance:            event = {"type": "withdrawn", "account": self.account_id,                     "amount": command["amount"]}            event_store.append(event)            self.apply(event)        else:            raise ValueError("invalid command")  class AccountProjection:    """投影器:把事件流变成读模型"""     def __init__(self):        self._position = 0        self.views = {}     def _load(self, account_id):        if account_id not in self.views:            self.views[account_id] = {"balance": 0, "count": 0}        return self.views[account_id]     def project(self, event):        view = self._load(event["account"])        if event["type"] == "deposited":            view["balance"] += event["amount"]        elif event["type"] == "withdrawn":            view["balance"] -= event["amount"]        view["count"] += 1        self._position += 1     def replay(self, event_store):        self.views.clear()        self._position = 0        for event in event_store:            self.project(event)     @property    def position(self):        return self._position  def main():    event_store = []    account = BankAccount("acc-001")    account.handle({"op": "deposit", "amount": 1000}, event_store)    account.handle({"op": "withdraw", "amount": 300}, event_store)     projection = AccountProjection()    print("命令已提交但尚未投影", projection.views)    projection.replay(event_store)    print("投影完成后的读模型", projection.views["acc-001"])    print("投影器位置", projection.position)  if __name__ == "__main__":    main()

1. event_store.append(event)把事件追加到事件存储，而不是直接覆盖余额字段，这就是“记录事实”。
2. self.apply(event)聚合通过 apply 事件恢复自身状态，这是 Event Sourcing 的“从事件重建状态”。
3. projection.replay(event_store)投影器全量重放事件流，重建读模型视图。
4. view["balance"] += event["amount"]投影函数把单个事件转成对读模型的一次更新。
5. self._position += 1投影器记录已处理位置，支撑断点续传与延迟监控。

常见误区

面试题