REA 是我最近大半年一直在折腾的一个轻量级事件编排引擎英文全名我习惯写成 Reliable Event Automation。说到底它解决的就是一件事把业务系统里“某个事件发生后还要依次做一大堆后续动作”的逻辑从散落的代码里收口成一张可配置、可重试、可观测的执行图。我先说下背景。我自己的项目当时已经不算小了十几个内部模块互相调用订单一创建下游要处理库存、通知、积分、优惠券、埋点光这一条链路就横跨了五个模块。最开始大家都是在各自的服务里监听消息、各自写定时任务补偿时间一长就会出现同一类事件被两个模块分别消费、补偿逻辑互相打架的情况。每次业务方提一句“能不能在支付成功后再增加一步”都要排查半天到底改哪里合适。这就是 REA 最初出现的动机。这篇文章不打算讲什么高大上的架构我会把设计思路、核心实现、以及我在实际排查中遇到过的坑都拆开聊一遍。如果你也在维护一套业务复杂度不低的后端系统正在被异步逻辑、定时任务和重试补偿折磨这篇文章应该能给你一些可以直接落地的参考。1. 为什么我会去做一个叫 REA 的项目1.1 我遇到的真实痛点要理解 REA 为什么长成现在这个样子得先还原我当时的工作场景。假设用户下单成功核心订单系统会触发一个事件然后需要做这些事扣减库存、生成优惠券、发送站内信、给用户累计积分、写入行为分析埋点。这些动作本质上没有严格的先后依赖但业务上希望它们都能完成任何一个失败都需要重试补偿。没引入 REA 之前团队里的做法是各做各的。库存模块自己监听 MQ收到消息后扣减库存失败就本地存表等定时任务扫描积分服务也监听同一份 MQ消费失败直接丢弃通知服务干脆在下单接口里同步调用……这个局面下最难以忍受的不是代码丑而是没有一个地方能回答“这个订单当前处理到哪一步了”这个问题。业务方问十万个为什么我们只能去翻各个服务的日志。更麻烦的是多模块重复消费。因为大家都在监听同一份消息一旦某个消息被重复发送这在 MQ 语义下很常见库存可能扣两次积分可能加两次。大家各自做幂等但每一家的幂等规则还不一样统一排查难度极大。所以我的核心诉求其实不复杂有一个统一入口接收事件能定义一条流程每个步骤是什么每一步要能单独重试跑完之后能清楚看到结果和耗时。这就是 REA 的雏形。它不是为了做一个通用平台而是为了把我自己的“事后补偿地狱”管起来。1.2 为什么不用现成的工作流引擎决定自研之前我认真调研过一些开源工作流引擎。功能确实强大支持人工审批、复杂条件分支、定时触发、可视化配置文档也不少。但我思考后还是放弃了原因主要有几个一是部署成本。引擎本身要独立部署自带的控制台又要占一套资源我的团队没有专门的运维精力去维护这类基础组件出了问题很难快速定位。二是学习成本。业务方和后端都想在可视化界面上拖拖拽拽但实际用起来复杂状态机的维护比写代码麻烦多了遇到边界情况你还得看引擎源码。对一个十人左右、要快速交付业务的小组来说这是很大的隐性成本。三是集成成本。现成引擎往往自带一套数据表和执行模型想接入现有业务代码、让它们共用一套事务边界要做非常多的适配。当时我们已经有大量的业务表和补偿任务强行迁移可能造成线上回归。REA 的定位因此非常明确嵌入式的、轻量的、SDK 化的编排能力而不是一套独立平台。我可以保留数据库里已有的数据表只新增几张运行表可以让业务方直接用代码写每个 Action 点。先解决一个问题事件进来之后后续动作由谁统一接管。1.3 REA 的三个目标可靠、事件驱动、自动化名字是我后来补的。Reliable Event Automation可以拆成三层理解Reliable可靠不丢事件、不重复执行、失败可追溯、支持指数退避重试Event事件驱动不依赖定时轮询去猜“有没有业务变化”而是由业务主动上报“发生了什么”Automation自动化一条事件进来后编排器自动完成剩余动作不需要人为干预。这三个目标直接对应了最终能力表。我用一个表格把它们对齐设计目标对应能力核心实现手段可靠事件不丢失、流程不重复执行事件落库、状态机 CAS可靠失败可重试、有死信兜底指数退避重试、最终失败告警事件驱动统一事件入口、按事件类型路由triggerEvent 与 Workflow 绑定自动化串行/并发执行 Action调度器按流程定义逐级执行可观测每一步执行明细可查Action 日志表记录输入输出摘要这张表在后面设计数据结构的时候起了很大作用。我每次犹豫要不要加新功能都会先对照这三个目标问自己它是为了让流程更可靠还是为了让更多人能编排如果不是就暂时不做。2. 核心细节解析与关键决策2.1 事件模型Event、Workflow、Action 三层抽象REA 的模型不算复杂我把它们分成三个层次Event事件业务侧上报的最小单位比如 order.created、payment.success、refund.closed。它只表达“发生了什么”不关心后续步骤。Workflow流程一个 Workflow 绑定一个触发事件类型内部包含一组有序的 Action 列表。Action动作流程里的最小执行单元可以是对内 HTTP 请求、本地方法调用、异步消息发送等。比较关键的一个设计是事件不能直接决定流程它只是触发的钥匙。同一种事件类型不同业务的流程可能不同所以还需要一个路由规则把 eventName 加业务标识比如支付渠道、用户等级映射到具体的 Workflow。这一点在后面的版本我才补上第一版是硬编码 eventName 到 Workflow 的一对一关系。一个流程定义长这样JSON 格式{ workflowName: orderNotifyFlow, triggerEvent: order.created, timeoutSeconds: 300, actions: [ { name: deductStock, type: http, config: { method: POST, url: http://internal-stock/inventory/deduct, timeoutMs: 3000 } }, { name: sendMessage, type: local, config: { beanClass: com.demo.NotificationSender, method: send } }, { name: grantPoints, type: http, config: { method: POST, url: http://internal-points/points/grant, timeoutMs: 5000 } } ] }这里重点不是 JSON 格式本身而是流程的生命周期。一个流程实例Instance会经历 PENDING → RUNNING → SUCCESS / FAILED / DEAD 这几个状态。开局必须 PENDING执行完毕后变成 SUCCESS如果某个 Action 重试最终失败整条流程进入 FAILED重试次数超过上限进入 DEAD也就是死信状态用来人工介入。2.2 同步串行还是并发执行第一版必须先简单写第一版的时候我面临一个选择Action 到底是串行好还是允许并发答案很明确第一版必须串行。原因在于并发会立刻引入三个问题一是乱序多个 Action 同时执行它们的完成顺序不确定有的步骤之间有隐含依赖二是一致性同时写流程状态和执行日志需要分布式锁复杂度陡增三是排查困难一旦失败你很难说清楚是哪个分支引起的。所以我把整个 Workflow 设计成一条顺序链一个 Action 完成后再执行下一个。不同 Workflow 之间并行Workflow 内部串行。这是我的稳妥策略。等到后面把并发需求整理清楚再针对特定节点加 parallel 配置而不是一开始就支持。实际上大多数异步流程的需求在串行模型下都能解决。所谓的“并发”更多是想节省总体耗时。但业务系统这几十毫秒的差距远没有失败可追踪重要。2.3 存储设计为什么用三张表而不是直接塞 Redis最初我也考虑过用 Redis 的 list 或者 stream 做事件队列理由当然是快。但一深入就发现了问题事件丢失风险。Redis 没有成熟的持久化语义如果宕机正好发生在写入之后、消费之前事件就没了可回溯性差。Redis 里面的数据不方便按时间、流程实例维度去查历史审计不满足。业务方经常需要看到“这个事件从哪来的执行了哪几步每步耗时多少”Redis 的简单数据类型很难支撑这种查询。所以我的最终方案是用 MySQL 来兜底所有流程数据Redis 只作为可选的高效缓存。表结构分三张work_event存初始事件字段包括 event_id、event_type、biz_key、payload JSON、create_time。work_instance存流程执行实例字段包括 instance_id、event_id、workflow_name、status、current_action_index、retry_count、next_retry_time、version、create_time、update_time。work_action_log存每个 Action 的执行明细字段包括 log_id、instance_id、action_name、status、request_summary、response_summary、error_msg、cost_ms、create_time。用三张表是为了把“入口信息”“流程状态”“执行明细”分开。工作里面排查问题时三个维度都方便单独查。比如业务方问“上周三有几个支付成功事件进入流程后失败了”我只需要在 work_event 表按事件类型和时间刷一下再把 instance_id join 到 work_instance 看状态最后查 action_log 就能定位是哪一步失败。这个方案牺牲了一点极致的性能换来了极致的可见性。对于一个内部编排组件来说数据可见性远比吞吐量重要。2.4 状态更新必须用 CAS不能用普通 UPDATE执行流程的时候调度器会扫描到一条 PENDING 状态的 instance然后准备执行。这里有一个很容易踩的大坑如果你只是用 statusRUNNING 去更新这条记录可能会发生多个调度线程同时处理同一个 instance。因为扫描和更新之间是有时间差的。扫描线程 A 查到一条 PENDING 记录正要执行另一个线程 B 也扫描到了同一记录也准备执行。两个线程同时执行同一个流程事件就会被重复处理。解决方法是 CASCompare And Swap更新UPDATE work_instance SET status RUNNING, version version 1, update_time NOW() WHERE instance_id ? AND status PENDING AND version ?这行 SQL 是 REA 里面最核心的一句。它保证只有一个线程能把状态从 PENDING 改成 RUNNING。如果影响行数为 0说明已经有别的线程抢先执行当前线程直接放弃。很多人会忽略 version 字段其实它是防止 ABA 问题的关键。有了 version你能确定工作是在预期的历史版本之上做的修改不至于拿旧状态覆盖新状态。3. 实操过程与核心环节实现3.1 最小可用版本要圈定边界REA 第一版我规定自己两个星期内上线。为了做到这一点我硬性砍掉了不少功能砍掉的东西以后再补砍掉可视化界面所有流程定义先写在本地配置文件里砍掉 Action 类型的动态扩展先只支持 HTTP 调用和本地方法调用砍掉分布式调度第一版只支持单机调度器砍掉并行节点全部串行砍掉条件分支一个 Workflow 一棵顺序树。砍完之后系统长这个样子业务方调用 REA SDK 的 api传入事件类型和数据 → SDK 把事件落库到 work_event → 同步或异步触发调度器 → 调度器扫描 PENDING 的 instance → 按 Workflow 定义逐个执行 Action → 更新状态和执行日志。这个最小闭环最核心的价值是它跑通了。跑通以后后面的优化和扩展才有基础。3.2 调度器核心实现扫描、抢占、执行调度器我用了一个简单的轮询模型。一个主循环周期性地扫描 work_instance 表找出所有满足执行条件且还没到 next_retry_time 的记录SELECT * FROM work_instance WHERE status PENDING AND next_retry_time NOW() ORDER BY create_time LIMIT 200这里为什么要 LIMIT 200这是为了避免一次扫太多导致数据库压力过大。单机调度器处理能力有限宁可多扫几次也不要一次拉回大量记录。这个值是我压测之后选的200 在当前 MySQL 配置下查询耗时基本在个位数毫秒。扫描到候选记录之后每条记录都执行一次 CAS 更新UPDATE work_instance SET status RUNNING, version version 1, update_time NOW() WHERE instance_id ? AND status PENDING如果更新成功就放进线程池执行如果失败说明被其他线程抢占了丢弃即可。执行 Action 的伪代码如下public void executeWorkflow(WorkflowInstance instance) { for (int i instance.getCurrentActionIndex(); i workflow.getActions().size(); i) { Action action workflow.getActions().get(i); try { ActionResult result actionExecutor.execute(action, instance.getPayload()); saveActionLog(instance.getInstanceId(), action.getName(), SUCCESS, result, null); instance.setCurrentActionIndex(i 1); updateInstanceIndexAndStatus(instance, RUNNING); } catch (Exception e) { boolean shouldRetry handleRetry(instance, i, e); if (!shouldRetry) { updateInstanceStatus(instance, FAILED); break; } else { updateInstanceWithNextRetry(instance, calculateNextRetryTime(instance.getRetryCount())); return; } } } updateInstanceStatus(instance, SUCCESS); }这段逻辑有几个关键点current_action_index 是断点续传的索引。如果流程在第三个 Action 失败下一次重试直接从第三个 Action 开始不用重跑前两个。这里和第一版串行的选择环环相扣。因为串行current_action_index 的推进是线性的非常容易理解和排查。如果支持了并行索引推进逻辑会复杂不少。3.3 重试与退避失败不能无脑重试Action 执行失败时最怕无脑立刻重试。如果下游服务已经过载立刻重试只会加重问题。所以 REA 的重试策略是指数退避加最大次数限制。退避间隔计算是baseDelayMs 1000 delay baseDelayMs * (2 ^ retryCount)再加上一个随机抖动 jitter避免多个实例同时重试引发“惊群效应”。比如某个 Action 失败retryCount0那么第一次重试就安排在 1 秒后第二次重试在 2 秒后第三次在 4 秒后第四次在 8 秒后。默认最大重试次数是 5 次达到上限后状态变成 DEAD。这里我想强调一下随机抖动。很多人写的重试代码没有抖动只按 2^n 退避结果就是大量同时失败的实例会在同一秒一起重试把下游再次打挂。加了 jitter 之后重试时间就散开了。REA 里头 jitter 范围是正负 20%这个比例是我在线上测试后调的。DEAD 状态的实例还需要一个兜底机制。我在调度器之外加了一个定时告警任务每隔 5 分钟扫一次 DEAD 状态的实例发到告警群同时记录到一张死信表方便人工之后批量回放。3.4 可观测设计每个 Action 都要留下痕迹前面说了 work_action_log 表实际实现的时候每一行记录不能只存一个 status。我的字段是这样的log_id BIGINT PRIMARY KEY AUTO_INCREMENT, instance_id VARCHAR(64) NOT NULL, action_name VARCHAR(128) NOT NULL, status VARCHAR(16) NOT NULL, request_summary TEXT, response_summary TEXT, error_msg TEXT, cost_ms BIGINT, create_time DATETIME NOT NULL关键点是 request_summary 和 response_summary。它们存的是截断后的入参和出参摘要而不是完整 payload。因为完整 payload 很多时候包含敏感信息全存下来既不安全也浪费空间。截断策略是统一 toString 后算一个长度超过 500 字符就切掉。另外入参摘要可以在排查时快速定位问题不需要再回原系统找日志。还有 cost_ms。这个字段太重要了业务方经常问“为什么支付成功之后通知要 5 秒才发出去”有了这个字段我可以直接查出来是哪一步耗时最长不用瞎猜。状态变更历史我额外加了一张 work_instance_log 表专门存 instance 状态流转历史。后来用上它主要是因为某次线上流程反复在 RUNNING 和 PENDING 之间横跳光看最终表根本看不出轨迹有了历史表才知道它已经重试了 6 次。4. 常见问题与排查技巧实录4.1 事件“消失”了事务和事件落库的次序问题REA 上线后遇到的第一个线上问题是业务方反馈“有一批订单创建成功了但是积分一直没加”。查 work_event 表发现事件记录压根不存在。这就不是 REA 调度的问题而是事件根本没有成功落库。什么情况下事件会“落库失败”绝大多数是事务次序问题。业务代码的写法是先调用 REA SDK 把事件写入 work_event再提交自己的业务事务。如果业务事务回滚了work_event 里的记录却没有回滚就变成“事件多出来了”反过来如果 SDK 先发了 MQ 消息业务事务还没提交下游就已经开始处理了也会导致处理了不该处理的业务。正确的姿势是把事件落库放到业务事务内部和业务数据一起提交。SDK 提供一个事务内写接口业务方在自己的事务里调用commit 之后事件才对外可见。REA 的调度器扫描到的也就都是已提交业务的事务事件不会出现半成品订单被触发流程的情况。这个模式业内叫 transactional outbox。REA 把它做成了一个内置逻辑事件落库必须和业务同事务提交后才允许被扫描。对业务方来说只是把调用点挪到了事务内成本很低但换来的收益非常大。4.2 重复执行因为扫描到同一条就无脑执行另一个高发问题是流程重复执行。业务方反馈说一个用户下单后收到了两条通知。查日志发现同一个 instance 被两个线程执行了两次。根本原因就是我在 2.4 里说到的竞态。第一版里我用的是普通 UPDATE没有 version没有 CAS导致两个调度线程同时抢到同一条 PENDING 记录。解决方法不复杂就是把更新改成条件更新并加上 status 条件和 version 条件。线上确认过影响行数为 1 的线程才执行后续动作。这边还有一个额外的经验如果你允许同一个 instance 被重新投递那就要格外小心 Action 的幂等设计。REA 的 Action 本身就是调用下游 HTTP 接口下游接口必须支持幂等。我在 Action 配置里增加了一个幂等键透传字段每次请求会给下游带上 reqIdinstanceId_actionName_retryCount下游可以拿这个做唯一校验。至少比没有好得多。4.3 下游响应慢导致任务积压后来系统流量上来了一些REA 开始出现积压。调度器扫描出 200 条 instance全部丢进线程池但下游一个 HTTP 服务响应特别慢把整个线程池占满了其他流程全部排队。这个问题的根源是共用线程池。我一开始把“扫描”和“执行”放到了同一个线程池里扫描线程如果被卡在执行任务上新的任务就扫不出来。优化方案是把扫描线程和执行线程池拆开扫描用一个只有 2 个线程的专用线程池执行用一个可动态配置线程数的池子。另外还引入了一个超时控制每个 Action 的 HTTP 请求 timeoutMs 必须显式配置超时就快速失败进入重试逻辑不占用后续 Action 的时间。还有一条排查经验积压的时候不要先去看线程池大小先去查 work_action_log 里面耗时最高的 action 是哪几个。多数情况下是下游服务抖动不是 REA 本身有问题。把下游服务治理好落回来就顺了。4.4 常见问题速查表症状定位思路处理方案事件根本没进系统查 work_event 是否有记录检查业务事务内事件落库是否同事务提交事件有记录但流程没执行查 instance 是否 PENDING检查调度器是否启动、next_retry_time 是否在未来流程重复执行查 instance 状态是否反复 RUNNING检查 CAS 更新条件、version 是否生效某一步一直失败查 action_log error_msg根据 error_msg 定位下游检查重试策略死信堆积查 DEAD 状态实例数量补告警、批量回放、人工处理积压严重查 action_log cost_ms拆线程池、加超时、治理下游这张表我一直贴在项目文档里每次有人来问问题我就让他先去查这一行。不是说它万能而是能让排查路径固定下来减少凭感觉乱翻日志的盲目性。4.5 关于 SDK 设计的一个小细节项目最后还要给业务方用所以 SDK 的体验很关键。REA SDK 当时的对外接口只有三个fireEvent(eventType, bizKey, payload)触发一个事件queryInstance(bizKey)查询某个业务唯一键对应的流程实例状态retryInstance(instanceId)手动重跑一条失败或死信流程。把接口收敛成这三个是吸取了某个早期版本的教训。一开始我提供了很多细粒度接口业务方拿过来乱用有人手动把 instance 状态改成 SUCCESS有人直接调内部 Action 执行方法。后来我意识到对使用方来说需要的不是灵活性而是稳定的语义。fireEvent 只管上报query 只管查询retry 只负责重跑权限和职责都清晰。内部方法一律私有化。这个设计理念我后来在别的项目里也一直沿用不要向使用方展示你内部是怎么实现的要让他们用一个简单、语义清晰的入口完成一件事。5. 写在最后我在实际操作中的几点体会这个项目做到这里其实还没有到“工具完备”的状态但已经用得很顺。如果你想做一个类似的事件编排组件我讲讲我的体会。第一目标必须具体。REA 能活着是因为它从第一天就很清楚自己解决什么问题不是要做一个大而全的中间件。你如果也要做先写清楚谁在用、要解决哪一类问题、做到哪一步算“可用”。第二可见性高于一切。我在这个项目里最满意的不是调度器代码而是每一条事件和每一次 Action 变动都留了记录。线上不管出什么问题我都能用一条 SQL 把链路还原出来。排查思路固定化比任何算法优化都更有实际价值。第三宁可做得小也不要急着扩大边界。我有几次想给 REA 增加并行能力和可视化编辑界面最后都压下来了。不是功能没用而是并行会增加状态机复杂度界面会分散核心逻辑的打磨精力。一个组件先把主干做扎实自然会有扩展的机会。如果你正在为系统里那些零散的异步逻辑发愁不妨从最朴素的数据库表加一个串行调度器开始。你会发现很多你以为需要引入新框架才能解决的问题用最直接的落库加乐观锁就能解决大半。稳定简单本身就值钱。 
长河君 · 2026/10/21 13:42:12 · 阅读 1.2k 轻量级事件编排引擎REA：可靠事件自动化的设计与实现
REA 是我最近大半年一直在折腾的一个轻量级事件编排引擎英文全名我习惯写成 Reliable Event Automation。说到底它解决的就是一件事把业务系统里“某个事件发生后还要依次做一大堆后续动作”的逻辑从散落的代码里收口成一张可配置、可重试、可观测的执行图。我先说下背景。我自己的项目当时已经不算小了十几个内部模块互相调用订单一创建下游要处理库存、通知、积分、优惠券、埋点光这一条链路就横跨了五个模块。最开始大家都是在各自的服务里监听消息、各自写定时任务补偿时间一长就会出现同一类事件被两个模块分别消费、补偿逻辑互相打架的情况。每次业务方提一句“能不能在支付成功后再增加一步”都要排查半天到底改哪里合适。这就是 REA 最初出现的动机。这篇文章不打算讲什么高大上的架构我会把设计思路、核心实现、以及我在实际排查中遇到过的坑都拆开聊一遍。如果你也在维护一套业务复杂度不低的后端系统正在被异步逻辑、定时任务和重试补偿折磨这篇文章应该能给你一些可以直接落地的参考。1. 为什么我会去做一个叫 REA 的项目1.1 我遇到的真实痛点要理解 REA 为什么长成现在这个样子得先还原我当时的工作场景。假设用户下单成功核心订单系统会触发一个事件然后需要做这些事扣减库存、生成优惠券、发送站内信、给用户累计积分、写入行为分析埋点。这些动作本质上没有严格的先后依赖但业务上希望它们都能完成任何一个失败都需要重试补偿。没引入 REA 之前团队里的做法是各做各的。库存模块自己监听 MQ收到消息后扣减库存失败就本地存表等定时任务扫描积分服务也监听同一份 MQ消费失败直接丢弃通知服务干脆在下单接口里同步调用……这个局面下最难以忍受的不是代码丑而是没有一个地方能回答“这个订单当前处理到哪一步了”这个问题。业务方问十万个为什么我们只能去翻各个服务的日志。更麻烦的是多模块重复消费。因为大家都在监听同一份消息一旦某个消息被重复发送这在 MQ 语义下很常见库存可能扣两次积分可能加两次。大家各自做幂等但每一家的幂等规则还不一样统一排查难度极大。所以我的核心诉求其实不复杂有一个统一入口接收事件能定义一条流程每个步骤是什么每一步要能单独重试跑完之后能清楚看到结果和耗时。这就是 REA 的雏形。它不是为了做一个通用平台而是为了把我自己的“事后补偿地狱”管起来。1.2 为什么不用现成的工作流引擎决定自研之前我认真调研过一些开源工作流引擎。功能确实强大支持人工审批、复杂条件分支、定时触发、可视化配置文档也不少。但我思考后还是放弃了原因主要有几个一是部署成本。引擎本身要独立部署自带的控制台又要占一套资源我的团队没有专门的运维精力去维护这类基础组件出了问题很难快速定位。二是学习成本。业务方和后端都想在可视化界面上拖拖拽拽但实际用起来复杂状态机的维护比写代码麻烦多了遇到边界情况你还得看引擎源码。对一个十人左右、要快速交付业务的小组来说这是很大的隐性成本。三是集成成本。现成引擎往往自带一套数据表和执行模型想接入现有业务代码、让它们共用一套事务边界要做非常多的适配。当时我们已经有大量的业务表和补偿任务强行迁移可能造成线上回归。REA 的定位因此非常明确嵌入式的、轻量的、SDK 化的编排能力而不是一套独立平台。我可以保留数据库里已有的数据表只新增几张运行表可以让业务方直接用代码写每个 Action 点。先解决一个问题事件进来之后后续动作由谁统一接管。1.3 REA 的三个目标可靠、事件驱动、自动化名字是我后来补的。Reliable Event Automation可以拆成三层理解Reliable可靠不丢事件、不重复执行、失败可追溯、支持指数退避重试Event事件驱动不依赖定时轮询去猜“有没有业务变化”而是由业务主动上报“发生了什么”Automation自动化一条事件进来后编排器自动完成剩余动作不需要人为干预。这三个目标直接对应了最终能力表。我用一个表格把它们对齐设计目标对应能力核心实现手段可靠事件不丢失、流程不重复执行事件落库、状态机 CAS可靠失败可重试、有死信兜底指数退避重试、最终失败告警事件驱动统一事件入口、按事件类型路由triggerEvent 与 Workflow 绑定自动化串行/并发执行 Action调度器按流程定义逐级执行可观测每一步执行明细可查Action 日志表记录输入输出摘要这张表在后面设计数据结构的时候起了很大作用。我每次犹豫要不要加新功能都会先对照这三个目标问自己它是为了让流程更可靠还是为了让更多人能编排如果不是就暂时不做。2. 核心细节解析与关键决策2.1 事件模型Event、Workflow、Action 三层抽象REA 的模型不算复杂我把它们分成三个层次Event事件业务侧上报的最小单位比如 order.created、payment.success、refund.closed。它只表达“发生了什么”不关心后续步骤。Workflow流程一个 Workflow 绑定一个触发事件类型内部包含一组有序的 Action 列表。Action动作流程里的最小执行单元可以是对内 HTTP 请求、本地方法调用、异步消息发送等。比较关键的一个设计是事件不能直接决定流程它只是触发的钥匙。同一种事件类型不同业务的流程可能不同所以还需要一个路由规则把 eventName 加业务标识比如支付渠道、用户等级映射到具体的 Workflow。这一点在后面的版本我才补上第一版是硬编码 eventName 到 Workflow 的一对一关系。一个流程定义长这样JSON 格式{ workflowName: orderNotifyFlow, triggerEvent: order.created, timeoutSeconds: 300, actions: [ { name: deductStock, type: http, config: { method: POST, url: http://internal-stock/inventory/deduct, timeoutMs: 3000 } }, { name: sendMessage, type: local, config: { beanClass: com.demo.NotificationSender, method: send } }, { name: grantPoints, type: http, config: { method: POST, url: http://internal-points/points/grant, timeoutMs: 5000 } } ] }这里重点不是 JSON 格式本身而是流程的生命周期。一个流程实例Instance会经历 PENDING → RUNNING → SUCCESS / FAILED / DEAD 这几个状态。开局必须 PENDING执行完毕后变成 SUCCESS如果某个 Action 重试最终失败整条流程进入 FAILED重试次数超过上限进入 DEAD也就是死信状态用来人工介入。2.2 同步串行还是并发执行第一版必须先简单写第一版的时候我面临一个选择Action 到底是串行好还是允许并发答案很明确第一版必须串行。原因在于并发会立刻引入三个问题一是乱序多个 Action 同时执行它们的完成顺序不确定有的步骤之间有隐含依赖二是一致性同时写流程状态和执行日志需要分布式锁复杂度陡增三是排查困难一旦失败你很难说清楚是哪个分支引起的。所以我把整个 Workflow 设计成一条顺序链一个 Action 完成后再执行下一个。不同 Workflow 之间并行Workflow 内部串行。这是我的稳妥策略。等到后面把并发需求整理清楚再针对特定节点加 parallel 配置而不是一开始就支持。实际上大多数异步流程的需求在串行模型下都能解决。所谓的“并发”更多是想节省总体耗时。但业务系统这几十毫秒的差距远没有失败可追踪重要。2.3 存储设计为什么用三张表而不是直接塞 Redis最初我也考虑过用 Redis 的 list 或者 stream 做事件队列理由当然是快。但一深入就发现了问题事件丢失风险。Redis 没有成熟的持久化语义如果宕机正好发生在写入之后、消费之前事件就没了可回溯性差。Redis 里面的数据不方便按时间、流程实例维度去查历史审计不满足。业务方经常需要看到“这个事件从哪来的执行了哪几步每步耗时多少”Redis 的简单数据类型很难支撑这种查询。所以我的最终方案是用 MySQL 来兜底所有流程数据Redis 只作为可选的高效缓存。表结构分三张work_event存初始事件字段包括 event_id、event_type、biz_key、payload JSON、create_time。work_instance存流程执行实例字段包括 instance_id、event_id、workflow_name、status、current_action_index、retry_count、next_retry_time、version、create_time、update_time。work_action_log存每个 Action 的执行明细字段包括 log_id、instance_id、action_name、status、request_summary、response_summary、error_msg、cost_ms、create_time。用三张表是为了把“入口信息”“流程状态”“执行明细”分开。工作里面排查问题时三个维度都方便单独查。比如业务方问“上周三有几个支付成功事件进入流程后失败了”我只需要在 work_event 表按事件类型和时间刷一下再把 instance_id join 到 work_instance 看状态最后查 action_log 就能定位是哪一步失败。这个方案牺牲了一点极致的性能换来了极致的可见性。对于一个内部编排组件来说数据可见性远比吞吐量重要。2.4 状态更新必须用 CAS不能用普通 UPDATE执行流程的时候调度器会扫描到一条 PENDING 状态的 instance然后准备执行。这里有一个很容易踩的大坑如果你只是用 statusRUNNING 去更新这条记录可能会发生多个调度线程同时处理同一个 instance。因为扫描和更新之间是有时间差的。扫描线程 A 查到一条 PENDING 记录正要执行另一个线程 B 也扫描到了同一记录也准备执行。两个线程同时执行同一个流程事件就会被重复处理。解决方法是 CASCompare And Swap更新UPDATE work_instance SET status RUNNING, version version 1, update_time NOW() WHERE instance_id ? AND status PENDING AND version ?这行 SQL 是 REA 里面最核心的一句。它保证只有一个线程能把状态从 PENDING 改成 RUNNING。如果影响行数为 0说明已经有别的线程抢先执行当前线程直接放弃。很多人会忽略 version 字段其实它是防止 ABA 问题的关键。有了 version你能确定工作是在预期的历史版本之上做的修改不至于拿旧状态覆盖新状态。3. 实操过程与核心环节实现3.1 最小可用版本要圈定边界REA 第一版我规定自己两个星期内上线。为了做到这一点我硬性砍掉了不少功能砍掉的东西以后再补砍掉可视化界面所有流程定义先写在本地配置文件里砍掉 Action 类型的动态扩展先只支持 HTTP 调用和本地方法调用砍掉分布式调度第一版只支持单机调度器砍掉并行节点全部串行砍掉条件分支一个 Workflow 一棵顺序树。砍完之后系统长这个样子业务方调用 REA SDK 的 api传入事件类型和数据 → SDK 把事件落库到 work_event → 同步或异步触发调度器 → 调度器扫描 PENDING 的 instance → 按 Workflow 定义逐个执行 Action → 更新状态和执行日志。这个最小闭环最核心的价值是它跑通了。跑通以后后面的优化和扩展才有基础。3.2 调度器核心实现扫描、抢占、执行调度器我用了一个简单的轮询模型。一个主循环周期性地扫描 work_instance 表找出所有满足执行条件且还没到 next_retry_time 的记录SELECT * FROM work_instance WHERE status PENDING AND next_retry_time NOW() ORDER BY create_time LIMIT 200这里为什么要 LIMIT 200这是为了避免一次扫太多导致数据库压力过大。单机调度器处理能力有限宁可多扫几次也不要一次拉回大量记录。这个值是我压测之后选的200 在当前 MySQL 配置下查询耗时基本在个位数毫秒。扫描到候选记录之后每条记录都执行一次 CAS 更新UPDATE work_instance SET status RUNNING, version version 1, update_time NOW() WHERE instance_id ? AND status PENDING如果更新成功就放进线程池执行如果失败说明被其他线程抢占了丢弃即可。执行 Action 的伪代码如下public void executeWorkflow(WorkflowInstance instance) { for (int i instance.getCurrentActionIndex(); i workflow.getActions().size(); i) { Action action workflow.getActions().get(i); try { ActionResult result actionExecutor.execute(action, instance.getPayload()); saveActionLog(instance.getInstanceId(), action.getName(), SUCCESS, result, null); instance.setCurrentActionIndex(i 1); updateInstanceIndexAndStatus(instance, RUNNING); } catch (Exception e) { boolean shouldRetry handleRetry(instance, i, e); if (!shouldRetry) { updateInstanceStatus(instance, FAILED); break; } else { updateInstanceWithNextRetry(instance, calculateNextRetryTime(instance.getRetryCount())); return; } } } updateInstanceStatus(instance, SUCCESS); }这段逻辑有几个关键点current_action_index 是断点续传的索引。如果流程在第三个 Action 失败下一次重试直接从第三个 Action 开始不用重跑前两个。这里和第一版串行的选择环环相扣。因为串行current_action_index 的推进是线性的非常容易理解和排查。如果支持了并行索引推进逻辑会复杂不少。3.3 重试与退避失败不能无脑重试Action 执行失败时最怕无脑立刻重试。如果下游服务已经过载立刻重试只会加重问题。所以 REA 的重试策略是指数退避加最大次数限制。退避间隔计算是baseDelayMs 1000 delay baseDelayMs * (2 ^ retryCount)再加上一个随机抖动 jitter避免多个实例同时重试引发“惊群效应”。比如某个 Action 失败retryCount0那么第一次重试就安排在 1 秒后第二次重试在 2 秒后第三次在 4 秒后第四次在 8 秒后。默认最大重试次数是 5 次达到上限后状态变成 DEAD。这里我想强调一下随机抖动。很多人写的重试代码没有抖动只按 2^n 退避结果就是大量同时失败的实例会在同一秒一起重试把下游再次打挂。加了 jitter 之后重试时间就散开了。REA 里头 jitter 范围是正负 20%这个比例是我在线上测试后调的。DEAD 状态的实例还需要一个兜底机制。我在调度器之外加了一个定时告警任务每隔 5 分钟扫一次 DEAD 状态的实例发到告警群同时记录到一张死信表方便人工之后批量回放。3.4 可观测设计每个 Action 都要留下痕迹前面说了 work_action_log 表实际实现的时候每一行记录不能只存一个 status。我的字段是这样的log_id BIGINT PRIMARY KEY AUTO_INCREMENT, instance_id VARCHAR(64) NOT NULL, action_name VARCHAR(128) NOT NULL, status VARCHAR(16) NOT NULL, request_summary TEXT, response_summary TEXT, error_msg TEXT, cost_ms BIGINT, create_time DATETIME NOT NULL关键点是 request_summary 和 response_summary。它们存的是截断后的入参和出参摘要而不是完整 payload。因为完整 payload 很多时候包含敏感信息全存下来既不安全也浪费空间。截断策略是统一 toString 后算一个长度超过 500 字符就切掉。另外入参摘要可以在排查时快速定位问题不需要再回原系统找日志。还有 cost_ms。这个字段太重要了业务方经常问“为什么支付成功之后通知要 5 秒才发出去”有了这个字段我可以直接查出来是哪一步耗时最长不用瞎猜。状态变更历史我额外加了一张 work_instance_log 表专门存 instance 状态流转历史。后来用上它主要是因为某次线上流程反复在 RUNNING 和 PENDING 之间横跳光看最终表根本看不出轨迹有了历史表才知道它已经重试了 6 次。4. 常见问题与排查技巧实录4.1 事件“消失”了事务和事件落库的次序问题REA 上线后遇到的第一个线上问题是业务方反馈“有一批订单创建成功了但是积分一直没加”。查 work_event 表发现事件记录压根不存在。这就不是 REA 调度的问题而是事件根本没有成功落库。什么情况下事件会“落库失败”绝大多数是事务次序问题。业务代码的写法是先调用 REA SDK 把事件写入 work_event再提交自己的业务事务。如果业务事务回滚了work_event 里的记录却没有回滚就变成“事件多出来了”反过来如果 SDK 先发了 MQ 消息业务事务还没提交下游就已经开始处理了也会导致处理了不该处理的业务。正确的姿势是把事件落库放到业务事务内部和业务数据一起提交。SDK 提供一个事务内写接口业务方在自己的事务里调用commit 之后事件才对外可见。REA 的调度器扫描到的也就都是已提交业务的事务事件不会出现半成品订单被触发流程的情况。这个模式业内叫 transactional outbox。REA 把它做成了一个内置逻辑事件落库必须和业务同事务提交后才允许被扫描。对业务方来说只是把调用点挪到了事务内成本很低但换来的收益非常大。4.2 重复执行因为扫描到同一条就无脑执行另一个高发问题是流程重复执行。业务方反馈说一个用户下单后收到了两条通知。查日志发现同一个 instance 被两个线程执行了两次。根本原因就是我在 2.4 里说到的竞态。第一版里我用的是普通 UPDATE没有 version没有 CAS导致两个调度线程同时抢到同一条 PENDING 记录。解决方法不复杂就是把更新改成条件更新并加上 status 条件和 version 条件。线上确认过影响行数为 1 的线程才执行后续动作。这边还有一个额外的经验如果你允许同一个 instance 被重新投递那就要格外小心 Action 的幂等设计。REA 的 Action 本身就是调用下游 HTTP 接口下游接口必须支持幂等。我在 Action 配置里增加了一个幂等键透传字段每次请求会给下游带上 reqIdinstanceId_actionName_retryCount下游可以拿这个做唯一校验。至少比没有好得多。4.3 下游响应慢导致任务积压后来系统流量上来了一些REA 开始出现积压。调度器扫描出 200 条 instance全部丢进线程池但下游一个 HTTP 服务响应特别慢把整个线程池占满了其他流程全部排队。这个问题的根源是共用线程池。我一开始把“扫描”和“执行”放到了同一个线程池里扫描线程如果被卡在执行任务上新的任务就扫不出来。优化方案是把扫描线程和执行线程池拆开扫描用一个只有 2 个线程的专用线程池执行用一个可动态配置线程数的池子。另外还引入了一个超时控制每个 Action 的 HTTP 请求 timeoutMs 必须显式配置超时就快速失败进入重试逻辑不占用后续 Action 的时间。还有一条排查经验积压的时候不要先去看线程池大小先去查 work_action_log 里面耗时最高的 action 是哪几个。多数情况下是下游服务抖动不是 REA 本身有问题。把下游服务治理好落回来就顺了。4.4 常见问题速查表症状定位思路处理方案事件根本没进系统查 work_event 是否有记录检查业务事务内事件落库是否同事务提交事件有记录但流程没执行查 instance 是否 PENDING检查调度器是否启动、next_retry_time 是否在未来流程重复执行查 instance 状态是否反复 RUNNING检查 CAS 更新条件、version 是否生效某一步一直失败查 action_log error_msg根据 error_msg 定位下游检查重试策略死信堆积查 DEAD 状态实例数量补告警、批量回放、人工处理积压严重查 action_log cost_ms拆线程池、加超时、治理下游这张表我一直贴在项目文档里每次有人来问问题我就让他先去查这一行。不是说它万能而是能让排查路径固定下来减少凭感觉乱翻日志的盲目性。4.5 关于 SDK 设计的一个小细节项目最后还要给业务方用所以 SDK 的体验很关键。REA SDK 当时的对外接口只有三个fireEvent(eventType, bizKey, payload)触发一个事件queryInstance(bizKey)查询某个业务唯一键对应的流程实例状态retryInstance(instanceId)手动重跑一条失败或死信流程。把接口收敛成这三个是吸取了某个早期版本的教训。一开始我提供了很多细粒度接口业务方拿过来乱用有人手动把 instance 状态改成 SUCCESS有人直接调内部 Action 执行方法。后来我意识到对使用方来说需要的不是灵活性而是稳定的语义。fireEvent 只管上报query 只管查询retry 只负责重跑权限和职责都清晰。内部方法一律私有化。这个设计理念我后来在别的项目里也一直沿用不要向使用方展示你内部是怎么实现的要让他们用一个简单、语义清晰的入口完成一件事。5. 写在最后我在实际操作中的几点体会这个项目做到这里其实还没有到“工具完备”的状态但已经用得很顺。如果你想做一个类似的事件编排组件我讲讲我的体会。第一目标必须具体。REA 能活着是因为它从第一天就很清楚自己解决什么问题不是要做一个大而全的中间件。你如果也要做先写清楚谁在用、要解决哪一类问题、做到哪一步算“可用”。第二可见性高于一切。我在这个项目里最满意的不是调度器代码而是每一条事件和每一次 Action 变动都留了记录。线上不管出什么问题我都能用一条 SQL 把链路还原出来。排查思路固定化比任何算法优化都更有实际价值。第三宁可做得小也不要急着扩大边界。我有几次想给 REA 增加并行能力和可视化编辑界面最后都压下来了。不是功能没用而是并行会增加状态机复杂度界面会分散核心逻辑的打磨精力。一个组件先把主干做扎实自然会有扩展的机会。如果你正在为系统里那些零散的异步逻辑发愁不妨从最朴素的数据库表加一个串行调度器开始。你会发现很多你以为需要引入新框架才能解决的问题用最直接的落库加乐观锁就能解决大半。稳定简单本身就值钱。