先交代一下背景。我维护的一个内部业务系统里最核心的订单表每天都在被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 行核心你可能要问有现成的中间件和框架可以用为什么还要自己写我确实认真对比过也不是没考虑过引入现成方案。但放在这个内部小系统面前每个方案都有自己的重量。这里做一个简单对比方案写入一致性审计能力部署复杂度适合场景直接改状态表最强单表事务几乎为零最低简单系统、低频数据消息中间件 状态存储依赖事务性消息一般需要额外消费端高引入独立集群跨服务事件流转开源事件存储类产品强强中高一般要求独立部署独立的事件溯源基础设施自研 REA应用层事件表强利用同一数据库事务强完整事件链最低一个表两根索引单服务内的可追溯状态管理最终我选了关系数据库 应用层事件表的路线。理由很直接我的业务系统本来就是一个单体应用所有需要审计的数据都在同一个数据库里。把事件表和业务表放在同一个事务里写入天然就能保证写业务事件和改状态的一致性。不必引入额外的消息队列不必处理网络分区也不必为了一个边界清晰的需求去学习一套新协议。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 的排序铁律是事件重放顺序只看seqcreated_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(funknown 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 的投影器采用增量模式核心步骤只有四步查询events表从last_processed_seq 1开始按seq顺序读取逐条应用事件到投影表处理完后把最新的事件序号写回projection_offsets表如果某条事件处理失败序号不推进下次启动时会从失败位置重新读取。对应的进度表结构很简单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 开始。我的快照频率是这样定的事件总量快照策略 500不需要500 ~ 5000每 1000 条事件做一次 5000每 5000 条事件做一次快照表的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记为 1payload 里放当时订单的完整状态快照。从这一刻起新产生的业务变更都走正规事件流seq从 2 开始递增。投影器处理订单时先加载基线事件把初始状态恢复出来再处理后续增量事件。这个方案成本很低但效果立竿见影存量数据可查、增量数据可追溯新旧两套逻辑在同一个事件流体系下做到了无缝衔接。5. REA 上线后的复盘与想继续做的事5.1 同样的事故处理效率从几小时降到几分钟REA 上线三个月后真的又发生了一次类似的状态误改事故。这一次我打开事件表执行了一条再简单不过的 SQLSELECT 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 用了半年后我最想推荐的一个设计。