MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Saga 模式在电商分布式系统中的应用案例

2024-05-054.6k 阅读

1. 电商分布式系统概述

在当今数字化时代，电商平台的规模和复杂性不断增长。传统的单体架构难以应对高并发、海量数据以及业务快速迭代的需求，因此分布式系统成为电商领域的首选架构模式。分布式系统将整个业务拆分成多个独立的服务，每个服务可以独立开发、部署和扩展，从而提高系统的可维护性、可扩展性和性能。

例如，一个典型的电商分布式系统可能包含用户服务、商品服务、订单服务、支付服务、库存服务等。这些服务之间通过网络进行通信，协同完成复杂的电商业务流程，如用户下单、支付、库存扣减等。然而，分布式系统引入了新的挑战，其中之一就是分布式事务管理。

2. 分布式事务问题

在单体应用中，事务管理相对简单，通过数据库的本地事务即可保证数据的一致性。但在分布式系统中，涉及多个服务和多个数据库，传统的本地事务无法满足需求。例如，在电商下单流程中，可能需要调用订单服务创建订单、库存服务扣减库存、支付服务处理支付等操作，这些操作分布在不同的服务和数据库中，需要保证要么所有操作都成功，要么所有操作都回滚，以确保数据的一致性。

分布式事务的常见解决方案包括两阶段提交（2PC）、三阶段提交（3PC）和Saga模式等。2PC虽然能保证强一致性，但存在单点故障、性能瓶颈等问题；3PC虽然在一定程度上解决了2PC的单点故障问题，但实现复杂，且在网络分区等异常情况下仍难以保证一致性。相比之下，Saga模式提供了一种更灵活、更适合高并发和复杂业务场景的分布式事务解决方案。

3. Saga模式简介

Saga模式由Hector Garcia - Molina和Kenneth Salem在1987年提出。其核心思想是将一个长事务拆分成多个本地事务，每个本地事务都有对应的补偿事务。当整个长事务中的某个本地事务失败时，Saga模式会按照顺序调用已经执行成功的本地事务的补偿事务，将系统恢复到事务开始前的状态。

以电商下单流程为例，假设下单流程包含创建订单、扣减库存、处理支付三个本地事务。如果在处理支付时失败，Saga模式会调用扣减库存的补偿事务（即增加库存）和创建订单的补偿事务（即删除订单），从而保证数据的一致性。

Saga模式的执行有两种方式：

  • 顺序执行：按照事务的顺序依次执行每个本地事务。如果某个事务失败，从失败的事务开始逆向执行补偿事务。
  • 并行执行：可以并行执行部分没有依赖关系的本地事务，以提高执行效率。但在并行执行时，需要更复杂的协调机制来确保事务的一致性。

4. Saga模式在电商分布式系统中的应用场景

4.1 订单创建与库存扣减

在电商下单流程中，首先要创建订单记录，然后扣减库存。如果库存不足，需要回滚订单创建操作。使用Saga模式，创建订单是一个本地事务，扣减库存是另一个本地事务。当扣减库存失败时，调用创建订单的补偿事务（删除订单记录）。

4.2 支付与订单状态更新

支付成功后，需要更新订单状态为已支付。如果支付过程中出现问题，需要回滚订单状态更新操作。支付操作和订单状态更新分别作为本地事务，当支付失败时，调用订单状态更新的补偿事务（恢复订单原来的状态）。

4.3 退款流程

当用户发起退款时，需要先检查订单状态是否允许退款，然后退还支付金额，最后增加库存。如果其中任何一步失败，需要回滚前面已经执行的操作。退款流程可以看作是一个Saga事务，每个步骤作为一个本地事务，并有相应的补偿事务。

5. 代码示例（以Java和Spring Boot为例）

5.1 项目结构

假设我们有一个简单的电商分布式系统示例，包含订单服务、库存服务和支付服务。项目结构如下：

ecommerce - saga
├── order - service
│   ├── src
│   │   ├── main
│   │   │   ├── java
│   │   │   │   ├── com
│   │   │   │   │   ├── example
│   │   │   │   │   │   ├── orderservice
│   │   │   │   │   │   │   ├── controller
│   │   │   │   │   │   │   │   ├── OrderController.java
│   │   │   │   │   │   │   ├── service
│   │   │   │   │   │   │   │   ├── OrderService.java
│   │   │   │   │   │   │   │   ├── OrderSagaService.java
│   │   │   │   │   │   │   ├── repository
│   │   │   │   │   │   │   │   ├── OrderRepository.java
│   │   │   │   │   │   │   ├── entity
│   │   │   │   │   │   │   │   ├── Order.java
│   │   │   ├── resources
│   │   │   │   ├── application.properties
│   │   │   ├── static
│   │   │   ├── templates
│   │   ├── test
│   │       ├── java
│   │       ├── resources
├── inventory - service
│   ├── src
│   │   ├── main
│   │   │   ├── java
│   │   │   │   ├── com
│   │   │   │   │   ├── example
│   │   │   │   │   │   ├── inventoryservice
│   │   │   │   │   │   │   ├── controller
│   │   │   │   │   │   │   │   ├── InventoryController.java
│   │   │   │   │   │   │   ├── service
│   │   │   │   │   │   │   │   ├── InventoryService.java
│   │   │   │   │   │   │   │   ├── InventorySagaService.java
│   │   │   │   │   │   │   ├── repository
│   │   │   │   │   │   │   │   ├── InventoryRepository.java
│   │   │   │   │   │   │   ├── entity
│   │   │   │   │   │   │   │   ├── Inventory.java
│   │   │   ├── resources
│   │   │   │   ├── application.properties
│   │   │   ├── static
│   │   │   ├── templates
│   │   ├── test
│   │       ├── java
│   │       ├── resources
├── payment - service
│   ├── src
│   │   ├── main
│   │   │   ├── java
│   │   │   │   ├── com
│   │   │   │   │   ├── example
│   │   │   │   │   │   ├── paymentservice
│   │   │   │   │   │   │   ├── controller
│   │   │   │   │   │   │   │   ├── PaymentController.java
│   │   │   │   │   │   │   ├── service
│   │   │   │   │   │   │   │   ├── PaymentService.java
│   │   │   │   │   │   │   │   ├── PaymentSagaService.java
│   │   │   │   │   │   │   ├── repository
│   │   │   │   │   │   │   │   ├── PaymentRepository.java
│   │   │   │   │   │   │   ├── entity
│   │   │   │   │   │   │   │   ├── Payment.java
│   │   │   ├── resources
│   │   │   │   ├── application.properties
│   │   │   ├── static
│   │   │   ├── templates
│   │   ├── test
│   │       ├── java
│   │       ├── resources
├── saga - orchestrator
│   ├── src
│   │   ├── main
│   │   │   ├── java
│   │   │   │   ├── com
│   │   │   │   │   ├── example
│   │   │   │   │   │   ├── sagaservice
│   │   │   │   │   │   │   ├── SagaOrchestrator.java
│   │   │   ├── resources
│   │   │   │   ├── application.properties
│   │   │   ├── static
│   │   │   ├── templates
│   │   ├── test
│   │       ├── java
│   │       ├── resources

5.2 订单服务

订单服务负责创建订单和处理订单相关的Saga事务。

订单实体类：

package com.example.orderservice.entity;

import javax.persistence.Entity;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;

@Entity
public class Order {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private String orderNumber;
    private double amount;
    // 订单状态，如 "created", "paid", "shipped" 等
    private String status;

    // 省略构造函数、getter 和 setter 方法
}

订单仓库接口：

package com.example.orderservice.repository;

import com.example.orderservice.entity.Order;
import org.springframework.data.jpa.repository.JpaRepository;

public interface OrderRepository extends JpaRepository<Order, Long> {
}

订单服务接口：

package com.example.orderservice.service;

import com.example.orderservice.entity.Order;

public interface OrderService {
    Order createOrder(String orderNumber, double amount);
    void cancelOrder(Long orderId);
}

订单服务实现类：

package com.example.orderservice.service;

import com.example.orderservice.entity.Order;
import com.example.orderservice.repository.OrderRepository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class OrderService implements OrderService {

    @Autowired
    private OrderRepository orderRepository;

    @Override
    public Order createOrder(String orderNumber, double amount) {
        Order order = new Order();
        order.setOrderNumber(orderNumber);
        order.setAmount(amount);
        order.setStatus("created");
        return orderRepository.save(order);
    }

    @Override
    public void cancelOrder(Long orderId) {
        orderRepository.deleteById(orderId);
    }
}

订单Saga服务接口：

package com.example.orderservice.service;

import com.example.orderservice.entity.Order;

public interface OrderSagaService {
    Order startCreateOrderSaga(String orderNumber, double amount);
    void rollbackCreateOrderSaga(Long orderId);
}

订单Saga服务实现类：

package com.example.orderservice.service;

import com.example.orderservice.entity.Order;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class OrderSagaService implements OrderSagaService {

    @Autowired
    private OrderService orderService;

    @Override
    public Order startCreateOrderSaga(String orderNumber, double amount) {
        return orderService.createOrder(orderNumber, amount);
    }

    @Override
    public void rollbackCreateOrderSaga(Long orderId) {
        orderService.cancelOrder(orderId);
    }
}

订单控制器：

package com.example.orderservice.controller;

import com.example.orderservice.entity.Order;
import com.example.orderservice.service.OrderSagaService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class OrderController {

    @Autowired
    private OrderSagaService orderSagaService;

    @PostMapping("/create - order")
    public Order createOrder(@RequestParam String orderNumber, @RequestParam double amount) {
        return orderSagaService.startCreateOrderSaga(orderNumber, amount);
    }
}

5.3 库存服务

库存服务负责扣减库存和处理库存相关的Saga事务。

库存实体类：

package com.example.inventoryservice.entity;

import javax.persistence.Entity;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;

@Entity
public class Inventory {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private String productCode;
    private int quantity;

    // 省略构造函数、getter 和 setter 方法
}

库存仓库接口：

package com.example.inventoryservice.repository;

import com.example.inventoryservice.entity.Inventory;
import org.springframework.data.jpa.repository.JpaRepository;

public interface InventoryRepository extends JpaRepository<Inventory, Long> {
    Inventory findByProductCode(String productCode);
}

库存服务接口：

package com.example.inventoryservice.service;

import com.example.inventoryservice.entity.Inventory;

public interface InventoryService {
    boolean deductInventory(String productCode, int quantity);
    void increaseInventory(String productCode, int quantity);
}

库存服务实现类：

package com.example.inventoryservice.service;

import com.example.inventoryservice.entity.Inventory;
import com.example.inventoryservice.repository.InventoryRepository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class InventoryService implements InventoryService {

    @Autowired
    private InventoryRepository inventoryRepository;

    @Override
    public boolean deductInventory(String productCode, int quantity) {
        Inventory inventory = inventoryRepository.findByProductCode(productCode);
        if (inventory != null && inventory.getQuantity() >= quantity) {
            inventory.setQuantity(inventory.getQuantity() - quantity);
            inventoryRepository.save(inventory);
            return true;
        }
        return false;
    }

    @Override
    public void increaseInventory(String productCode, int quantity) {
        Inventory inventory = inventoryRepository.findByProductCode(productCode);
        if (inventory != null) {
            inventory.setQuantity(inventory.getQuantity() + quantity);
            inventoryRepository.save(inventory);
        }
    }
}

库存Saga服务接口：

package com.example.inventoryservice.service;

import com.example.inventoryservice.entity.Inventory;

public interface InventorySagaService {
    boolean startDeductInventorySaga(String productCode, int quantity);
    void rollbackDeductInventorySaga(String productCode, int quantity);
}

库存Saga服务实现类：

package com.example.inventoryservice.service;

import com.example.inventoryservice.entity.Inventory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class InventorySagaService implements InventorySagaService {

    @Autowired
    private InventoryService inventoryService;

    @Override
    public boolean startDeductInventorySaga(String productCode, int quantity) {
        return inventoryService.deductInventory(productCode, quantity);
    }

    @Override
    public void rollbackDeductInventorySaga(String productCode, int quantity) {
        inventoryService.increaseInventory(productCode, quantity);
    }
}

库存控制器：

package com.example.inventoryservice.controller;

import com.example.inventoryservice.service.InventorySagaService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class InventoryController {

    @Autowired
    private InventorySagaService inventorySagaService;

    @PostMapping("/deduct - inventory")
    public boolean deductInventory(@RequestParam String productCode, @RequestParam int quantity) {
        return inventorySagaService.startDeductInventorySaga(productCode, quantity);
    }
}

5.4 支付服务

支付服务负责处理支付和处理支付相关的Saga事务。

支付实体类：

package com.example.paymentservice.entity;

import javax.persistence.Entity;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;

@Entity
public class Payment {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private String paymentId;
    private double amount;
    // 支付状态，如 "paid", "failed" 等
    private String status;

    // 省略构造函数、getter 和 setter 方法
}

支付仓库接口：

package com.example.paymentservice.repository;

import com.example.paymentservice.entity.Payment;
import org.springframework.data.jpa.repository.JpaRepository;

public interface PaymentRepository extends JpaRepository<Payment, Long> {
}

支付服务接口：

package com.example.paymentservice.service;

import com.example.paymentservice.entity.Payment;

public interface PaymentService {
    Payment processPayment(String paymentId, double amount);
    void refundPayment(Long paymentId);
}

支付服务实现类：

package com.example.paymentservice.service;

import com.example.paymentservice.entity.Payment;
import com.example.paymentservice.repository.PaymentRepository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class PaymentService implements PaymentService {

    @Autowired
    private PaymentRepository paymentRepository;

    @Override
    public Payment processPayment(String paymentId, double amount) {
        Payment payment = new Payment();
        payment.setPaymentId(paymentId);
        payment.setAmount(amount);
        payment.setStatus("paid");
        return paymentRepository.save(payment);
    }

    @Override
    public void refundPayment(Long paymentId) {
        paymentRepository.deleteById(paymentId);
    }
}

支付Saga服务接口：

package com.example.paymentservice.service;

import com.example.paymentservice.entity.Payment;

public interface PaymentSagaService {
    Payment startProcessPaymentSaga(String paymentId, double amount);
    void rollbackProcessPaymentSaga(Long paymentId);
}

支付Saga服务实现类：

package com.example.paymentservice.service;

import com.example.paymentservice.entity.Payment;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class PaymentSagaService implements PaymentSagaService {

    @Autowired
    private PaymentService paymentService;

    @Override
    public Payment startProcessPaymentSaga(String paymentId, double amount) {
        return paymentService.processPayment(paymentId, amount);
    }

    @Override
    public void rollbackProcessPaymentSaga(Long paymentId) {
        paymentService.refundPayment(paymentId);
    }
}

支付控制器：

package com.example.paymentservice.controller;

import com.example.paymentservice.entity.Payment;
import com.example.paymentservice.service.PaymentSagaService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class PaymentController {

    @Autowired
    private PaymentSagaService paymentSagaService;

    @PostMapping("/process - payment")
    public Payment processPayment(@RequestParam String paymentId, @RequestParam double amount) {
        return paymentSagaService.startProcessPaymentSaga(paymentId, amount);
    }
}

5.5 Saga 协调器

Saga协调器负责编排整个Saga事务。

Saga协调器接口：

package com.example.sagaservice;

import com.example.orderservice.entity.Order;
import com.example.paymentservice.entity.Payment;

public interface SagaOrchestrator {
    boolean executeOrderSaga(String orderNumber, double amount, String productCode, int quantity, String paymentId);
}

Saga协调器实现类：

package com.example.sagaservice;

import com.example.inventoryservice.service.InventorySagaService;
import com.example.orderservice.entity.Order;
import com.example.orderservice.service.OrderSagaService;
import com.example.paymentservice.entity.Payment;
import com.example.paymentservice.service.PaymentSagaService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class SagaOrchestratorImpl implements SagaOrchestrator {

    @Autowired
    private OrderSagaService orderSagaService;
    @Autowired
    private InventorySagaService inventorySagaService;
    @Autowired
    private PaymentSagaService paymentSagaService;

    @Override
    public boolean executeOrderSaga(String orderNumber, double amount, String productCode, int quantity, String paymentId) {
        try {
            // 创建订单
            Order order = orderSagaService.startCreateOrderSaga(orderNumber, amount);
            // 扣减库存
            if (!inventorySagaService.startDeductInventorySaga(productCode, quantity)) {
                rollbackOrderSaga(order.getId());
                return false;
            }
            // 处理支付
            Payment payment = paymentSagaService.startProcessPaymentSaga(paymentId, amount);
            return true;
        } catch (Exception e) {
            // 捕获异常并回滚Saga
            e.printStackTrace();
            return false;
        }
    }

    private void rollbackOrderSaga(Long orderId) {
        // 回滚支付
        paymentSagaService.rollbackProcessPaymentSaga(null);
        // 回滚库存扣减
        inventorySagaService.rollbackDeductInventorySaga(null, 0);
        // 回滚订单创建
        orderSagaService.rollbackCreateOrderSaga(orderId);
    }
}

6. 实际应用中的考虑因素

6.1 网络通信问题

在分布式系统中，网络通信故障是常见的问题。Saga模式的执行依赖于各个服务之间的可靠通信。为了应对网络故障，可以采用重试机制、消息队列等技术。例如，当某个服务调用失败时，Saga协调器可以重试一定次数；或者将Saga事务的步骤发送到消息队列，由消息队列保证消息的可靠投递和处理。

6.2 数据一致性与最终一致性

Saga模式保证的是最终一致性，而不是强一致性。在一些对数据一致性要求极高的场景下，可能需要结合其他技术，如分布式锁、版本控制等，来确保数据的准确性。例如，在库存扣减时，可以使用分布式锁防止并发扣减导致库存数据不一致。

6.3 性能优化

Saga模式的执行涉及多个服务调用，可能会影响系统性能。可以通过并行执行部分没有依赖关系的本地事务、优化数据库查询等方式来提高性能。同时，合理设置重试次数和重试间隔，避免过度重试导致的性能问题。

6.4 日志记录与监控

在Saga事务执行过程中，详细的日志记录对于故障排查和系统监控非常重要。记录每个本地事务的执行时间、结果以及补偿事务的执行情况等信息，可以帮助开发人员快速定位问题。同时，通过监控工具实时监控Saga事务的执行状态，及时发现异常并进行处理。

7. 总结

Saga模式为电商分布式系统中的分布式事务管理提供了一种灵活、可靠的解决方案。通过将长事务拆分成多个本地事务，并为每个本地事务提供补偿事务，Saga模式能够在复杂的业务场景下保证数据的最终一致性。结合实际代码示例，我们可以看到Saga模式在电商下单、库存扣减、支付等核心业务流程中的具体应用。在实际应用中，需要充分考虑网络通信、数据一致性、性能优化以及日志记录与监控等因素，以确保Saga模式的稳定运行和系统的高效性能。随着电商业务的不断发展和分布式系统的日益复杂，Saga模式有望在更多的场景中得到应用和优化。