saga分布式事务

本文主要是介绍saga分布式事务，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

一、开篇

在支付和交易业务中，会涉及长事务的场景。依靠单一的数据库事务无法解决整体问题，这个时候，就需要分布式事务来解决这个问题。

二、技术选型

1.saga

Saga是一种在分布式系统中处理事务的模式，它通过将一个大的事务拆分为一系列小的、相互关联的子事务来实现。每个子事务独立执行，并且可以具有回滚和补偿机制，以保证整个事务的一致性。

子事务执行协调方式

编排模式

基于事件，整个业务流程散落到各个业务系统中，比较复杂，流程难以全局理解，而且需要下游实现事件接收和发放。

控制模式

Saga提供一个控制类，其方便参与者之前的协调工作。控制类协调整个流程，下游可以无感知迁移。

2.2pc

成功情况

失败情况

分布式事务2PC（Two-Phase Commit）是一种用于在分布式系统中保持事务一致性的协议。它是一种基于协调者（Coordinator）和参与者（Participant）之间的交互来实现的。

2PC协议的主要目标是在分布式环境下确保所有参与者要么都提交事务，要么都回滚事务，以保持全局事务的一致性。

三、技术实现

基于axon框架实现(控制模式)

OrderFacadeService

订单服务facade层

/**
* 订单服务app层，1.发起创建订单 2.创建订单axon控制器。
*/
@Aggregate(cache = "orderCache")
public class OrderFacadeServiceImpl implements OrdreFacadeService{@Autowiredprivate CommandGateway commandGateWay;@Autowiredprivate PaymentFacadeService paymentClient;@Autowiredprivate LogisticsFacadeService logisticsClient;/*** 创建订单，OrderDomainService会处理对应命令*/@Overridepublic CreateOrderRespDto createOrder(CreateOrderReqDto req){var command = new CreateOrderCommand(req);return commandGateway.sendAndWait(command);}/*** 订单创建成功后创建支付*/@EventSourcingHandlerpublic void on(OrderCreatedEvent event) {var resp = paymentClient.createPay(CreatePaymentReqDto.convertTo(event));if(resp.success){// 发送支付成功事件var event= new PayOrderCreatedEvent(resp.getPaymentOrder());AggregateLifecycle.apply(event);}else{//TODO 发起创建失败事件}}/*** 支付成功后创建物流*/@EventSourcingHandlerpublic void on(PayOrderCreatedEvent event {var resp = logisticsClient.createPay(CreateLogisticsReqDto.convertTo(event));if(resp.success){// 发送创建订单结束var event= new OrderCreateEndEvent(resp.getPaymentOrder());AggregateLifecycle.apply(event);}else{//TODO 发起创建失败事件}}/***  订单创建结束后，OrderDomainService会处理对应命令*/@EventSourcingHandlerpublic void on(OrderCreateEndEvent event) {var command = new OrderCreateEndCommand(event);return commandGateway.sendAndWait(command);}}

OrderDomainService

订单领域层服务

@Service
@Aggregate(cache = "orderCache")
public class OrderDomainServiceImpl implements OrderDomainService{private OrderRepository orderRepository;/*** 创建订单*/@CommandHandler@Overridepublic OrderCreateRespDto handle(OrderCreateCommand command){Order order = Order.convertTo(command);// 存储订单orderRepository.save(order);// 发布订单已创建事件AggregateLifecycle.apply(new OrderCreatedEvent(order));return OrderCreateRespDto.convertTo(order, command);}/*** 创建订单结束*/@CommandHandler@Overridepublic CreateOrderEndRespDto handle(CreateOrderEndCommand command){Order order = Order.convertTo(command);// 更新订单orderRepository.updateCreateEnd(order);return CreateOrderEndRespDto.convertTo(order, command);}
}

todo: 在创建订单完成时候还可以再发起超时事件(延迟消息)，延迟校验创建结果。

基于axon框架实现(编排模式)

OrderFacadeService

订单服务facade层

@Aggregate(cache = "orderCache")
public class OrderFacadeServiceImpl implements OrdreFacadeService{@Autowiredprivate CommandGateway commandGateWay;/*** 创建订单，OrderDomainService会处理对应命令*/@Overridepublic CreateOrderRespDto createOrder(CreateOrderReqDto req){var command = new CreateOrderCommand(req);return commandGateway.sendAndWait(command);}/***  订单创建结束后，OrderDomainService会处理对应命令*/@EventSourcingHandlerpublic void on(OrderCreateEndEvent event) {var command = new OrderCreateEndCommand(event);return commandGateway.sendAndWait(command);}}

OrderDomainService

订单领域层服务

@Service
@Aggregate(cache = "orderCache")
public class OrderDomainServiceImpl implements OrderDomainService{private OrderRepository orderRepository;/*** 开始创建订单*/ @CommandHandler@Overridepublic OrderCreateRespDto handle(OrderCreateCommand command){Order order = Order.convertTo(command);// 存储订单orderRepository.save(order);// 发布订单已创建事件，由支付服务订阅AggregateLifecycle.apply(new OrderCreatedEvent(order));return OrderCreateRespDto.convertTo(order, command);}/** 订单创建结束*/    @CommandHandler@Overridepublic CreateOrderEndRespDto handle(CreateOrderEndCommand command){Order order = Order.convertTo(command);// 更新订单orderRepository.updateCreateEnd(order);return CreateOrderEndRespDto.convertTo(order, command);}
}

PaymentFacadeService

支付服务facade层

@Aggregate(cache = "orderCache")
public class PayFacadeServiceImpl implements PayFacadeService{@Autowiredprivate CommandGateway commandGateWay;/*** 订阅订单创建成功事件*/@EventSourcingHandlerpublic void on(OrderCreatedEvent event) {var command = new CreatePaymentCommand(event);commandGateway.sendAndWait(command);}}

PaymentDomainService

支付服务领域层

@Service
@Aggregate(cache = "orderCache")
public class PaymentDomainServiceImpl implements PaymentDomainService{private PaymentRepository paymentRepository;@CommandHandler@Overridepublic CreatePaymentEndRespDto handle(PaymentCreatCommand command){PaymentOrder order = PaymentOrder.convertTo(command);// 存储订单订单paymentRepository.save(order);// 发布已支付事件AggregateLifecycle.apply(new PaymentCreatedEvent(order));return CreatePaymentEndRespDto .convertTo(order, command);}
}