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

高度细化的SAGA模式实现：基于Spring Boot与RabbitMQ的跨服务事务

article 2024/11/20 17:34:30

场景与技术栈

场景：电商系统中的订单创建流程，涉及订单服务（Order Service）、库存服务（Inventory Service）、支付服务（Payment Service）。

技术栈：

Java 11
Spring Boot 2.7.0
MySQL 8.0
RabbitMQ

项目结构与组件

order-service
inventory-service
payment-service
saga-coordinator

代码实现

1. Saga Coordinator

@SpringBootApplicationpublic class SagaCoordinatorApplication {    public static void main(String[] args) {        SpringApplication.run(SagaCoordinatorApplication.class, args);    }}@Service@RequiredArgsConstructorpublic class SagaService {    private final SagaOrderService sagaOrderService;    private final SagaInventoryService sagaInventoryService;    private final SagaPaymentService sagaPaymentService;    private final SagaTransactionRepository sagaTransactionRepository;    public void placeOrder(Long orderId, Long productId, int quantity, double amount) throws ExecutionException, InterruptedException {        SagaTransaction transaction = new SagaTransaction();        transaction.setOrderId(orderId);        transaction.setProductId(productId);        transaction.setQuantity(quantity);        transaction.setAmount(amount);        transaction.setStatus(SagaTransaction.Status.INITIATED);        sagaTransactionRepository.save(transaction);        CompletableFuture.runAsync(() -> sagaOrderService.createOrder(transaction))            .thenRunAsync(() -> sagaInventoryService.decreaseInventory(transaction))            .thenRunAsync(() -> sagaPaymentService.charge(transaction))            .exceptionally(ex -> {                rollback(transaction);                throw new RuntimeException("Saga failed", ex);            });    }    private void rollback(SagaTransaction transaction) {        CompletableFuture.runAsync(() -> sagaPaymentService.refund(transaction))            .thenRunAsync(() -> sagaInventoryService.increaseInventory(transaction))            .thenRunAsync(() -> sagaOrderService.cancelOrder(transaction))            .exceptionally(ex -> {                transaction.setStatus(SagaTransaction.Status.FAILED);                sagaTransactionRepository.save(transaction);                throw new RuntimeException("Rollback failed", ex);            });    }}

2. SagaOrderService

@Servicepublic class SagaOrderService {    private final OrderService orderService;    private final RabbitTemplate rabbitTemplate;    @Autowired    public SagaOrderService(OrderService orderService, RabbitTemplate rabbitTemplate) {        this.orderService = orderService;        this.rabbitTemplate = rabbitTemplate;    }    public void createOrder(SagaTransaction transaction) {        // 创建订单逻辑        // ...        // 发送创建订单完成的消息        rabbitTemplate.convertAndSend("order.create", transaction);    }    public void cancelOrder(SagaTransaction transaction) {        // 取消订单逻辑        // ...        // 发送取消订单完成的消息        rabbitTemplate.convertAndSend("order.cancel", transaction);    }}

3. SagaInventoryService

@Servicepublic class SagaInventoryService {    private final InventoryService inventoryService;    private final RabbitTemplate rabbitTemplate;    @Autowired    public SagaInventoryService(InventoryService inventoryService, RabbitTemplate rabbitTemplate) {        this.inventoryService = inventoryService;        this.rabbitTemplate = rabbitTemplate;    }    public void decreaseInventory(SagaTransaction transaction) {        // 扣减库存逻辑        // ...        // 发送扣减库存完成的消息        rabbitTemplate.convertAndSend("inventory.decrease", transaction);    }    public void increaseInventory(SagaTransaction transaction) {        // 增加库存逻辑        // ...        // 发送增加库存完成的消息        rabbitTemplate.convertAndSend("inventory.increase", transaction);    }}

4. SagaPaymentService

@Servicepublic class SagaPaymentService {    private final PaymentService paymentService;    private final RabbitTemplate rabbitTemplate;    @Autowired    public SagaPaymentService(PaymentService paymentService, RabbitTemplate rabbitTemplate) {        this.paymentService = paymentService;        this.rabbitTemplate = rabbitTemplate;    }    public void charge(SagaTransaction transaction) {        // 收款逻辑        // ...        // 发送收款完成的消息        rabbitTemplate.convertAndSend("payment.charge", transaction);    }    public void refund(SagaTransaction transaction) {        // 退款逻辑        // ...        // 发送退款完成的消息        rabbitTemplate.convertAndSend("payment.refund", transaction);    }}

SagaTransaction Entity

@Entitypublic class SagaTransaction {    @Id    @GeneratedValue(strategy = GenerationType.IDENTITY)    private Long id;    private Long orderId;    private Long productId;    private int quantity;    private double amount;    private Status status; // INITIATED, COMPLETED, ROLLING_BACK, ROLLED_BACK, FAILED    // Getters & Setters}

消息队列监听

在每个微服务中，需要添加消息队列的监听器，以便在接收到消息时执行相应的操作。例如，在order-service中：

@Componentpublic class OrderSagaListener {    private final SagaOrderService sagaOrderService;    @Autowired    public OrderSagaListener(SagaOrderService sagaOrderService) {        this.sagaOrderService = sagaOrderService;    }    @RabbitListener(queues = "order.create")    public void handleOrderCreate(SagaTransaction transaction) {        sagaOrderService.createOrder(transaction);    }    @RabbitListener(queues = "order.cancel")    public void handleOrderCancel(SagaTransaction transaction) {        sagaOrderService.cancelOrder(transaction);    }}