Saga 模式：微服务中的分布式事务
1. 概述
在单体应用中，你在方法上放置 @Transactional，数据库则处理其余部分：要么所有内容都被提交，要么所有内容都被回滚。当拆分为微服务时，每个服务都有自己的数据库，所以单个 ACID 事务无法再覆盖整个业务流程。
本文介绍 Saga 模式，这是解决该问题最常见的方法。读完本文后，你将理解两种 Saga 类型（编排和编舞），能用 Java 21 编写一个小型 Saga 编排器，测试补偿流程，并了解在生产环境中使用 Saga 时的常见错误。
2. 问题：一个业务流程，多个数据库
考虑包含三个服务的订单流程：
- InventoryService 管理仓库中的货物。
- PaymentService 扣除客户的款项。
- ShippingService 创建发货单。
假设第 1 步成功但第 2 步由于客户余额不足而失败。此时货物已在仓库中被保留，但没有付款。你无法从 PaymentService 对 InventoryService 的数据库进行“回滚”。
经典解决方案是 两阶段提交（2PC）。然而，在微服务中很少使用 2PC，原因是：
- 它在等待各方响应期间锁定资源，导致吞吐量大幅下降。
- 协调器是单点故障。
- 许多系统，如 Kafka、MongoDB 或第三方 REST API，不支持 XA 事务。
3. Saga 是什么？
Saga 将一个大事务分解为 一系列本地事务。每个本地事务在一个服务中运行，并直接提交到该服务的数据库。每个步骤都伴随一个 补偿事务，即一个业务操作来“撤销”该步骤。
| 步骤 | 操作 | 补偿 |
|---|---|---|
| 1 | 保留货物 (reserve) |
释放货物 (release) |
| 2 | 扣费 (charge) |
退款 (refund) |
| 3 | 创建发货单 (ship) |
取消发货单 (cancel) |
如果第 n 步失败，saga 以 相反的顺序 运行步骤 n-1, n-2, …, 1 的补偿。
与真正的回滚相比有两个重要区别：
- 补偿 不会删除历史。它创建一个新的变更，例如退款事务。在短时间内，其他服务可能已经看到了中间状态。
- Saga 保证 最终一致性，但不保证像 ACID 那样的隔离性。
有两种方式来协调 saga：
- 编舞（Choreography）： 每个服务监听事件并自行决定下一步。没有中央协调器。
- 编排（Orchestration）： 编排器按顺序调用每个服务，并负责在出现错误时运行补偿。
4. 用 Java 21 实现编排型 Saga
本节构建一个纯 Java 编排器，不依赖任何框架，以便你清楚地了解内部机制。
4.1. 定义一个 Saga 步骤
import java.util.function.Consumer;
public interface SagaStep<C> {
String name();
void execute(C context);
void compensate(C context);
static <C> SagaStep<C> of(String name, Consumer<C> action, Consumer<C> compensation) {
return new SagaStep<>() {
public String name() { return name; }
public void execute(C ctx) { action.accept(ctx); }
public void compensate(C ctx) { compensation.accept(ctx); }
};
}
}每个步骤有两个行为：execute 用于正向流程，compensate 用于补偿流程。泛型 C 是各步骤共享的上下文，例如订单信息。工厂方法 of 帮助你用 lambda 简洁地声明步骤。
4.2. 编排器
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.List;
public record SagaResult(boolean success, String failedStep, String reason) {
static SagaResult ok() { return new SagaResult(true, null, null); }
static SagaResult failed(String step, String reason) { return new SagaResult(false, step, reason); }
}
public class SagaOrchestrator<C> {
private final List<SagaStep<C>> steps;
public SagaOrchestrator(List<SagaStep<C>> steps) {
this.steps = List.copyOf(steps);
}
public SagaResult run(C context) {
Deque<SagaStep<C>> completed = new ArrayDeque<>();
for (SagaStep<C> step : steps) {
try {
step.execute(context);
completed.push(step);
} catch (RuntimeException ex) {
compensate(completed, context);
return SagaResult.failed(step.name(), ex.getMessage());
}
}
return SagaResult.ok();
}
private void compensate(Deque<SagaStep<C>> completed, C context) {
while (!completed.isEmpty()) {
SagaStep<C> step = completed.pop();
try {
step.compensate(context);
} catch (RuntimeException ex) {
// Production: 保存以便重试，不能忽略
System.err.println("Compensation failed at " + step.name() + ": " + ex.getMessage());
}
}
}
}主要要点：
completed是一个 栈（使用push/pop的ArrayDeque）。每个成功的步骤都被压入栈，所以弹出时我们得到正确的逆序。- 失败的步骤 不在 栈中。该步骤尚未提交任何内容，因此无需补偿。
- 补偿失败不应该停止其他补偿。因此每个
compensate调用都被包装在自己的try/catch中。
4.3. 模拟服务
import java.math.BigDecimal;
import java.util.*;
public record Order(String id, String customerId, String sku, int quantity, BigDecimal amount) {}
public class InventoryService {
private final Map<String, Integer> stock = new HashMap<>(Map.of("BOOK-01", 10));
public void reserve(Order o) {
int left = stock.get(o.sku()) - o.quantity();
if (left < 0) throw new IllegalStateException("Out of stock");
stock.put(o.sku(), left);
}
public void release(Order o) { stock.merge(o.sku(), o.quantity(), Integer::sum); }
public int available(String sku) { return stock.get(sku); }
}
public class PaymentService {
private final Map<String, BigDecimal> balances = new HashMap<>(Map.of("alice", new BigDecimal("100")));
public void charge(Order o) {
BigDecimal balance = balances.get(o.customerId());
if (balance.compareTo(o.amount()) < 0) throw new IllegalStateException("Insufficient funds");
balances.put(o.customerId(), balance.subtract(o.amount()));
}
public void refund(Order o) { balances.merge(o.customerId(), o.amount(), BigDecimal::add); }
public BigDecimal balance(String customerId) { return balances.get(customerId); }
}
public class ShippingService {
private final Set<String> shipments = new HashSet<>();
public void ship(Order o) { shipments.add(o.id()); }
public void cancel(Order o) { shipments.remove(o.id()); }
public boolean hasShipment(String orderId) { return shipments.contains(orderId); }
}实际上，每个上述方法都是 REST/gRPC 调用或发送给另一个服务的消息。这个模拟实现仅用于演示和测试。
4.4. 组装订单 Saga
import java.math.BigDecimal;
import java.util.List;
public class OrderSagaDemo {
public static void main(String[] args) {
var inventory = new InventoryService();
var payment = new PaymentService();
var shipping = new ShippingService();
var saga = new SagaOrchestrator<Order>(List.of(
SagaStep.of("reserve-stock", inventory::reserve, inventory::release),
SagaStep.of("charge-payment", payment::charge, payment::refund),
SagaStep.of("create-shipment", shipping::ship, shipping::cancel)
));
var cheap = new Order("ORD-1", "alice", "BOOK-01", 2, new BigDecimal("40"));
var pricey = new Order("ORD-2", "alice", "BOOK-01", 3, new BigDecimal("500"));
System.out.println(saga.run(cheap));
System.out.println(saga.run(pricey));
System.out.println("Stock: " + inventory.available("BOOK-01"));
System.out.println("Balance: " + payment.balance("alice"));
}
}结果：
SagaResult[success=true, failedStep=null, reason=null]
SagaResult[success=false, failedStep=charge-payment, reason=Insufficient funds]
Stock: 8
Balance: 60订单 ORD-1 完成了全部三个步骤。订单 ORD-2 保留了 3 本书，但在扣费步骤失败。编排器随后调用 inventory.release，因此库存回到 8（10 − 第一个订单的 2）而不是 5。
5. 使用事件的编舞型 Saga
在编舞型中，没有服务“知道”整个流程。每个服务在完成工作后发出事件，其他服务响应这些事件。下面的示例在单个进程中使用 Spring 的 ApplicationEventPublisher 来演示。在真实系统中，你可以用 Kafka 或 RabbitMQ 替换它。
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>public record StockReserved(Order order) {}
public record PaymentFailed(Order order, String reason) {}
@Component
public class PaymentHandler {
private final PaymentService payment;
private final ApplicationEventPublisher events;
public PaymentHandler(PaymentService payment, ApplicationEventPublisher events) {
this.payment = payment;
this.events = events;
}
@EventListener
public void on(StockReserved event) {
try {
payment.charge(event.order());
} catch (IllegalStateException ex) {
events.publishEvent(new PaymentFailed(event.order(), ex.getMessage()));
}
}
}
@Component
public class InventoryCompensationHandler {
private final InventoryService inventory;
public InventoryCompensationHandler(InventoryService inventory) {
this.inventory = inventory;
}
@EventListener
public void on(PaymentFailed event) {
inventory.release(event.order());
}
}PaymentHandler 监听 StockReserved。如果扣费失败，它发布 PaymentFailed。反过来，InventoryCompensationHandler 接收此事件并释放货物。为了使代码能够运行，InventoryService 和 PaymentService 也需要声明为 bean。
两种方式的快速比较：
- 编舞 当流程只有 2-3 步时简单且耦合度低。当流程更长时，逻辑分散在多个服务中，难以跟踪且容易形成事件循环。
- 编排 将流程集中在一个地方，易于阅读、测试和维护状态。代价是编排器成为需要谨慎运维的关键组件。
6. 测试补偿行为
添加测试依赖：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>import org.junit.jupiter.api.Test;
import java.math.BigDecimal;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
class OrderSagaTest {
private final InventoryService inventory = new InventoryService();
private final PaymentService payment = new PaymentService();
private final ShippingService shipping = new ShippingService();
private final SagaOrchestrator<Order> saga = new SagaOrchestrator<>(List.of(
SagaStep.of("reserve-stock", inventory::reserve, inventory::release),
SagaStep.of("charge-payment", payment::charge, payment::refund),
SagaStep.of("create-shipment", shipping::ship, shipping::cancel)
));
@Test
void givenInsufficientFunds_whenRunSaga_thenStockIsReleased() {
var order = new Order("ORD-9", "alice", "BOOK-01", 3, new BigDecimal("500"));
SagaResult result = saga.run(order);
assertThat(result.success()).isFalse();
assertThat(result.failedStep()).isEqualTo("charge-payment");
assertThat(inventory.available("BOOK-01")).isEqualTo(10);
assertThat(payment.balance("alice")).isEqualByComparingTo("100");
assertThat(shipping.hasShipment("ORD-9")).isFalse();
}
@Test
void givenValidOrder_whenRunSaga_thenAllStepsCommitted() {
var order = new Order("ORD-10", "alice", "BOOK-01", 1, new BigDecimal("30"));
assertThat(saga.run(order).success()).isTrue();
assertThat(inventory.available("BOOK-01")).isEqualTo(9);
assertThat(payment.balance("alice")).isEqualByComparingTo("70");
assertThat(shipping.hasShipment("ORD-10")).isTrue();
}
}第一个测试检查 Saga 最重要的事情：失败后，所有副作用都已被撤销。库存回到 10，余额保持不变，没有运单被创建。第二个测试确认成功流程提交了全部三个步骤。
7. 常见错误和如何避免
注意： 本文中的编排器将状态保存在内存中。如果进程中途崩溃，你不知道进行到哪一步，也无法运行补偿。在生产环境中，应在每一步之后将 Saga 状态（当前步骤、已完成步骤）保存到数据库，并在重启时允许恢复。
补偿必须幂等。 消息可能被重新发送多次，因此运行两次 refund 不应该退款两次。保存已处理的交易的 orderId 并忽略重复请求。
补偿不应该“放弃”。 第 4.2 节中的 catch 日志仅用于演示。在实际应用中，失败的补偿应该使用退避重试。多次失败后应推送到死信队列以供人工处理。
双写问题。 如果服务既提交数据库又向 Kafka 发送消息，其中一个操作可能失败。使用 事务性发件箱（Transactional Outbox）：在同一本地事务中将事件写入 outbox 表，然后另一个进程读取该表并发布。
隔离性缺失。 在扣费和退款之间，用户可能会看到余额暂时减少。你可以通过 语义锁 来减少影响，例如在 Saga 完成之前将订单状态设置为 PENDING。另一种方法是重新排列步骤，把最容易失败的步骤放在前面。
有些步骤无法补偿。 发出的电子邮件无法撤销。应将这类步骤（称为 枢纽 或 可重试步骤）放在 Saga 末尾，即在所有可能失败的步骤成功之后。
何时不应该使用 Saga：
- 所有数据都在一个数据库中。此时只需使用本地事务即可。
- 业务需要绝对的强一致性，不允许任何中间状态，例如某些核心会计分类账业务。此时应重新考虑服务边界的划分。
- 流程中只有一个服务写入数据，其他服务仅读取。
8. 结论
Saga 用一系列伴随补偿事务的本地事务取代分布式事务，从而使系统在无需 2PC 的情况下达到最终一致性。你已经构建了一个小型编排器，使用栈以反向顺序进行补偿，看到了编舞型如何通过事件进行协调，并编写了测试以确保出现错误时所有副作用都被撤销。
将 Saga 投入生产时，记住四件事：持久化保存 Saga 状态、编写幂等补偿并支持重试、使用事务性发件箱避免双写、以及设计步骤顺序使得无法补偿的操作位于流程末尾。
同主题文章
服务间通信：REST、gRPC 还是消息队列？
比较 REST、gRPC 和消息队列在微服务架构中的应用，包含具体 Java 示例，助你为不同场景选择合适的工具。
单体还是微服务:如何正确选择
分析在单体和微服务之间选择的实际标准,并附带演示各架构运维成本的代码示例。