分布式 SAGA 模式全解与 Java 入门示例

分布式 SAGA 模式全解与 Java 入门示例

术语更正：本文讨论的是分布式事务的SAGA 模式（非“sage”）。SAGA 通过将一个跨服务的长事务拆分为多个本地事务，并在失败时按逆序执行补偿事务，实现最终一致性。它特别适合长事务、复杂流程、可接受短暂中间状态的业务场景，如电商下单全流程、物流履约、金融审批等。

一、核心概念与适用场景

• 核心思想
  • 将一个全局事务拆分为有序的本地事务链：LT1 → LT2 → … → LTn。
  • 每个 LT 成功后立即提交（释放资源、无全局锁），并生成对应的补偿事务 CTi用于撤销影响。
  • 任意 LT 失败时，按逆序执行已成功步骤的补偿：CTn → … → CT1，使数据回到一致状态。
• 关键角色
  • 事务发起者 Initiator：触发 SAGA。
  • 参与者 Participant：执行本地事务与补偿事务的服务。
  • 协调器 Coordinator：维护全局状态、推进流程、失败回滚（编排式/协同式）。
• 适用场景
  • 长事务/长时间等待（如用户支付、物流运输）。
  • 多服务串行/并行的复杂流程。
  • 低侵入改造需求（相比 TCC 少接口改造，只需新增补偿）。
  • 可接受最终一致性而非强一致。
• 与其他方案对比（简表）

• 核心挑战
  • 补偿逻辑精准性（有些操作不可逆，需要替代补偿如退款/召回）。
  • 幂等性（网络重试导致重复执行）。
  • 并发冲突（同一资源多 SAGA 并发修改）。
  • 中间状态可见性/隔离性（需通过状态标记、版本号、业务规则缓解）。

二、两种实现模式图解与对比

• 编排式（Choreography，去中心化）
  • 每个参与者通过事件/消息驱动下一步；失败则广播补偿。
  • 优点：无单点、耦合低；缺点：流程分散、全局状态难追踪、易循环依赖。
• 协同式（Orchestration，中心化）
  • 协调器统一定义流程与回滚顺序，依次调用参与者；失败按逆序补偿。
  • 优点：流程集中、易维护与观测；缺点：协调器单点风险（需高可用）。

示意时序（简化）：

编排式： LT1→发“T1成功”→LT2→发“T2成功”→LT3 若 LT2 失败→发“T2失败”→LT1 执行 CT1 协同式： 协调器→LT1→LT2→LT3 若 LT2 失败→协调器→CT2→CT1

• 选型建议
  • ≤3 步的简单流程：编排式实现更快。
  • 多步骤/多分支/需可视化编排：协同式更稳。

三、Java 极简示例 协调式 SAGA（无框架）

目标：模拟“扣款 → 扣库存”，失败则“恢复库存 → 冲正扣款”。强调幂等与防悬挂。

• 1. 领域与幂等键

publicclassSagaContext{publicfinalStringsagaId=java.util.UUID.randomUUID().toString();publicfinalStringbusinessKey="order-1001";// 可扩展：超时时间、重试次数、状态等}

• 2. 事务步骤接口

publicinterfaceSagaStep{// 正向本地事务：true=成功，false=失败booleanexecute(SagaContextctx);// 补偿事务：true=补偿成功，false=需重试/告警booleancompensate(SagaContextctx);}

• 3. 两个参与者示例

importjava.math.BigDecimal;importjava.util.concurrent.ConcurrentHashMap;importjava.util.concurrent.atomic.AtomicInteger;publicclassAccountStepimplementsSagaStep{// 模拟账户余额（生产请用 DB）privatestaticfinalConcurrentHashMap<String,BigDecimal>BALANCE=newConcurrentHashMap<>();// 幂等与防悬挂：sagaId -> 已执行动作（避免重复执行/补偿后正向再执行）privatestaticfinalConcurrentHashMap<String,String>EXEC_LOG=newConcurrentHashMap<>();static{BALANCE.put("A001",newBigDecimal("1000"));}@Overridepublicbooleanexecute(SagaContextctx){Stringdone=EXEC_LOG.putIfAbsent(ctx.sagaId+":minus","1");if(done!=null)returntrue;// 幂等：已执行过BigDecimalcur=BALANCE.get("A001");if(cur.compareTo(newBigDecimal("100"))<0)returnfalse;BALANCE.put("A001",cur.subtract(newBigDecimal("100")));System.out.printf("[Account] 扣款成功，余额=%s，sagaId=%s%n",BALANCE.get("A001"),ctx.sagaId);returntrue;}@Overridepublicbooleancompensate(SagaContextctx){// 防悬挂：若正向未执行过，也要记录补偿痕迹，避免正向后补执行Stringpend=EXEC_LOG.putIfAbsent(ctx.sagaId+":compMinus","1");if("1".equals(pend)){System.out.printf("[Account] 补偿已记录或执行过，sagaId=%s%n",ctx.sagaId);returntrue;}BigDecimalcur=BALANCE.get("A001");BALANCE.put("A001",cur.add(newBigDecimal("100")));System.out.printf("[Account] 冲正成功，余额=%s，sagaId=%s%n",BALANCE.get("A001"),ctx.sagaId);returntrue;}}publicclassInventoryStepimplementsSagaStep{// 模拟库存（生产请用 DB）privatestaticfinalConcurrentHashMap<String,AtomicInteger>STOCK=newConcurrentHashMap<>();privatestaticfinalConcurrentHashMap<String,String>EXEC_LOG=newConcurrentHashMap<>();static{STOCK.put("P100",newAtomicInteger(10));}@Overridepublicbooleanexecute(SagaContextctx){Stringdone=EXEC_LOG.putIfAbsent(ctx.sagaId+":deduct","1");if(done!=null)returntrue;AtomicIntegers=STOCK.get("P100");if(s.decrementAndGet()<0){// 回滚本地变更（演示用，生产需事务内操作）s.incrementAndGet();returnfalse;}System.out.printf("[Inventory] 扣减库存成功，库存=%d，sagaId=%s%n",s.get(),ctx.sagaId);returntrue;}@Overridepublicbooleancompensate(SagaContextctx){Stringpend=EXEC_LOG.putIfAbsent(ctx.sagaId+":compDeduct","1");if("1".equals(pend)){System.out.printf("[Inventory] 补偿已记录或执行过，sagaId=%s%n",ctx.sagaId);returntrue;}STOCK.get("P100").incrementAndGet();System.out.printf("[Inventory] 恢复库存成功，库存=%d，sagaId=%s%n",STOCK.get("P100").get(),ctx.sagaId);returntrue;}}

• 4. 协调器与回滚

importjava.util.ArrayList;importjava.util.List;publicclassSagaCoordinator{privatefinalList<SagaStep>steps=newArrayList<>();publicSagaCoordinatoraddStep(SagaStepstep){steps.add(step);returnthis;}publicvoidexecute(SagaContextctx){List<Integer>done=newArrayList<>();try{for(inti=0;i<steps.size();i++){if(!steps.get(i).execute(ctx)){thrownewRuntimeException("步骤["+i+"]执行失败，触发回滚");}done.add(i);}System.out.printf("[Saga] 执行成功，sagaId=%s%n",ctx.sagaId);}catch(Exceptionex){System.out.printf("[Saga] 执行失败，开始补偿，sagaId=%s，原因=%s%n",ctx.sagaId,ex.getMessage());// 逆序补偿for(inti=done.size()-1;i>=0;i--){booleancompOk=steps.get(i).compensate(ctx);if(!compOk){System.err.printf("[Saga] 补偿步骤[%d]失败，需人工介入，sagaId=%s%n",i,ctx.sagaId);}}}}publicstaticvoidmain(String[]args){SagaContextctx=newSagaContext();newSagaCoordinator().addStep(newAccountStep()).addStep(newInventoryStep()).execute(ctx);}}

• 5. 运行与验证
  • 正常：库存充足时，输出余额900、库存9。
  • 异常：将库存初始改为0，会触发“扣库存失败 → 恢复库存 → 冲正扣款”，余额回到1000、库存10。
• 关键点
  • 幂等：通过ConcurrentHashMap.putIfAbsent记录已执行动作。
  • 防悬挂：补偿先写日志，避免补偿后再执行正向。
  • 无全局锁：每个步骤本地事务提交，提升吞吐。

🔥 关注公众号【云技纵横】，目前正在更新分布式缓存进阶技巧和干货

四、生产落地要点与框架选型

• 幂等与去重
  • 为每个 SAGA 分配全局唯一事务ID（sagaId），在参与者的本地表中记录“动作类型+状态+业务键”，用唯一索引/版本号保证幂等。
• 可靠消息与“发件箱”模式
  • 协调器/参与者更新本地事务后，将事件写入本地发件箱表，再由转发器可靠投递到 MQ，确保“状态变更与事件发送”原子性。
• 超时、重试与死信队列
  • 对可重试异常使用指数退避与最大重试次数；多次失败入DLQ并告警人工介入。
• 并发与隔离
  • 通过语义锁/版本号/交换式更新/重读值等策略降低脏写风险；必要时采用业务排队或分区锁。
• 协调器高可用
  • 协同式需做主从/集群、持久化状态、故障转移与可观测性（指标/日志/追踪）。
• 框架选型建议
  • Seata SAGA：基于状态机引擎编排，支持条件选择、并发、子流程、参数映射、重试/捕获、补偿触发等，适合复杂流程与可视化编排。
  • 阿里云 SOFABoot Saga：提供参与者开发范式与防悬挂等工程化实践，适合金融级场景。

五、常见问题快速排查清单

• 补偿重复执行导致“多退/多冲正”
  • 检查补偿接口幂等键（sagaId+action），使用状态机或去重表拦截重复补偿。
• 补偿失败或一直重试
  • 入DLQ、触发告警、提供管理端重试/跳过，必要时人工介入。
• 正向在补偿后“后发先至”（悬挂）
  • 在补偿成功时写入已补偿标记，正向执行前先校验，若已补偿则直接失败。
• 并发扣减同一资源出现“负库存/错账”
  • 使用版本号/条件更新或分区串行化；结合语义锁降低冲突窗口。
• 流程变更难维护
  • 采用状态机编排集中管理流程，变更只需改状态图/DSL，降低耦合与回归成本。