6.7 KiB
tags, create time
| tags | create time | |||||
|---|---|---|---|---|---|---|
|
2026-05-24 19:52 |
MQ CQRS 与 Event Sourcing
概述
CQRS（Command Query Responsibility Segregation）将读写模型分离，Event Sourcing 用事件流代替状态快照。两者结合并通过 MQ 广播事件，可以构建高性能、可审计、可追溯的系统架构。本文详解这两个模式的原理、组合方式、与传统 CRUD 的对比，以及事件存储的核心设计。
正文
CQRS 概述
传统 CRUD 模式下，同一个数据模型既用于写入也用于查询。这在简单场景下工作良好，但随着业务复杂化，读写的性能需求和模型结构会逐渐分化。
CQRS 的核心思想：Command（写）和 Query（读）使用不同的模型。
- 写模型（Command Side）：专注于业务规则校验和状态变更，使用领域模型（Domain Model），可以高度规范化。
- 读模型（Query Side）：专注于查询性能，使用反规范化的视图模型（View Model），可以针对不同查询场景定制。
graph LR
Client["客户端"] --> CmdAPI["Command API"]
Client --> QueryAPI["Query API"]
CmdAPI --> WriteDB["写模型 (Domain)"]
WriteDB -->|"事件通过 MQ 广播"| MQ["Message Queue"]
MQ --> ReadModel1["读模型 A (列表视图)"]
MQ --> ReadModel2["读模型 B (统计视图)"]
ReadModel1 --> QueryAPI
ReadModel2 --> QueryAPI
style Client fill:#4A90D9,color:#fff
style CmdAPI fill:#F5A623,color:#fff
style QueryAPI fill:#6EC1E0,color:#fff
style WriteDB fill:#D0021B,color:#fff
style MQ fill:#F5A623,color:#fff
style ReadModel1 fill:#6EC1E0,color:#fff
style ReadModel2 fill:#6EC1E0,color:#fff
Event Sourcing
传统方式存储数据的"当前状态"——每次更新都是覆盖写入。Event Sourcing 反其道而行：不存储当前状态，存储所有导致状态变更的事件。
以电商订单为例：
| 传统 CRUD | Event Sourcing |
|---|---|
UPDATE orders SET status='paid' |
追加事件 OrderPaid{orderId, amount, time} |
| 只保留最终状态 | 保留完整的事件历史 |
| 无法追溯"为什么变成这样" | 可以重放任意时间点的状态 |
Event Sourcing 的优势：
- 完整审计追踪：每个状态变更都有记录，满足金融、医疗等合规要求。
- 时间旅行：通过重放事件，可以重建任意历史时刻的状态。
- 调试友好：出 bug 时可以精确回放复现问题。
Event Sourcing 的挑战：
- 事件版本兼容：事件 Schema 演进时需要保证新旧版本兼容。
- 查询复杂：获取当前状态需要重放所有事件（可用 Snapshot 优化）。
- 最终一致性：读模型通过 MQ 异步更新，存在短暂延迟。
[!question] Event Sourcing 让历史可追溯，但也带来事件版本兼容的挑战。你会如何处理事件 Schema 的演进？
CQRS + Event Sourcing 的组合
CQRS 和 Event Sourcing 天然互补：
- Command Side 采用 Event Sourcing，将状态变更以事件形式追加到 Event Store。
- Event Store 通过 MQ 将事件广播给所有关心的消费者。
- Query Side（Read Model）消费事件，构建针对查询优化的视图。
这种架构下，MQ 是连接读写两侧的桥梁——写入端产生事件，MQ 负责分发，读取端消费事件并更新视图。
与传统 CRUD 的对比
| 维度 | 传统 CRUD | CQRS + Event Sourcing |
|---|---|---|
| 数据一致性 | 强一致（同一数据库） | 最终一致（读模型异步更新） |
| 查询性能 | 受限于写模型结构 | 读模型可针对查询优化 |
| 审计追踪 | 需要额外设计审计表 | 天然支持，事件即审计日志 |
| 扩展性 | 读写耦合，难以独立扩展 | 读写独立扩展 |
| 复杂度 | 低 | 高（事件设计、版本管理、最终一致性） |
| 适用场景 | 简单 CRUD 应用 | 复杂业务、高并发读、审计要求高 |
[!question] CQRS 带来最终一致性，用户下单后立刻查询可能看不到最新状态。在你的业务场景中，这种延迟可以接受吗？如何在 UX 层面缓解？
事件存储设计
事件版本
事件一旦写入就不可修改（Immutable），但业务会演进。处理方式：
- Upcast：读取旧版本事件时，在内存中转换为新版本。
- 多版本共存：事件中携带版本号，消费者根据版本号分别处理。
快照（Snapshot）
当事件数量很大时，每次重放全部事件来获取当前状态会很慢。Snapshot 在特定时间点保存状态快照，重放时只需从最近的 Snapshot 开始。
事件重放
重放是 Event Sourcing 的核心能力——从头（或从 Snapshot）依次应用事件，重建状态。读模型的重建、bug 修复后的数据修复、新读模型的初始化，都依赖事件重放。
Go 代码：简化版 Event Sourcing
package main
import "time"
// Event 事件定义：每个事件代表一次状态变更
type Event struct {
ID string
AggregateID string // 聚合根 ID（如订单 ID）
Type string // 事件类型
Data []byte // 事件数据
Version int // 事件版本号
Timestamp time.Time
}
// EventStore 事件存储接口
type EventStore interface {
Save(events []Event) error
Load(aggregateID string) ([]Event, error)
}
// OrderAggregate 订单聚合根
type OrderAggregate struct {
ID string
Status string
Amount float64
}
// Apply 将单个事件应用到聚合根，更新状态
func (o *OrderAggregate) Apply(event Event) {
switch event.Type {
case "OrderCreated":
o.Status = "created"
case "OrderPaid":
o.Status = "paid"
case "OrderShipped":
o.Status = "shipped"
}
}
// ReplayEvents 从事件流重建聚合根状态
func ReplayEvents(aggregateID string, store EventStore) (*OrderAggregate, error) {
events, err := store.Load(aggregateID)
if err != nil {
return nil, err
}
agg := &OrderAggregate{ID: aggregateID}
for _, event := range events {
agg.Apply(event) // 依次应用每个事件
}
return agg, nil
}
核心逻辑：ReplayEvents 从 Event Store 加载指定聚合根的所有事件，依次调用 Apply 重建当前状态。这就是 Event Sourcing 的精髓——状态是事件的函数，State = f(events)。