第6章：处理业务逻辑（Managing Business Logic）
一、导读
1.1 本章学习目标
微服务架构中最大的挑战之一是跨服务的事务管理。由于每个服务拥有独立的数据库，传统的 ACID 分布式事务（2PC）不再适用。第6章重点介绍 Saga 模式——将长事务拆解为一系列本地事务，每个本地事务都有对应的补偿操作。
本章的核心学习目标包括：
- 理解 Database-per-Service 对事务的影响：为什么传统的 ACID 事务不再适用。
- 掌握 Saga 模式的核心思想：将长事务拆分为本地事务 + 补偿操作。
- 区分 Choreography 和 Orchestration 两种 Saga 实现方式：各自的优缺点和适用场景。
- 理解 Saga 的隔离性挑战：语义锁、可重试操作、补偿事务的设计。
- 掌握事务消息模式：保证业务操作和消息发送的原子性。
- 了解 ACID 分布式事务（2PC）的局限性：为什么不推荐在微服务中使用。
1.2 为什么本章如此重要
Richardson 明确指出：跨服务的数据一致性是微服务架构中最困难的问题。在单体架构中，一个 @Transactional 注解就能保证多个操作的原子性。但在微服务中，一个业务操作可能需要跨越 3-4 个服务，每个服务有自己的数据库，传统的分布式事务（2PC）因为性能差、可用性低而不适用于微服务。
Saga 模式是目前微服务中处理跨服务事务的主流方案。理解 Saga 的设计原则和实现方式，是构建可靠微服务的关键。
1.3 核心问题引导
为什么 2PC（两阶段提交）不适合微服务架构？
Saga 的补偿操作如何设计？如果补偿操作本身失败了怎么办？
Choreography 和 Orchestration 两种 Saga 各有什么优缺点？
Saga 的隔离性问题如何解决？语义锁是什么？
事务消息（Transactional Messaging）如何保证业务操作和消息发送的原子性？
什么场景下可以使用"最终一致性"代替"强一致性"？
二、核心概念详解
2.1 分布式事务的挑战
2.1.1 为什么不能用 2PC
两阶段提交（2PC）是传统分布式事务的标准方案，但在微服务中有严重缺陷：
// 2PC 的工作流程：
// 阶段1（Prepare）：协调者询问所有参与者是否可以提交
// 阶段2（Commit/Rollback）：
// - 如果所有参与者都回答"是" → 协调者发送 Commit
// - 如果任一参与者回答"否" → 协调者发送 Rollback
// 2PC 的问题：
// 1. 同步阻塞：所有参与者在 Prepare 后必须持有锁等待 Commit
// 2. 单点故障：协调者故障会导致所有参与者阻塞
// 3. 性能差：跨数据库的锁持有时间长，吞吐量低
// 4. 可用性低：任何一个服务不可用，整个事务都无法完成
// 5. 不适合异构数据库：不同服务可能使用不同类型的数据库
// 结论：2PC 不适合微服务架构2.1.2 Saga 的核心思想
Saga 将一个长事务拆分为一系列本地事务。每个本地事务更新一个服务的数据库，并发布消息触发下一个本地事务。如果某个本地事务失败，Saga 执行之前成功事务的补偿操作。
// 创建订单的 Saga：
// 步骤1: OrderService.createOrder()
// → 本地事务：在 orders_db 中创建订单
// → 发布事件：OrderCreated
//
// 步骤2: PaymentService.processPayment()
// → 本地事务：在 payments_db 中创建支付记录
// → 发布事件：PaymentCompleted
//
// 步骤3: InventoryService.reserveStock()
// → 本地事务：在 inventory_db 中预留库存
// → 发布事件：StockReserved
//
// 步骤4: OrderService.confirmOrder()
// → 本地事务：更新订单状态为 CONFIRMED
// → 发布事件：OrderConfirmed
// 补偿流程（如果步骤3失败）：
// 补偿步骤2: PaymentService.refund()
// → 本地事务：在 payments_db 中标记退款
// 补偿步骤1: OrderService.cancelOrder()
// → 本地事务：在 orders_db 中标记订单取消2.2 Choreography Saga
2.2.1 工作原理
Choreography Saga 没有中央协调器——每个服务在完成自己的本地事务后，发布事件触发下一个服务的操作。服务之间通过事件"协作"完成整个流程。
// Choreography Saga：订单创建流程
//
// [OrderService]
// 1. 创建订单 → 发布 OrderCreated
// ↓
// [PaymentService] 监听 OrderCreated
// 2. 处理支付 → 发布 PaymentCompleted
// ↓
// [InventoryService] 监听 PaymentCompleted
// 3. 预留库存 → 发布 StockReserved
// ↓
// [OrderService] 监听 StockReserved
// 4. 确认订单 → 发布 OrderConfirmed
//
// 补偿流程（如果步骤3失败）：
// [InventoryService] 发布 StockReservationFailed
// ↓
// [PaymentService] 监听 StockReservationFailed
// 补偿步骤2: 退款 → 发布 PaymentRefunded
// ↓
// [OrderService] 监听 PaymentRefunded
// 补偿步骤1: 取消订单 → 发布 OrderCancelled
// Choreography Saga 的优点：
// 1. 简单：不需要额外的协调组件
// 2. 松耦合：服务之间只知道事件，不知道其他服务
// 3. 适合简单的流程（3-4步）
// Choreography Saga 的缺点：
// 1. 流程难以追踪：业务逻辑分散在多个服务中
// 2. 循环依赖风险：服务之间可能形成事件循环
// 3. 不适合复杂流程：步骤多时，事件链难以管理2.3 Orchestration Saga
2.3.1 工作原理
Orchestration Saga 有一个中央协调器（Orchestrator），由它统一指挥各服务执行操作。协调器维护整个 Saga 的状态，决定下一步执行哪个操作，以及失败时执行哪些补偿。
// Orchestration Saga：订单创建流程
//
// [OrderSagaOrchestrator]
// 状态：STARTED
// 步骤1: 调用 OrderService.createOrder()
// → 成功：状态 = ORDER_CREATED
// 步骤2: 调用 PaymentService.processPayment()
// → 成功：状态 = PAYMENT_COMPLETED
// 步骤3: 调用 InventoryService.reserveStock()
// → 成功：状态 = STOCK_RESERVED
// 步骤4: 调用 OrderService.confirmOrder()
// → 成功：状态 = COMPLETED
//
// 补偿流程（如果步骤3失败）：
// 状态：COMPENSATING
// 补偿步骤2: 调用 PaymentService.refund()
// 补偿步骤1: 调用 OrderService.cancelOrder()
// 状态：COMPENSATED
// Orchestration Saga 的优点：
// 1. 流程清晰：所有逻辑在协调器中，易于理解和追踪
// 2. 无循环依赖：服务只与协调器通信
// 3. 适合复杂流程：可以处理条件分支、并行步骤
// Orchestration Saga 的缺点：
// 1. 额外的协调组件：增加了系统复杂度
// 2. 协调器可能成为单点：需要高可用设计
// 3. 服务之间通过协调器耦合2.3.2 使用 Temporal 实现 Saga 编排
typescript
// Temporal Workflow 实现 Saga 编排
import { proxyActivities } from '@temporalio/workflow';
const activities = proxyActivities<typeof import('./activities')>({
startToCloseTimeout: '30 seconds',
retry: { maximumAttempts: 3 }
});
export async function createOrderSaga(orderId: string) {
try {
const order = await activities.createOrder(orderId);
const payment = await activities.processPayment(orderId);
const stock = await activities.reserveStock(orderId);
const confirmed = await activities.confirmOrder(orderId);
return confirmed;
} catch (error) {
// 自动执行补偿
await compensate(orderId);
throw error;
}
}
async function compensate(orderId: string) {
try { await activities.refundPayment(orderId); } catch (e) { /* log */ }
try { await activities.cancelOrder(orderId); } catch (e) { /* log */ }
}2.4 Saga 的隔离性
2.4.1 隔离性挑战
Saga 不具备传统事务的隔离性——在 Saga 执行过程中，其他事务可以看到部分完成的状态。这可能导致语义上的问题。
// 场景：两个用户同时抢购最后一件商品
// Saga A: 用户A购买 → 步骤1创建订单 → 步骤2预留库存
// Saga B: 用户B购买 → 步骤1创建订单 → 步骤2预留库存（库存不足！）
//
// 问题：Saga A 的步骤1完成后，Saga B 的步骤1也完成了
// 两个订单都创建了，但只有一件库存
// Saga B 的步骤2失败 → 补偿步骤1取消订单
// 用户体验差：下单成功但最终被取消
// 解决方案：语义锁
// 在步骤1创建订单时，将库存项标记为"预留中"
// Saga B 的步骤1检测到"预留中"标记 → 等待或失败2.4.2 语义锁
java
// 语义锁实现
public class OrderService {
public Order createOrder(CreateOrderCommand cmd) {
// 检查库存可用性（语义锁）
for (OrderItem item : cmd.getItems()) {
InventoryItem inv = inventoryClient.checkAvailability(item.getProductId());
if (inv.getAvailableQuantity() < item.getQuantity()) {
throw new InsufficientStockException(item.getProductId());
}
// 标记为"预留中"（语义锁）
inventoryClient.markAsPending(item.getProductId(), cmd.getOrderId());
}
Order order = Order.create(cmd);
orderRepository.save(order);
return order;
}
}2.5 事务消息
2.5.1 问题描述
如何保证"保存业务数据"和"发送消息"的原子性？如果先保存数据再发消息，可能在发消息前崩溃；如果先发消息再保存数据，可能在保存前崩溃。
2.5.2 Outbox 模式
// Outbox 模式：
// 在同一个数据库事务中，同时写入业务数据和 Outbox 事件表
// 独立的 Event Publisher 进程扫描 Outbox 表，发送消息到 Kafka
// 数据库事务：
BEGIN TRANSACTION
INSERT INTO orders (id, customer_id, status) VALUES ('123', '456', 'CREATED')
INSERT INTO outbox_events (aggregate_id, event_type, payload)
VALUES ('123', 'OrderCreated', '{"orderId":"123",...}')
COMMIT
// Event Publisher（独立进程）：
// 1. 扫描 outbox_events 中未发送的记录
// 2. 发送到 Kafka
// 3. 标记为已发送（或删除）
//
// 保证：业务数据和消息最终一致三、实战案例
3.1 案例一：Choreography Saga 实现
java
// OrderService 监听支付完成事件
@KafkaListener(topics = "payment-events", groupId = "order-service")
public void handlePaymentEvent(PaymentEvent event) {
if (event.getType().equals("PaymentCompleted")) {
Order order = orderRepository.findById(event.getOrderId());
order.setStatus(OrderStatus.PAID);
orderRepository.save(order);
eventPublisher.publish(new OrderPaidEvent(order.getId()));
} else if (event.getType().equals("PaymentRefunded")) {
Order order = orderRepository.findById(event.getOrderId());
order.setStatus(OrderStatus.CANCELLED);
orderRepository.save(order);
}
}
// PaymentService 监听订单创建事件
@KafkaListener(topics = "order-events", groupId = "payment-service")
public void handleOrderEvent(OrderEvent event) {
if (event.getType().equals("OrderCreated")) {
try {
Payment payment = paymentService.process(event.getOrderId(), event.getAmount());
eventPublisher.publish(new PaymentCompletedEvent(event.getOrderId()));
} catch (PaymentException e) {
eventPublisher.publish(new PaymentFailedEvent(event.getOrderId(), e.getMessage()));
}
}
}3.2 案例二：Orchestration Saga 实现
java
// 订单创建编排器
@Component
public class CreateOrderSagaOrchestrator {
@Autowired private OrderService orderService;
@Autowired private PaymentClient paymentClient;
@Autowired private InventoryClient inventoryClient;
public void execute(CreateOrderCommand cmd) {
SagaState state = new SagaState(cmd.getOrderId());
try {
// 步骤1：创建订单
Order order = orderService.createOrder(cmd);
state.addCompletedStep("createOrder", order.getId());
// 步骤2：处理支付
Payment payment = paymentClient.processPayment(
new PaymentRequest(order.getId(), order.getTotalAmount()));
state.addCompletedStep("processPayment", payment.getId());
// 步骤3：预留库存
inventoryClient.reserveStock(
new StockReservationRequest(order.getId(), order.getItems()));
state.addCompletedStep("reserveStock", "ok");
// 步骤4：确认订单
orderService.confirmOrder(order.getId());
state.setStatus("COMPLETED");
} catch (Exception e) {
state.setStatus("COMPENSATING");
compensate(state);
state.setStatus("COMPENSATED");
throw new SagaFailedException(e);
}
}
private void compensate(SagaState state) {
List<String> steps = state.getCompletedSteps();
Collections.reverse(steps);
for (String step : steps) {
try {
switch (step) {
case "createOrder":
orderService.cancelOrder(state.getOrderId());
break;
case "processPayment":
paymentClient.refund(state.getOrderId());
break;
case "reserveStock":
inventoryClient.releaseStock(state.getOrderId());
break;
}
} catch (Exception e) {
// 补偿失败需要人工介入
log.error("Compensation failed for step: " + step, e);
}
}
}
}四、本章小结
本章深入讲解了微服务中的业务逻辑管理和分布式事务：
核心知识点：
- 2PC 不适合微服务（同步阻塞、单点故障、性能差）。
- Saga 将长事务拆分为本地事务 + 补偿操作。
- Choreography Saga 通过事件协作，简单但流程分散。
- Orchestration Saga 由协调器统一指挥，流程清晰但增加复杂度。
- Saga 的隔离性问题通过语义锁解决。
- Outbox 模式保证业务操作和消息发送的原子性。
五、思考与练习
思考题
在你的项目中，哪些业务流程需要使用 Saga？
Choreography 和 Orchestration 两种模式你会如何选择？
补偿操作失败（如退款服务不可用）应该如何处理？
练习题
使用 Temporal 实现一个 Orchestration Saga，包含订单、支付、库存三个服务。
设计一个 Choreography Saga 的事件流，画出事件序列图。
实现 Outbox 模式的 Event Publisher。