当前位置：首页 > news >正文

编排式 Saga 模式

news 2025/1/8 10:37:43

编排式 Saga 模式（Orchestrated Saga）是指由一个中央协调者（Orchestrator）控制多个服务间的事务执行。与协作式 Saga 模式不同，编排式 Saga 模式不依赖于事件驱动，而是通过协调者来控制整个 Saga 流程的执行。协调者负责调用各个参与服务，确保每个子事务按顺序执行，并在某个子事务失败时触发补偿操作。

编排式 Saga 模式实现步骤

我们将实现一个基于编排式 Saga 模式的跨服务数据保存方案，使用 Spring Boot 来开发微服务，使用 RabbitMQ 或 Kafka 作为消息队列进行通信，并通过一个中央协调者来管理整个 Saga 流程。

以下是如何使用编排式 Saga 模式来实现一个典型的跨服务操作：订单创建和库存扣减。

1. 架构设计

我们有两个微服务：

Order Service：负责创建订单。
Inventory Service：负责管理库存。
Saga Orchestrator Service：协调整个 Saga 流程，包括执行各服务事务并在失败时触发补偿操作。

2. 技术栈

Spring Boot：用于开发微服务。
Spring Cloud：用于服务注册、发现和治理。
Spring AMQP / Kafka：用于服务间消息传递（可选择 RabbitMQ 或 Kafka）。
Spring Data JPA：用于数据库操作。
Transactional Outbox Pattern：用来确保跨服务操作的一致性。

3. 系统流程

Order Service：接收创建订单请求，调用 Saga Orchestrator Service 开始 Saga 流程。
Saga Orchestrator：协调 Inventory Service 扣减库存，等到确认成功后，继续后续操作（如创建订单）。
Inventory Service：接收扣减库存请求，执行库存扣减，如果成功，通知 Saga Orchestrator。如果失败，则触发补偿操作。
补偿操作：如果任何一个服务的事务失败，Saga Orchestrator 会调用补偿操作回滚之前的事务，确保最终一致性。

4. Spring Boot 示例实现

4.1 创建 Order Service

Order Service 负责处理订单请求，并与 Saga Orchestrator 配合，触发 Saga 流程。

// OrderService.java
@Service
public class OrderService {@Autowiredprivate SagaOrchestrator sagaOrchestrator;// 创建订单@Transactionalpublic void createOrder(Order order) {// Step 1: 创建订单orderRepository.save(order);// Step 2: 调用 Saga Orchestrator 开始整个流程sagaOrchestrator.startSaga(order);}
}

4.2 创建 Saga Orchestrator Service

Saga Orchestrator Service 是整个 Saga 模式的核心，它负责协调各个服务之间的事务执行。首先，它会启动 Saga 事务，接着协调 Inventory Service 执行库存扣减操作，并处理补偿操作。

// SagaOrchestrator.java
@Service
public class SagaOrchestrator {@Autowiredprivate InventoryService inventoryService;@Autowiredprivate OrderRepository orderRepository;// 启动 Saga 流程@Transactionalpublic void startSaga(Order order) {try {// Step 1: 调用库存服务扣减库存boolean inventorySuccess = inventoryService.decreaseInventory(order.getItemId(), order.getQuantity());if (!inventorySuccess) {throw new Exception("Inventory insufficient");}// Step 2: 库存扣减成功后，继续创建订单order.setStatus("Created");orderRepository.save(order);} catch (Exception e) {// Step 3: 如果出错，执行补偿操作compensate(order);}}// 补偿方法，回滚库存操作private void compensate(Order order) {// 回滚库存，增加库存inventoryService.rollbackInventory(order.getItemId(), order.getQuantity());// 回滚订单，设置订单为失败状态order.setStatus("Failed");orderRepository.save(order);}
}

4.3 创建 Inventory Service

Inventory Service 负责扣减库存并通知 Saga Orchestrator 执行后续操作。

// InventoryService.java
@Service
public class InventoryService {@Autowiredprivate InventoryRepository inventoryRepository;// 扣减库存@Transactionalpublic boolean decreaseInventory(Long itemId, int quantity) {Inventory inventory = inventoryRepository.findByItemId(itemId);if (inventory.getStock() < quantity) {return false; // 库存不足}inventory.setStock(inventory.getStock() - quantity);inventoryRepository.save(inventory);return true; // 库存扣减成功}// 补偿操作，回滚库存@Transactionalpublic void rollbackInventory(Long itemId, int quantity) {Inventory inventory = inventoryRepository.findByItemId(itemId);inventory.setStock(inventory.getStock() + quantity); // 恢复库存inventoryRepository.save(inventory);}
}

4.4 消息队列（RabbitMQ 或 Kafka）集成

为了实现 Saga 模式的跨服务通信，我们可以使用消息队列来传递消息。这里我们使用 RabbitMQ 作为消息队列。

在 application.properties 中配置 RabbitMQ：

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/

在 SagaOrchestrator 和 InventoryService 中，我们可以通过 RabbitTemplate 来发送和接收消息。

// SagaOrchestrator.java
@Autowired
private RabbitTemplate rabbitTemplate;// 启动 Saga 流程时，发送事件
public void startSaga(Order order) {// 发送一个消息，通知库存服务处理库存rabbitTemplate.convertAndSend("inventoryExchange", "inventory.decrease", order);
}// 监听库存扣减消息的回调
@RabbitListener(queues = "inventory.decrease.queue")
public void handleInventoryDecrease(Order order) {try {// 扣减库存并继续订单处理boolean inventorySuccess = inventoryService.decreaseInventory(order.getItemId(), order.getQuantity());if (!inventorySuccess) {throw new Exception("Inventory insufficient");}// 订单处理继续order.setStatus("Created");orderRepository.save(order);} catch (Exception e) {// 执行补偿操作compensate(order);}
}

4.5 设置消息队列的交换机和队列

@Configuration
public class RabbitMQConfig {@Beanpublic TopicExchange inventoryExchange() {return new TopicExchange("inventoryExchange");}@Beanpublic Queue inventoryDecreaseQueue() {return new Queue("inventory.decrease.queue");}@Beanpublic Binding inventoryDecreaseBinding() {return BindingBuilder.bind(inventoryDecreaseQueue()).to(inventoryExchange()).with("inventory.decrease");}@Beanpublic Jackson2JsonMessageConverter jackson2JsonMessageConverter() {return new Jackson2JsonMessageConverter();}@Beanpublic RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory,Jackson2JsonMessageConverter jackson2JsonMessageConverter) {RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);rabbitTemplate.setMessageConverter(jackson2JsonMessageConverter);return rabbitTemplate;}
}