阿里Seata真香，肝一下Saga模式源码-云智核

当前位置：首页 >IT科技 >阿里Seata真香，肝一下Saga模式源码

阿里Seata真香，肝一下Saga模式源码

发布时间：2025-11-05 12:08:57 来源：云智核作者：IT科技

本文转载自微信公众号「程序员jinjunzhu」，阿里作者jinjunzhu。香肝下转载本文请联系程序员jinjunzhu公众号。模码

saga模式是式源分布式事务中使用比较多的一种模式，主要应用在多节点长流程的阿里应用中，对一个全局事务，香肝下如果某个节点抛出了异常，模码则从当前这个节点依次往前补偿事务。式源一阶段正向服务和二阶段补偿服务都需要由业务代码来实现。阿里今天我们就来看看它的香肝下源码实现。

状态机定义

以一个典型的模码电商购物流程为例，我们定义3个服务，式源订单服务(OrderServer)，阿里账户服务(AccountService)和库存服务(StorageService),香肝下这里我们把订单服务当做聚合服务，也就是模码TM。

当外部下单时，订单服务首先会创建一个订单，然后调用账户服务扣减金额，最后调用库存服务扣减库存。这个流程入下图：

seata的saga模式是基于状态机来实现了，状态机对状态的控制需要一个JSON文件，这个JSON文件定义如下：

状态机是运行在TM中的，也就是b2b信息网我们上面定义的订单服务。订单服务创建订单时需要开启一个全局事务，这时就需要启动状态机，代码如下：

可以看到，上面代码定义的buyGoodsOnline，正是JSON文件中name的属性值。

状态机初始化

那上面创建订单代码中的stateMachineEngine这个bean是在哪里定义的呢?订单服务的demo中有一个类StateMachineConfiguration来进行定义，代码如下：

b2b供应网

可以看到，我们在DbStateMachineConfig中配置了状态机的json文件，同时配置了applicationId和txServiceGroup。在DbStateMachineConfig初始化的时候，子类DefaultStateMachineConfig的init的方法会把json文件解析成状态机，并注册。

注册的过程中往seata_state_machine_def这张表里插入了1条记录，表里的content字段保存了我们的JOSON文件内容，其他字段值数据如下图：

附:根据前面的JSON文件，我们debug跟踪到的StateMachineImpl的内容如下：

启动状态机

在第一节创建订单的代码中，startWithBusinessKey方法进行了整个事务的启动，这个方法还有一个异步模式startWithBusinessKeyAsync，这里我们只分析同步模式，源代码如下：

网站模板

上面的代码中我们可以看出，启动状态记得时候主要做了2件事情，一个是记录状态机开始的状态，一个是发送消息到EventBus，下面我们详细看一下这2个过程。

开启全局事务

上面的代码分析中，有一个记录状态机开始状态的代码，如下：

这里调用了类DbAndReportTcStateLogStore的recordStateMachineStarted方法，我们来看一下，代码如下：

上面executeUpdate方法在子类AbstractStore,debug一下executeUpdate这个方法可以看到，这里执行的sql如下：

可以看到，这个全局事务记录在了表seata_state_machine_inst，记录的是我们启动状态机的参数，status记录的状态是"RU"也就是RUNNING。

分支事务处理

上一节我们提到，启动状态机后，向EventBus发了一条消息，这个消息的消费者是ProcessCtrlEventConsumer，我们看一下这个类的代码：

ProcessControllerImpl类的process方法有2个处理逻辑，process和route，代码如下：

这里的处理逻辑有些复杂，先上一张UML类图，跟着这张图，可以捋清楚代码的调用逻辑：

我们先来看一下CustomizeBusinessProcessor中的process方法：

这里的代码不好理解，我们分四步来研究。

第一步，我们看一下StateMachineProcessHandler类中process方法，这个方法代理了ServiceTaskStateHandler的process方法，代码如下：

从这个方法我们看到，代理对stateHandler.process加入了前置和后置增强，增强类是ServiceTaskHandlerInterceptor，前置后置增强分别调用了interceptor的preProcess和postProcess。

第二步，我们来看一下增强逻辑。ServiceTaskHandlerInterceptor的preProcess和postProcess方法，代码如下：

从这个代码我们能看到，分支事务执行前，封装了一个StateInstanceImpl赋值给了ProcessContext，分支事务执行后，对这个StateInstanceImpl进行了修改，这个StateInstanceImpl有3个作用：

传入StateMachineInstanceImpl的stateMap用于重试或交易补偿

记录了分支事务的执行情况，同时支持持久化到seata_state_inst表

传入TaskStateRouter用作判断全局事务结束

第三步，我们看一下被代理的方法stateHandler.process(context)，正常执行逻辑中stateHandler的实现类是ServiceTaskStateHandler，代码如下：

可以看到，process这个方法是一个核心的业务处理，它用发射触发了JSON中定义ServiceTask的方法，并且根据状态触发了Next对象，即流程中的下一个ServiceTask。

第四步，我们再看一下CustomizeBusinessProcessor的route方法，代码如下：

我们看一下DefaultRouterHandler的route方法，代码如下：

看一下StateMachineProcessRouter的route方法，这里也是用了代理模式，代码如下：

这里的代理只实现了一个后置增强，做的事情就是结束状态机。

下面我们来看一下StateRouter，UML类图如下：

从UML类图我们看到，除了EndStateRouter，只有一个TaskStateRouter了。而EndStateRouter并没有做什么事情，因为关闭状态机的逻辑已经由代理做了。这里我们看一下TaskStateRouter，代码如下：

可以看到，route的作用是帮状态机确定下一个流程节点，然后放入到当前的context中的stateInstruction。

到这里，我们就分析完成了状态机的原理，ProcessControllerImpl类中。

需要注意的是，这里获取到下一个节点后，并没有直接处理，而是使用观察者模式，先发送到EventBus，等待观察者来处理，循环往复，直到EndStateRouter结束状态机。

这里观察者模式的Event是ProcessContext，里面包含了Instruction，而Instruction里面包含了State，这个State里面就决定了下一个处理的节点直到结束。UML类图如下：

总结

seata中间件中的saga模式使用比较广泛，但是代码还是比较复杂的。我从下面几个方面进行了梳理：

saga模式额外引入了3张表，我们也可以根据跟全局事务和分支事务相关的2张表来跟踪代码，我之前给出的demo，如果事务成功，这2张表的写sql按照状态机执行顺序给出一个成功sql，代码如下：

这篇文章我主要从一个正常的流程研究了saga模式的源代码，还有好多细节没有做分析，比如全局事务失败后的回滚或补偿逻辑，以后有机会再交流。

随便看看