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

Saga 模式的可扩展性设计与实践

2024-07-016.6k 阅读

Saga 模式基础概念

什么是 Saga 模式

Saga 模式最早由 Hector Garcia - Molina 和 Kenneth Salem 在 1987 年发表的论文 “Sagas” 中提出。它被设计用于解决分布式系统中的长事务问题。在传统的单体应用中，事务通过数据库的 ACID（原子性、一致性、隔离性、持久性）特性来保证数据的一致性。然而，在分布式系统里，涉及多个服务和数据库，要实现像单体应用那样的强一致性事务是非常困难且代价高昂的。

Saga 模式将一个长事务分解为多个本地短事务，每个短事务都由一个服务来管理。这些短事务按照顺序依次执行，如果其中任何一个短事务失败，Saga 会执行一系列的补偿操作，将系统恢复到事务开始前的状态。

Saga 模式的核心组件

  1. Saga 事务：一个 Saga 事务由多个步骤组成，每个步骤都是一个本地事务。例如，在一个电商系统的订单处理 Saga 中，步骤可能包括创建订单、扣除库存、更新用户积分等。
  2. 补偿事务：与每个 Saga 步骤相对应的是一个补偿事务。当某个 Saga 步骤失败时，相应的补偿事务会被执行，以撤销该步骤所做的操作。比如，如果扣除库存步骤失败，补偿事务会将库存恢复到原始状态。
  3. Saga 协调器：负责管理 Saga 事务的执行流程，决定何时执行 Saga 步骤以及在失败时如何触发补偿事务。Saga 协调器可以有不同的实现方式，比如基于编排（Choreography）或基于集中式的协调（Orchestration）。

Saga 模式与传统事务的区别

传统事务依赖数据库的锁机制和日志记录来保证 ACID 特性，适用于单体应用场景。而 Saga 模式则是为分布式系统设计，它不依赖于全局锁，通过本地事务和补偿机制来实现最终一致性。传统事务追求强一致性，而 Saga 模式更注重系统的可用性和性能，在一定时间内达到数据的一致性。

Saga 模式的可扩展性设计原则

水平扩展能力

  1. 分布式事务处理：Saga 模式天然支持分布式事务处理，通过将长事务分解为多个本地事务，每个本地事务可以在不同的服务实例上独立执行。例如，在一个大型电商系统中，订单服务、库存服务和支付服务可以各自处理自己的本地事务，这些服务可以根据业务负载水平扩展实例数量。当订单量增加时，可以增加订单服务的实例；当库存操作频繁时，可以增加库存服务的实例。
  2. 负载均衡：为了实现水平扩展，需要在 Saga 协调器和各个服务之间引入负载均衡机制。例如，可以使用 Nginx 作为反向代理，将请求均匀分配到多个服务实例上。对于 Saga 协调器，如果采用集中式协调方式，可以使用像 ZooKeeper 这样的分布式协调服务来实现协调器的高可用性和水平扩展。ZooKeeper 可以管理协调器的多个实例，当某个实例负载过高时，新的请求可以被分配到其他实例上。

松耦合设计

  1. 服务间解耦：Saga 模式中的各个服务应该保持松耦合。每个服务只负责自己的本地事务和补偿事务，不依赖于其他服务的内部实现细节。例如，订单服务在创建订单时，只需要调用库存服务的扣除库存接口，而不需要知道库存服务是如何管理库存数据的。这种解耦方式使得各个服务可以独立开发、测试和部署，提高了系统的可维护性和扩展性。
  2. 消息驱动：采用消息驱动的方式可以进一步解耦服务。Saga 协调器可以通过消息队列（如 Kafka、RabbitMQ）来发送和接收消息，通知各个服务执行 Saga 步骤或补偿事务。当订单创建成功后，订单服务可以向消息队列发送一条消息，库存服务从消息队列中消费该消息并执行扣除库存的操作。这样，服务之间通过消息进行通信，避免了直接的同步调用，减少了服务之间的依赖。

可插拔的补偿策略

  1. 策略定义：Saga 模式应该支持可插拔的补偿策略。不同的业务场景可能需要不同的补偿策略，比如有的业务可以采用重试策略，当某个本地事务失败时，尝试重新执行一定次数；有的业务则需要采用回滚到某个特定状态的策略。例如，在一个金融转账 Saga 中，如果资金扣除成功但转账失败，补偿策略可以是将扣除的资金退还到原账户。
  2. 实现方式：可以通过定义补偿策略接口，然后为不同的业务场景实现具体的补偿策略类。在 Saga 协调器中，可以根据业务规则选择合适的补偿策略。例如，在 Java 中，可以定义一个 CompensationStrategy 接口，然后实现 RetryCompensationStrategy 和 RollbackCompensationStrategy 等具体策略类。

Saga 模式的实践案例

电商订单处理 Saga

  1. 业务场景：在电商系统中，当用户下单后，需要依次执行创建订单、扣除库存、更新用户积分和处理支付等操作。如果其中任何一个操作失败，需要撤销之前已经执行的操作。
  2. Saga 步骤设计
    • 创建订单：订单服务创建订单记录，并返回订单 ID。
    @Service
public class OrderService {
    @Autowired
    private OrderRepository orderRepository;

    public Order createOrder(OrderRequest orderRequest) {
        Order order = new Order();
        order.setOrderNo(UUID.randomUUID().toString());
        order.setProductList(orderRequest.getProductList());
        order.setStatus(OrderStatus.CREATED);
        return orderRepository.save(order);
    }
}
    • 扣除库存：库存服务根据订单中的商品列表，扣除相应的库存。
    @Service
public class InventoryService {
    @Autowired
    private InventoryRepository inventoryRepository;

    public void deductInventory(List<Product> productList) {
        for (Product product : productList) {
            Inventory inventory = inventoryRepository.findByProductId(product.getId());
            inventory.setQuantity(inventory.getQuantity() - product.getQuantity());
            inventoryRepository.save(inventory);
        }
    }
}
    • 更新用户积分：用户服务根据订单金额更新用户的积分。
    @Service
public class UserService {
    @Autowired
    private UserRepository userRepository;

    public void updateUserPoints(String userId, int points) {
        User user = userRepository.findById(userId).orElseThrow(() -> new UserNotFoundException());
        user.setPoints(user.getPoints() + points);
        userRepository.save(user);
    }
}
    • 处理支付：支付服务处理订单的支付操作。
    @Service
public class PaymentService {
    public PaymentResult processPayment(PaymentRequest paymentRequest) {
        // 模拟支付逻辑
        boolean success = Math.random() > 0.5;
        if (success) {
            return new PaymentResult(PaymentStatus.SUCCESS, "Payment successful");
        } else {
            return new PaymentResult(PaymentStatus.FAILED, "Payment failed");
        }
    }
}
  3. 补偿事务设计
    • 取消订单：订单服务将订单状态更新为取消，并删除相关的订单记录（如果需要）。
    @Service
public class OrderService {
    @Autowired
    private OrderRepository orderRepository;

    public void cancelOrder(String orderId) {
        Order order = orderRepository.findById(orderId).orElseThrow(() -> new OrderNotFoundException());
        order.setStatus(OrderStatus.CANCELED);
        orderRepository.save(order);
    }
}
    • 恢复库存：库存服务根据订单中的商品列表，将库存恢复到原始状态。
    @Service
public class InventoryService {
    @Autowired
    private InventoryRepository inventoryRepository;

    public void restoreInventory(List<Product> productList) {
        for (Product product : productList) {
            Inventory inventory = inventoryRepository.findByProductId(product.getId());
            inventory.setQuantity(inventory.getQuantity() + product.getQuantity());
            inventoryRepository.save(inventory);
        }
    }
}
    • 扣除用户积分：用户服务根据订单金额，将用户积分恢复到之前的状态。
    @Service
public class UserService {
    @Autowired
    private UserRepository userRepository;

    public void deductUserPoints(String userId, int points) {
        User user = userRepository.findById(userId).orElseThrow(() -> new UserNotFoundException());
        user.setPoints(user.getPoints() - points);
        userRepository.save(user);
    }
}
    • 撤销支付：支付服务如果支付成功，发起退款操作（这里简化为模拟）。
    @Service
public class PaymentService {
    public void reversePayment(String paymentId) {
        // 模拟撤销支付逻辑
        System.out.println("Payment reversed for paymentId: " + paymentId);
    }
}
  4. Saga 协调器实现：这里采用基于编排的方式，各个服务之间通过消息进行通信。
    • 消息定义：定义创建订单、扣除库存等消息。
    public class CreateOrderMessage {
    private OrderRequest orderRequest;

    public CreateOrderMessage(OrderRequest orderRequest) {
        this.orderRequest = orderRequest;
    }

    public OrderRequest getOrderRequest() {
        return orderRequest;
    }
}

public class DeductInventoryMessage {
    private List<Product> productList;

    public DeductInventoryMessage(List<Product> productList) {
        this.productList = productList;
    }

    public List<Product> getProductList() {
        return productList;
    }
}
    • 消息队列配置：以 RabbitMQ 为例，配置消息队列和交换机。
    @Configuration
public class RabbitMQConfig {
    @Bean
    public Queue createOrderQueue() {
        return QueueBuilder.durable("create - order - queue").build();
    }

    @Bean
    public Queue deductInventoryQueue() {
        return QueueBuilder.durable("deduct - inventory - queue").build();
    }

    @Bean
    public DirectExchange orderExchange() {
        return new DirectExchange("order - exchange");
    }

    @Bean
    public Binding createOrderBinding(Queue createOrderQueue, DirectExchange orderExchange) {
        return BindingBuilder.bind(createOrderQueue).to(orderExchange).with("create.order");
    }

    @Bean
    public Binding deductInventoryBinding(Queue deductInventoryQueue, DirectExchange orderExchange) {
        return BindingBuilder.bind(deductInventoryQueue).to(orderExchange).with("deduct.inventory");
    }
}
    • 消息消费者实现：订单服务和库存服务等实现消息消费者。
    @Component
public class OrderMessageConsumer {
    @Autowired
    private OrderService orderService;

    @RabbitListener(queues = "create - order - queue")
    public void handleCreateOrderMessage(CreateOrderMessage message) {
        Order order = orderService.createOrder(message.getOrderRequest());
        // 发送扣除库存消息
        List<Product> productList = message.getOrderRequest().getProductList();
        rabbitTemplate.convertAndSend("order - exchange", "deduct.inventory", new DeductInventoryMessage(productList));
    }
}

@Component
public class InventoryMessageConsumer {
    @Autowired
    private InventoryService inventoryService;

    @RabbitListener(queues = "deduct - inventory - queue")
    public void handleDeductInventoryMessage(DeductInventoryMessage message) {
        inventoryService.deductInventory(message.getProductList());
        // 后续继续发送更新用户积分等消息
    }
}

旅游预订 Saga

  1. 业务场景：用户预订旅游产品，包括预订酒店、预订机票和租车等操作。如果其中任何一个操作失败，需要取消之前已经预订的项目。
  2. Saga 步骤设计
    • 预订酒店：酒店服务根据用户的预订请求，在系统中预订酒店房间，并返回预订确认信息。
    class HotelService:
    def book_hotel(self, booking_request):
        # 模拟酒店预订逻辑
        if booking_request.room_type in ['single', 'double']:
            booking_id = 'hotel_' + str(uuid.uuid4())
            return {'booking_id': booking_id,'status': 'booked'}
        else:
            raise ValueError('Unsupported room type')
    • 预订机票：机票服务根据用户的行程信息，预订机票，并返回机票预订确认信息。
    class FlightService:
    def book_flight(self, flight_request):
        # 模拟机票预订逻辑
        if flight_request.destination in ['New York', 'London']:
            ticket_id = 'flight_' + str(uuid.uuid4())
            return {'ticket_id': ticket_id,'status': 'booked'}
        else:
            raise ValueError('Unsupported destination')
    • 租车：租车服务根据用户的租车需求，提供车辆预订服务，并返回租车确认信息。
    class CarRentalService:
    def rent_car(self, car_request):
        # 模拟租车逻辑
        if car_request.car_type in ['sedan', 'SUV']:
            rental_id = 'car_' + str(uuid.uuid4())
            return {'rental_id': rental_id,'status': 'booked'}
        else:
            raise ValueError('Unsupported car type')
  3. 补偿事务设计
    • 取消酒店预订：酒店服务根据预订 ID，取消酒店预订。
    class HotelService:
    def cancel_hotel_booking(self, booking_id):
        # 模拟取消酒店预订逻辑
        print(f'Canceling hotel booking with ID {booking_id}')
    • 取消机票预订：机票服务根据机票预订 ID，取消机票预订。
    class FlightService:
    def cancel_flight_booking(self, ticket_id):
        # 模拟取消机票预订逻辑
        print(f'Canceling flight booking with ID {ticket_id}')
    • 取消租车：租车服务根据租车 ID，取消租车预订。
    class CarRentalService:
    def cancel_car_rental(self, rental_id):
        # 模拟取消租车逻辑
        print(f'Canceling car rental with ID {rental_id}')
  4. Saga 协调器实现：这里采用基于集中式协调的方式，使用 Python 的 asyncio 库来管理 Saga 事务流程。 
    import asyncio


class TravelSagaCoordinator:
    def __init__(self):
        self.hotel_service = HotelService()
        self.flight_service = FlightService()
        self.car_rental_service = CarRentalService()

    async def execute_saga(self, travel_request):
        try:
            hotel_result = self.hotel_service.book_hotel(travel_request.hotel_request)
            flight_result = self.flight_service.book_flight(travel_request.flight_request)
            car_result = self.car_rental_service.rent_car(travel_request.car_request)
            return {'status':'success'}
        except Exception as e:
            await self.compensate(hotel_result.get('booking_id') if 'booking_id' in hotel_result else None,
                                  flight_result.get('ticket_id') if 'ticket_id' in flight_result else None,
                                  car_result.get('rental_id') if'rental_id' in car_result else None)
            return {'status': 'failed','reason': str(e)}

    async def compensate(self, hotel_booking_id, flight_ticket_id, car_rental_id):
        tasks = []
        if hotel_booking_id:
            tasks.append(asyncio.create_task(self.hotel_service.cancel_hotel_booking(hotel_booking_id)))
        if flight_ticket_id:
            tasks.append(asyncio.create_task(self.flight_service.cancel_flight_booking(flight_ticket_id)))
        if car_rental_id:
            tasks.append(asyncio.create_task(self.car_rental_service.cancel_car_rental(car_rental_id)))
        await asyncio.gather(*tasks)


class TravelRequest:
    def __init__(self, hotel_request, flight_request, car_request):
        self.hotel_request = hotel_request
        self.flight_request = flight_request
        self.car_request = car_request


if __name__ == '__main__':
    hotel_request = {'room_type':'single'}
    flight_request = {'destination': 'New York'}
    car_request = {'car_type':'sedan'}
    travel_request = TravelRequest(hotel_request, flight_request, car_request)
    coordinator = TravelSagaCoordinator()
    result = asyncio.run(coordinator.execute_saga(travel_request))
    print(result)

Saga 模式实践中的挑战与应对

数据一致性问题

  1. 问题描述：虽然 Saga 模式通过补偿机制来保证最终一致性，但在实际运行过程中，可能会出现由于网络延迟、服务故障等原因导致补偿事务执行不完全或失败的情况，从而影响数据的一致性。例如，在电商订单处理 Saga 中，当支付失败触发补偿事务时，恢复库存操作可能因为网络问题没有成功执行，导致库存数据不一致。
  2. 应对策略
    • 重试机制：对于补偿事务执行失败的情况，可以引入重试机制。在一定时间间隔内多次尝试执行补偿事务，直到成功为止。例如，在 Java 中，可以使用 Spring Retry 框架来实现重试逻辑。
    @Service
@Retryable(value = {InventoryException.class}, maxAttempts = 3, backoff = @Backoff(delay = 1000))
public void restoreInventory(List<Product> productList) {
    for (Product product : productList) {
        Inventory inventory = inventoryRepository.findByProductId(product.getId());
        inventory.setQuantity(inventory.getQuantity() + product.getQuantity());
        inventoryRepository.save(inventory);
    }
}
    • 日志记录与监控：详细记录 Saga 事务和补偿事务的执行日志，通过监控系统实时跟踪事务的执行状态。一旦发现有事务执行异常，及时通知运维人员进行干预。例如，可以使用 ELK（Elasticsearch、Logstash、Kibana） 堆栈来收集和分析日志数据。

性能开销

  1. 问题描述：Saga 模式由于涉及多个本地事务和可能的补偿事务，以及消息通信等操作，会带来一定的性能开销。特别是在高并发场景下，消息队列的处理能力、服务间的网络通信延迟等因素都可能影响系统的整体性能。
  2. 应对策略
    • 优化消息队列：选择高性能的消息队列，并合理配置其参数。例如，对于 Kafka，可以调整分区数量、副本因子等参数来提高消息的处理能力。同时，采用批量处理消息的方式，减少消息发送和接收的次数。
    • 服务性能优化：对各个服务进行性能优化，如优化数据库查询语句、使用缓存等。在电商订单处理 Saga 中，库存服务可以使用 Redis 缓存库存数据，减少对数据库的直接查询次数，提高响应速度。

版本兼容性

  1. 问题描述：随着业务的发展，Saga 模式中的各个服务可能会进行版本升级。不同版本的服务之间可能存在接口不兼容的问题，导致 Saga 事务无法正常执行。例如，订单服务升级后，其创建订单接口的参数格式发生了变化，而库存服务仍然按照旧的接口格式发送消息，就会导致订单创建失败。
  2. 应对策略
    • 版本控制与兼容性测试：建立严格的版本控制机制，在服务升级前进行充分的兼容性测试。可以使用契约测试工具（如 Pact）来确保不同服务之间的接口兼容性。Pact 可以生成消费者和提供者之间的契约文件，在服务升级时验证新的服务是否符合契约。
    • 接口演进策略：采用兼容的接口演进策略，如在升级接口时保留旧接口一段时间，或者通过添加新的参数来实现功能扩展，同时保证旧的参数仍然可用。这样可以避免对依赖该接口的其他服务造成影响。