Saga 解决什么问题
分布式系统中，一个业务操作涉及多个服务。传统 ACID 事务无法跨越服务边界——Saga 用本地事务 + 补偿实现最终一致性。
示例：下单流程
1. 创建订单（order-service） ✅ 成功
2. 扣减库存（inventory-service） ✅ 成功
3. 扣款（payment-service） ❌ 失败
4. 补偿：恢复库存（按补偿单号幂等） ✅ 补偿成功
5. 补偿：取消订单（状态=CANCELLED） ✅ 补偿成功
最终状态：订单取消，库存恢复，没有扣款。系统回到操作前的一致状态。
编排式 Saga（Orchestration）
由中央协调器（Saga Orchestrator）控制流程：
import java.util.function.Supplier;
@Service
public class OrderSagaOrchestrator {
private final OrderService orderService;
private final InventoryService inventoryService;
private final PaymentService paymentService;
public Mono<Void> executeOrderSaga(OrderRequest request) {
// 补偿动作存 Supplier，失败时才订阅；直接存 Mono 会变成「先执行、后保存」
List<Supplier<Mono<Void>>> compensations = new ArrayList<>();
return orderService.create(request)
.doOnNext(order -> compensations.add(
() -> orderService.cancel(order.id())))
.flatMap(order ->
inventoryService.deduct(order.symbol(), order.quantity())
.doOnNext(inv -> compensations.add(
() -> inventoryService.restore(order.symbol(), order.quantity())))
.flatMap(inv ->
paymentService.charge(order.userId(), order.totalAmount())
.doOnNext(payment -> compensations.add(
() -> paymentService.refund(payment.id())))
)
)
.then()
.onErrorResume(err -> executeCompensations(compensations)
.then(Mono.error(err))); // 补偿成功时仍向上游报告原始失败；补偿失败则抛出 CompensationFailedException
}
private Mono<Void> executeCompensations(List<Supplier<Mono<Void>>> compensations) {
List<Throwable> failures = new ArrayList<>();
// 逆序执行补偿；串行，不并行
return Flux.fromIterable(compensations.reversed())
.concatMap(supplier -> supplier.get()
.doOnError(failures::add)
.onErrorResume(e -> Mono.empty())) // 单步失败不阻断其余补偿
.then(Mono.defer(() -> failures.isEmpty()
? Mono.empty()
: Mono.error(new CompensationFailedException(failures))));
}
}
关键点：
- 正向操作按顺序执行
- 每个成功步骤记录补偿操作
- 失败时逆序执行补偿（先补偿后执行的步骤）
- 补偿也有失败可能，需要日志记录人工介入
注意上面的 compensations 只存在内存里：进程在扣款成功后崩溃，补偿列表会一起消失，重启后没有任何线索。生产实现要在每个正向步骤完成时把「已执行步骤 + 对应补偿动作」写进 saga_logs（与业务写在同一本地事务里），恢复时从日志重建并继续补偿；内存列表只能作为单进程内的快速路径。
协同式 Saga（Choreography）
没有中央协调器，各服务通过事件驱动完成 Saga：
OrderService:
创建订单 → 发布 OrderCreated Event
InventoryService（订阅 OrderCreated）:
扣减库存 → 发布 InventoryDeducted Event
或库存不足 → 发布 InventoryInsufficient Event
PaymentService（订阅 InventoryDeducted）:
扣款 → 发布 PaymentProcessed Event
或扣款失败 → 发布 PaymentFailed Event
OrderService（订阅 PaymentProcessed）:
更新订单状态 = CONFIRMED
OrderService（订阅 PaymentFailed）：
取消订单 → 发布 OrderCancelled Event
InventoryService（订阅 OrderCancelled）：
恢复库存
协同式更松散，但需要事件总线支持（第 12.1 节）。
补偿设计原则
| 原则 | 说明 | 示例 |
|---|---|---|
| 幂等性 | 补偿可安全重复执行 | 补偿流水以 (saga_id, step) 唯一键去重；quantity + 1 执行两次会加两次，不幂等 |
| 可观测 | 补偿操作记录日志 | Saga 执行日志表 |
| 异步友好 | 补偿也是响应式 | inventoryService.restore() 返回 Mono |
| 有限重试 | 补偿失败重试 N 次 | 3 次后告警人工处理 |
Saga 状态机
复杂 Saga 需要显式状态机：
public enum SagaState {
STARTED,
ORDER_CREATED,
INVENTORY_DEDUCTED,
PAYMENT_PROCESSED,
COMPLETED,
COMPENSATING,
COMPENSATED,
FAILED;
public boolean isTerminal() {
return this == COMPLETED || this == COMPENSATED || this == FAILED;
}
}
@Component
public class SagaStateMachine {
private final SagaLogRepository logRepo;
public Mono<Void> transition(String sagaId, SagaState from, SagaState to) {
return logRepo.save(new SagaLog(sagaId, from, to, Instant.now(), null))
.then();
}
public Mono<SagaState> getCurrentState(String sagaId) {
return logRepo.findLatestBySagaId(sagaId)
.map(SagaLog::toState);
}
}
Saga 日志持久化到数据库，支持故障恢复后状态重建。注意上面的 transition 只是追加日志，没有校验当前状态：并发或重复投递可能写出两条互相矛盾的转移。生产实现应把期望的起始状态作为条件写进更新，例如 UPDATE saga_state SET to_state = :to WHERE saga_id = :id AND to_state = :from，影响行数为 0 就说明状态已被别的执行者改过；或者在 append-only 日志表上对 (saga_id, seq) 加唯一约束、用版本号做乐观锁。冲突时应拒绝并让调用方重读状态，而不是覆盖。
响应式 Saga 的陷阱
陷阱 1：补偿顺序
// 错误：并行执行补偿
cancelOrder().subscribe(); // 异步，可能晚于 restoreInventory
restoreInventory().subscribe();
// 正确：顺序执行
restoreInventory()
.then(cancelOrder())
.subscribe();
陷阱 2：忽略补偿失败
// 错误：补偿失败被吞掉
.onErrorResume(err -> compensation().onErrorResume(e -> Mono.empty()));
// 正确：补偿失败记录告警，并且仍然把原始失败上抛
.onErrorResume(err -> compensation()
.doOnError(compErr -> alert("Compensation failed: " + compErr))
.onErrorResume(compErr -> Mono.empty())
.then(Mono.error(err)));
补偿失败被 onErrorResume 变成完成信号后，调用方会以为 Saga 成功；必须显式把原始异常重新抛出，或者返回一个明确的 SagaFailedException。
陷阱 3：状态不一致窗口
正向操作和补偿之间有时间窗口。在此期间查询状态，可能看到不一致。
解决：Saga 状态查询接口
@GetMapping("/sagas/{sagaId}")
public Mono<SagaStatus> getSagaStatus(@PathVariable String sagaId) {
return sagaStateMachine.getCurrentState(sagaId)
.switchIfEmpty(Mono.error(new SagaNotFoundException(sagaId)))
.map(state -> new SagaStatus(sagaId, state, state.isTerminal()));
}
eventd v2 的 Saga 实现
eventd v2 使用编排式 Saga（代码集中，易于调试）：
OrderSagaOrchestrator（eventd-api）
→ OrderService（R2DBC 事务）
→ InventoryService（HTTP 调用）
→ PaymentService（HTTP 调用）
失败 → 补偿链（逆序）
→ refundPayment()
→ restoreInventory()
→ cancelOrder()
Saga 日志写入 saga_logs 表，支持运营后台查询：
CREATE TABLE saga_logs (
id BIGSERIAL PRIMARY KEY,
saga_id VARCHAR(64) NOT NULL,
from_state VARCHAR(32),
to_state VARCHAR(32) NOT NULL,
occurred_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
error_msg TEXT
);
-- 按 Saga 倒序取最新状态
CREATE INDEX idx_saga_logs_saga_id ON saga_logs (saga_id, occurred_at DESC);
对应的日志记录：
public record SagaLog(
String sagaId, SagaState from, SagaState to,
Instant occurredAt, String errorMsg
) {
public SagaState toState() { return to; }
}
小结
| 模式 | 控制 | 复杂度 | 调试 | 适用 |
|---|---|---|---|---|
| 编排式 | 中央协调器 | 中 | 易（单点） | 流程固定、步骤少 |
| 协同式 | 事件驱动 | 高 | 难（分散） | 流程多变、服务多 |
| 要素 | 实现 |
|---|---|
| 补偿 | 逆序、幂等、响应式 |
| 日志 | Saga 状态机持久化 |
| 恢复 | 启动时重放未完成的 Saga（每步必须幂等，否则重放会重复扣款/加库存） |
| 告警 | 补偿失败通知人工 |
下一节 12.3 eventd v2 全链路集成 ：Netty → WebFlux → R2DBC → Cache 端到端可运行架构。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。