自研REA：用事件溯源实现业务状态可追溯与审计

先交代一下背景。我维护的一个内部业务系统里，最核心的订单表每天都在被UPDATE status反复改写。某天出过一次事故：一段促销脚本因为 WHERE 条件没有加括号，导致条件判断完全失真，把一批已付款订单的状态直接刷成了已取消。最难受的不是恢复数据，而是根本不知道是谁、在什么时间、因为什么逻辑改的——状态被覆盖之后，只留下一个无痕的结果。

之后我花了一个周末，写了一个叫rea的小工具。名字没什么高深含义，就是我在终端里顺手敲出来的三个字母，后来才给他补了个全称：Real-time Event Archive。核心思路很简单：不再直接更新业务表的状态列，而是把所有业务变更先写进一张有序、不可变的事件表，业务当前状态全部由投影器从事件流中计算出来。这篇文章不聊大道理，只讲我从建模到上线、再到踩坑修复的完整过程，希望能给也想折腾事件溯源思路的人一个可参考的最小实现。

1. 为什么我放着现成框架不用，非要写一个叫 REA 的东西

1.1 一行业务 UPDATE 引发的数据事故

当时出事的脚本大概长这样：

UPDATE orders SET status = 'cancelled' WHERE created_at < '2024-06-01' AND user_id = 123;

我拿这个当例子不是为了讲 SQL 语法，而是想说：这种直接改状态表的写法，天然就有两个致命问题。第一，状态被覆盖之后，原来的值去哪了？没人知道。第二，谁改的、什么时候改的、为什么改，都无从追溯。数据库里只剩一个孤零零的cancelled。

事故发生后，我们只能靠业务日志和数据库 binlog 反推数据变化。又慢又痛苦，而且同类问题隔两周又会以另一种形式出现：运营手填了一个状态、接口重复调用导致状态覆盖、某个后台任务把全部单据批量重置……每一次都要从头查一遍。

后来我在复盘时意识到：如果这个订单对象的核心数据不是"一个可变的状态字段"，而是一串"只追加、不可改的事件记录"，那所有问题都会被瓦解掉。比如OrderPlaced、PaymentReceived、OrderShipped按顺序老老实实躺在表里，想审计就查事件列表，想恢复某个时点状态就从起点重放到那个时点。这就是事件溯源最基本的价值。

1.2 现成方案对比：与其接一堆依赖，不如写 500 行核心

你可能要问：有现成的中间件和框架可以用，为什么还要自己写？

我确实认真对比过，也不是没考虑过引入现成方案。但放在这个内部小系统面前，每个方案都有自己的重量。这里做一个简单对比：

最终我选了"关系数据库 + 应用层事件表"的路线。理由很直接：我的业务系统本来就是一个单体应用，所有需要审计的数据都在同一个数据库里。把事件表和业务表放在同一个事务里写入，天然就能保证"写业务事件"和"改状态"的一致性。不必引入额外的消息队列，不必处理网络分区，也不必为了一个边界清晰的需求去学习一套新协议。

1.3 REA 的定位与目标边界

REA 从一开始就不是奔着"通用事件溯源框架"去的。他的边界很明确：

• 只解决单服务内的状态可追溯；
• 不跨服务传递消息；
• 不做完整 CQRS，查询和写入仍然共用一个库；
• 不处理分布式事务。

听起来限制很多，但正是这些限制让我能在一个周末里把核心逻辑跑通。REA 解决的实际问题是：给系统增加一条可靠、可审计、可回放的状态变化记录链，并基于这条链维护出若干套可查询的投影视图。

如果你也想在内部项目中引入类似思路，我的建议是先不要过度设计。事件溯源的核心不是"把系统全部改造成事件驱动"，而是"把关键业务对象的变化过程保留下来"。

2. 事件溯源的核心建模：先分清"事实"和"状态"

2.1 业务表直接存状态的三个隐患

在讲 REA 的表结构之前，我想先聊清楚一个建模层面的问题：为什么"状态"不应该作为唯一事实？

第一个隐患是信息丢失。状态列只保留最后的结果，中间发生了什么全部被抹掉。第二个隐患是旁路修改没有约束。只要是能连上数据库的人，都能随手UPDATE，而且绕过所有业务校验。第三个隐患是统计口径漂移。比如你想算"每个订单从创建到发货的平均时长"，如果status字段会被反复覆盖，你根本拿不到可靠的"发货时间"。

有一个比喻我经常用：状态字段就像拍立得照片，只能看到按下快门那一刻的画面；事件流则像录像带，每一帧都记录着真实发生的过程。做业务系统久了你会发现，很多时候"过程"比"结果"值钱得多。

2.2 REA 的事件表结构

REA 落地时先建了一张事件表，结构如下：

CREATE TABLE events ( id BIGSERIAL PRIMARY KEY, aggregate_type TEXT NOT NULL, aggregate_id BIGINT NOT NULL, event_type TEXT NOT NULL, seq BIGINT NOT NULL, payload JSONB NOT NULL DEFAULT '{}', actor_id BIGINT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), UNIQUE (aggregate_type, aggregate_id, seq) ); CREATE INDEX idx_events_created_at ON events (created_at); CREATE INDEX idx_events_aggregate_lookup ON events (aggregate_type, aggregate_id, seq);

每一列都不是随便加的，我逐个解释：

• aggregate_type：标识业务对象的类型，比如order、payment、inventory。因为一张表要承载多个业务对象的事件，所以必须有这个区分字段。
• aggregate_id：对应业务对象的主键。比如订单 ID、支付单 ID。
• event_type：事件名，表示发生了什么，比如OrderPlaced、OrderShipped。
• seq：同一聚合对象内单调递增的序号，从 1 开始。这个字段是所有重放逻辑的基础，重要性我会在下一节专门讲。
• payload：事件的业务数据，用 JSONB 存储，灵活性最高。
• actor_id：记录操作人或者调用方身份，审计时最有用。
• created_at：事件产生的时间，主要给展示用，不用于排序。

为什么不用单独的表存储每个聚合对象的状态？因为对于 REA 来说，事件表本身就是唯一的权威数据源。状态表只是投影，随时可以由事件流重建。这张表也是整个系统里唯一"永远只插入、不更新、不删除"的业务表。

2.3 为什么 seq 比 created_at 更适合排序

这是我在设计时踩过的最深的一个坑，值得单独拿出来说。

很多人第一反应是用created_at给事件排序，毕竟直观。但在并发环境下，时间戳排序有两个致命问题：第一，两个事件可能在同一微秒产生，created_at完全一样；第二，服务器的时钟可能回拨，导致后发生的事件时间戳反而更早。

所以 REA 的排序铁律是：事件重放顺序只看seq，created_at只用于展示和统计。seq由数据库序列生成，配合UNIQUE (aggregate_type, aggregate_id, seq)约束，保证每个聚合对象内部的事件严格递增且不重复。

举个实际例子。订单创建后，用户先提交取消，随后运营又手动恢复发货。如果按created_at排序，这两个事件的时间戳完全可能相同，导致重放时顺序不确定，最终状态一会儿是cancelled、一会儿是shipped。但如果按seq重放，顺序永远固定，不会出现灵异状态。

2.4 事件 payload 的版本演进

业务系统不可能一成不变，事件里携带的字段也会调整。REA 的处理方法是在事件类型里带上版本号，或者给每个事件加一个event_version字段。比如OrderPlacedV1、OrderPlacedV2。

不要小看这一步。我见过不少项目上线半年后，业务要给事件加字段，结果直接改 JSON 结构，旧事件在解析时全部报错。REA 在投影器里用统一的解析入口处理版本转换：

def parse_order_payload(payload: dict, version: int) -> dict: if version == 1: return { "items": payload.get("items", []), "items_count": len(payload.get("items", [])), } if version == 2: return { "items": payload.get("item_ids", []), "items_count": payload.get("items_count", 0), } raise ValueError(f"unknown event version: {version}")

这种显式的版本分支可能看起来笨拙，但它解决了"老数据 + 新代码"的兼容问题，也让事件流的不可变性真正立得住。

3. 从事件流到业务状态：投影器的设计与实现

3.1 投影到底是什么？

事件表里存的只是"发生了什么"，但运营系统没法直接查事件表。比如"当前订单状态是什么""今天销售额是多少"这类问题，事件表无法回答。这时候就需要投影器。

投影器的概念可以理解为：从一段事件流出发，计算出一个可查询的视图。REA 里最典型的一个投影，是把某个订单的全部事件重放成订单当前状态：

def project_order(order_id: int): events = get_events("order", order_id) state = { "status": "created", "items": [], "total_amount": 0, } for evt in events: payload = parse_order_payload(evt.payload, evt.event_version) if evt.event_type == "OrderPlaced": state["status"] = "placed" state["items"] = payload["items"] state["total_amount"] = evt.payload["total_amount"] elif evt.event_type == "PaymentReceived": state["status"] = "paid" state["paid_at"] = evt.created_at elif evt.event_type == "OrderShipped": state["status"] = "shipped" state["shipped_at"] = evt.created_at elif evt.event_type == "OrderCancelled": state["status"] = "cancelled" state["cancelled_at"] = evt.created_at return state

投影器不是只能维护当前状态。REA 里我同时维护了两类投影：一类是业务当前状态表，用于日常读写；另一类是小时级统计表，用于运营看板。两类投影都来自同一条事件流，只是计算结果不同。

3.2 增量投影的工作流

每次有新事件就全量重放显然不现实。REA 的投影器采用增量模式，核心步骤只有四步：

1. 查询events表，从last_processed_seq + 1开始按seq顺序读取；
2. 逐条应用事件到投影表；
3. 处理完后，把最新的事件序号写回projection_offsets表；
4. 如果某条事件处理失败，序号不推进，下次启动时会从失败位置重新读取。

对应的进度表结构很简单：

CREATE TABLE projection_offsets ( projection_name TEXT PRIMARY KEY, last_processed_seq BIGINT NOT NULL, updated_at TIMESTAMPTZ NOT NULL DEFAULT now() );

投影器代码的核心是一个循环，我习惯用 Python 写原型：

def run_projection(projection_name: str): offset = get_offset(projection_name) events = fetch_events_after(offset) for evt in events: apply_event(projection_name, evt) set_offset(projection_name, evt.seq)

这里最关键的设计是"先处理事件，再记录进度"，且两者最好在同一个事务里完成。如果先记进度，再出错了，事件就会永久漏掉；如果先处理事件再记进度，出错了最多重复处理一次，配合幂等设计可以兜底。

3.3 快照策略：什么时候必须做

事件全部重放虽然简单，但事件量大了以后性能一定会下降。比如一个订单产生了几千个状态变更事件，每次查详情都要从第 1 个事件开始重放，响应时间会肉眼可见地变慢。

REA 用快照来解决这个问题。快照本质上就是某个时刻投影状态的序列化副本，记录中会额外保存当时的seq值。重建流程变成：先加载最近的一份快照，再从快照记录的下一个seq开始重放增量事件，而不是永远从 0 开始。

我的快照频率是这样定的：

快照表的seq字段是灵魂。没有它，加载快照后不知道从哪里继续重放；有了它，系统的重建路径就完整了。

3.4 并发写入与顺序保证

写入事件和更新投影表，一定要保证事务边界。REA 的做法是把"写入事件"和"更新投影表"放在同一个数据库事务里。

举个例子。用户支付成功后，需要写PaymentReceived事件，同时要把订单状态表里的状态改成paid。如果先写事件、事务未提交，投影侧就更新状态，一旦事务回滚，状态就错了。更好的做法是：

BEGIN; INSERT INTO events (aggregate_type, aggregate_id, event_type, seq, payload, actor_id) VALUES ('order', 10086, 'PaymentReceived', 3, '{"amount": 9980}', 42); UPDATE order_projection SET status = 'paid', paid_at = now() WHERE order_id = 10086; COMMIT;

这两步在同一个事务里，要么都成功，要么都回滚，状态永远跟事件保持一致。

4. 踩坑实录：REA 上线后我遇到的四个"意料之外"

4.1 第一个坑：时间戳不能用来排序

这个坑前面提了一嘴，但我想把实际场景完整还原一下。

REA 上线第二周，运营反馈某订单状态总是跳变：明明已经发货了，好端端又变成已取消；过一阵再刷新又变回已发货。我查了事件数据，发现OrderShipped和OrderCancelled两个事件的created_at时间戳完全一样。

原因很简单：两个请求并发执行，业务代码各自取了当前时间写事件。因为服务器时钟精度问题，两个时间戳撞了。而我的投影器一开始是按created_at排序读取事件，导致同一时间戳下事件顺序不确定，重放结果时好时坏。

换成按seq排序后，问题彻底消失。经验就是：在事件表设计里，created_at只服务于"给人看"，所有程序逻辑一律用seq做主键。

4.2 第二个坑：幂等没有做，重放导致金额翻倍

增量投影器为了防止漏处理事件，采取了"出错就重试"的策略。这本身没错，但我漏了幂等保护。

事故场景是：一次统计投影处理某个事件时，数据库连接超时，事务回滚了。重启后投影器从失败处重新读取，这次连接恢复正常，事件被正常处理。听起来没问题是吧？问题出在第一次事务并没有真正提交，但统计逻辑已经把金额累加到了内存变量里，代码在事务提交前就把累计值写到了另外一张缓存表。最后结果是同一笔订单金额被统计了两次。

修复方法很简单，给统计投影表加一个事件来源唯一标识：

ALTER TABLE hourly_sales_projection ADD COLUMN source_event_id BIGINT UNIQUE;

写入时带上事件 ID，重复处理时靠唯一约束直接跳过。这个坑给我的教训是：任何基于事件流的重放逻辑，都要默认"同一个事件可能被处理两次"来设计，幂等不是可选优化，是必要条件。

4.3 第三个坑：事件 payload 结构演变时，老数据解析失败

项目跑了一个季度后，运营要求订单创建事件里除了商品列表，还要加一个件数汇总。我图省事，直接在OrderPlaced事件里改 payload 结构，旧的 payload 是items数组，新的 payload 变成了item_ids加items_count。

结果就是：重放老订单时，投影器读取items_count直接报 KeyError，整个投影进程卡死。

后来我统一改为带版本号的解析，就像第 2.4 节那样。事件类型可以叫OrderPlacedV2，也可以在事件表里加一个event_version字段。不管哪种方式，核心原则是：旧事件必须能被新代码安全解析，绝不能因为业务演进让历史数据变脏。

4.4 第四个坑：与存量表共存时的"楚河汉界"

REA 没法直接把线上正在跑的旧系统一刀切换掉。当时存量订单已经有几百万条，不可能全部通过INSERT事件补回来。

我的做法是"基线事件"。

在开始启用 REA 的时间点，先为每一个存量订单写一条BaselineCreated事件，seq记为 1，payload 里放当时订单的完整状态快照。从这一刻起，新产生的业务变更都走正规事件流，seq从 2 开始递增。投影器处理订单时，先加载基线事件把初始状态恢复出来，再处理后续增量事件。

这个方案成本很低，但效果立竿见影：存量数据可查、增量数据可追溯，新旧两套逻辑在同一个事件流体系下做到了无缝衔接。

5. REA 上线后的复盘与想继续做的事

5.1 同样的事故，处理效率从几小时降到几分钟

REA 上线三个月后，真的又发生了一次类似的状态误改事故。这一次，我打开事件表执行了一条再简单不过的 SQL：

SELECT id, seq, event_type, actor_id, created_at FROM events WHERE aggregate_type = 'order' AND aggregate_id = 10086 ORDER BY seq;

所有变更过程整整齐齐列出来，谁操作的、什么时间、改了什么都一目了然。定位问题只用了不到十分钟。除此之外，REA 还顺手解决了另一个问题：以前想做销售统计要写临时 SQL，现在小时级统计投影器始终在维护，运营看板直接查投影表就行。

一个为审计而生的设计，最终白赚了统计能力。

5.2 值得继续扩展的三个方向

如果后面还有精力，我会往三个方向延伸：

第一，给投影器加状态机约束。比如cancelled状态不允许再流转到shipped，在写事件之前先校验当前聚合状态是否允许该事件发生，非法流转直接拒绝。第二，把统计类投影从主库拆到只读副本，实现读模型和写模型的进一步分离，避免大查询拖累主库。第三，做一个粗粒度的后台时间线页面，把同一个订单的所有事件按seq渲染成可视化的纵向时间线，运营排查问题时就不再需要找我查表了。

至于跨服务事件总线，我大概率不会往里加。REA 的价值恰恰来自他的小而克制，一旦变成又一个大而全的平台，就又回到了老路上。

5.3 最后分享一个小技巧

如果你也要自己实现投影器，我特别建议给每个投影表加一个projection_version字段。当投影逻辑发生变化时，不要原地改旧记录，而是把这个字段值加一，然后写一个迁移函数，对存量行按新逻辑重新计算。

这听起来多了一步，但实际帮你省掉大量麻烦。因为投影器的核心价值就是"可重建"，既然能从事件流重建，旧版本逻辑的产物就不应该被手工修补，而是应该通过版本升级整体重算。这是 REA 用了半年后，我最想推荐的一个设计。