编排式 Saga 模式

news2025/1/12 0:44:28

编排式 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 {

    @Autowired
    private SagaOrchestrator sagaOrchestrator;

    // 创建订单
    @Transactional
    public 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 {

    @Autowired
    private InventoryService inventoryService;

    @Autowired
    private OrderRepository orderRepository;

    // 启动 Saga 流程
    @Transactional
    public 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 {

    @Autowired
    private InventoryRepository inventoryRepository;

    // 扣减库存
    @Transactional
    public 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; // 库存扣减成功
    }

    // 补偿操作，回滚库存
    @Transactional
    public 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 {

    @Bean
    public TopicExchange inventoryExchange() {
        return new TopicExchange("inventoryExchange");
    }

    @Bean
    public Queue inventoryDecreaseQueue() {
        return new Queue("inventory.decrease.queue");
    }

    @Bean
    public Binding inventoryDecreaseBinding() {
        return BindingBuilder.bind(inventoryDecreaseQueue()).to(inventoryExchange()).with("inventory.decrease");
    }
    
    @Bean
    public Jackson2JsonMessageConverter jackson2JsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory,
                                         Jackson2JsonMessageConverter jackson2JsonMessageConverter) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(jackson2JsonMessageConverter);
        return rabbitTemplate;
    }
}