Saga模式源码方法教程

本篇内容主要讲解"Saga模式源码方法教程"，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习"Saga模式源码方法教程"吧!

状态机定义

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

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

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

{     "Name": "buyGoodsOnline",     "Comment": "buy a goods on line, add order, deduct account, deduct storage ",     "StartState": "SaveOrder",     "Version": "0.0.1",     "States": {         "SaveOrder": {             "Type": "ServiceTask",             "ServiceName": "orderSave",             "ServiceMethod": "saveOrder",             "CompensateState": "DeleteOrder",             "Next": "ChoiceAccountState",             "Input": [                 "$.[businessKey]",                 "$.[order]"             ],             "Output": {                 "SaveOrderResult": "$.#root"             },             "Status": {                 "#root == true": "SU",                 "#root == false": "FA",                 "$Exception{java.lang.Throwable}": "UN"             }         },         "ChoiceAccountState":{             "Type": "Choice",             "Choices":[                 {                     "Expression":"[SaveOrderResult] == true",                     "Next":"ReduceAccount"                 }             ],             "Default":"Fail"         },         "ReduceAccount": {             "Type": "ServiceTask",             "ServiceName": "accountService",             "ServiceMethod": "decrease",             "CompensateState": "CompensateReduceAccount",             "Next": "ChoiceStorageState",             "Input": [                 "$.[businessKey]",                 "$.[userId]",                 "$.[money]",                 {                     "throwException" : "$.[mockReduceAccountFail]"                 }             ],             "Output": {                 "ReduceAccountResult": "$.#root"             },             "Status": {                 "#root == true": "SU",                 "#root == false": "FA",                 "$Exception{java.lang.Throwable}": "UN"             },             "Catch": [                 {                     "Exceptions": [                         "java.lang.Throwable"                     ],                     "Next": "CompensationTrigger"                 }             ]         },         "ChoiceStorageState":{             "Type": "Choice",             "Choices":[                 {                     "Expression":"[ReduceAccountResult] == true",                     "Next":"ReduceStorage"                 }             ],             "Default":"Fail"         },         "ReduceStorage": {             "Type": "ServiceTask",             "ServiceName": "storageService",             "ServiceMethod": "decrease",             "CompensateState": "CompensateReduceStorage",             "Input": [                 "$.[businessKey]",                 "$.[productId]",                 "$.[count]",                 {                     "throwException" : "$.[mockReduceStorageFail]"                 }             ],             "Output": {                 "ReduceStorageResult": "$.#root"             },             "Status": {                 "#root == true": "SU",                 "#root == false": "FA",                 "$Exception{java.lang.Throwable}": "UN"             },             "Catch": [                 {                     "Exceptions": [                         "java.lang.Throwable"                     ],                     "Next": "CompensationTrigger"                 }             ],             "Next": "Succeed"         },         "DeleteOrder": {             "Type": "ServiceTask",             "ServiceName": "orderSave",             "ServiceMethod": "deleteOrder",             "Input": [                 "$.[businessKey]",                 "$.[order]"             ]         },         "CompensateReduceAccount": {             "Type": "ServiceTask",             "ServiceName": "accountService",             "ServiceMethod": "compensateDecrease",             "Input": [                 "$.[businessKey]",                 "$.[userId]",                 "$.[money]"             ]         },         "CompensateReduceStorage": {             "Type": "ServiceTask",             "ServiceName": "storageService",             "ServiceMethod": "compensateDecrease",             "Input": [                 "$.[businessKey]",                 "$.[productId]",                 "$.[count]"             ]         },         "CompensationTrigger": {             "Type": "CompensationTrigger",             "Next": "Fail"         },         "Succeed": {             "Type":"Succeed"         },         "Fail": {             "Type":"Fail",             "ErrorCode": "PURCHASE_FAILED",             "Message": "purchase failed"         }     } }

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

StateMachineEngine stateMachineEngine = (StateMachineEngine) ApplicationContextUtils.getApplicationContext().getBean("stateMachineEngine");  Map startParams = new HashMap<>(3); String businessKey = String.valueOf(System.currentTimeMillis()); startParams.put("businessKey", businessKey); startParams.put("order", order); startParams.put("mockReduceAccountFail", "true"); startParams.put("userId", order.getUserId()); startParams.put("money", order.getPayAmount()); startParams.put("productId", order.getProductId()); startParams.put("count", order.getCount());  //sync test StateMachineInstance inst = stateMachineEngine.startWithBusinessKey("buyGoodsOnline", null, businessKey, startParams);

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

状态机初始化

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

public class StateMachineConfiguration {      @Bean     public ThreadPoolExecutorFactoryBean threadExecutor(){         ThreadPoolExecutorFactoryBean threadExecutor = new ThreadPoolExecutorFactoryBean();         threadExecutor.setThreadNamePrefix("SAGA_ASYNC_EXE_");         threadExecutor.setCorePoolSize(1);         threadExecutor.setMaxPoolSize(20);         return threadExecutor;     }      @Bean     public DbStateMachineConfig dbStateMachineConfig(ThreadPoolExecutorFactoryBean threadExecutor, DataSource hikariDataSource) throws IOException {         DbStateMachineConfig dbStateMachineConfig = new DbStateMachineConfig();         dbStateMachineConfig.setDataSource(hikariDataSource);         dbStateMachineConfig.setThreadPoolExecutor((ThreadPoolExecutor) threadExecutor.getObject());     /**      *这里配置了json文件的路径，TM在初始化的时候，会把json文件解析成StateMachineImpl类，如果数据库没有保存这个状态机，则存入数据库seata_state_machine_def表，      *如果数据库有记录，则取最新的一条记录，并且注册到StateMachineRepositoryImpl，      *注册的Map有2个，一个是stateMachineMapByNameAndTenant，key格式是(stateMachineName + "_" + tenantId),      *一个是stateMachineMapById，key是stateMachine.getId()      *具体代码见StateMachineRepositoryImpl类registryStateMachine方法      *这个注册的触发方法在DefaultStateMachineConfig的初始化方法init()，这个类是DbStateMachineConfig的父类      */         dbStateMachineConfig.setResources(new PathMatchingResourcePatternResolver().getResources("classpath*:statelang/*.json"));//json文件         dbStateMachineConfig.setEnableAsync(true);         dbStateMachineConfig.setApplicationId("order-server");         dbStateMachineConfig.setTxServiceGroup("my_test_tx_group");         return dbStateMachineConfig;     }      @Bean     public ProcessCtrlStateMachineEngine stateMachineEngine(DbStateMachineConfig dbStateMachineConfig){         ProcessCtrlStateMachineEngine stateMachineEngine = new ProcessCtrlStateMachineEngine();         stateMachineEngine.setStateMachineConfig(dbStateMachineConfig);         return stateMachineEngine;     }      @Bean     public StateMachineEngineHolder stateMachineEngineHolder(ProcessCtrlStateMachineEngine stateMachineEngine){         StateMachineEngineHolder stateMachineEngineHolder = new StateMachineEngineHolder();         stateMachineEngineHolder.setStateMachineEngine(stateMachineEngine);         return stateMachineEngineHolder;     } }

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

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

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

id = null tenantId = null appName = "SEATA" name = "buyGoodsOnline" comment = "buy a goods on line, add order, deduct account, deduct storage " version = "0.0.1" startState = "SaveOrder" status = {StateMachine$Status@9135} "AC" recoverStrategy = null isPersist = true type = "STATE_LANG" content = null gmtCreate = null states = {LinkedHashMap@9137}  size = 11    "SaveOrder" -> {ServiceTaskStateImpl@9153}     "ChoiceAccountState" -> {ChoiceStateImpl@9155}     "ReduceAccount" -> {ServiceTaskStateImpl@9157}     "ChoiceStorageState" -> {ChoiceStateImpl@9159}     "ReduceStorage" -> {ServiceTaskStateImpl@9161}     "DeleteOrder" -> {ServiceTaskStateImpl@9163}     "CompensateReduceAccount" -> {ServiceTaskStateImpl@9165}     "CompensateReduceStorage" -> {ServiceTaskStateImpl@9167}     "CompensationTrigger" -> {CompensationTriggerStateImpl@9169}     "Succeed" -> {SucceedEndStateImpl@9171}     "Fail" -> {FailEndStateImpl@9173}

启动状态机

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

public StateMachineInstance startWithBusinessKey(String stateMachineName, String tenantId, String businessKey,                                                  Map startParams) throws EngineExecutionException {     return startInternal(stateMachineName, tenantId, businessKey, startParams, false, null); } private StateMachineInstance startInternal(String stateMachineName, String tenantId, String businessKey,                                            Map startParams, boolean async, AsyncCallback callback)     throws EngineExecutionException {     //省略部分源代码   //创建一个状态机实例   //默认值tenantId="000001"     StateMachineInstance instance = createMachineInstance(stateMachineName, tenantId, businessKey, startParams);      /**    * ProcessType.STATE_LANG这个枚举只有一个元素    * OPERATION_NAME_START = "start"    * callback是null    * getStateMachineConfig()返回DbStateMachineConfig    */     ProcessContextBuilder contextBuilder = ProcessContextBuilder.create().withProcessType(ProcessType.STATE_LANG)         .withOperationName(DomainConstants.OPERATION_NAME_START).withAsyncCallback(callback).withInstruction(             new StateInstruction(stateMachineName, tenantId)).withStateMachineInstance(instance)         .withStateMachineConfig(getStateMachineConfig()).withStateMachineEngine(this);      Map contextVariables;     if (startParams != null) {         contextVariables = new ConcurrentHashMap<>(startParams.size());         nullSafeCopy(startParams, contextVariables);     } else {         contextVariables = new ConcurrentHashMap<>();     }     instance.setContext(contextVariables);//把启动参数赋值给状态机实例的context     //给ProcessContextImpl的variables加参数     contextBuilder.withStateMachineContextVariables(contextVariables);      contextBuilder.withIsAsyncExecution(async);      //上面定义的建造者创建一个ProcessContextImpl     ProcessContext processContext = contextBuilder.build();      //这个条件是true     if (instance.getStateMachine().isPersist() && stateMachineConfig.getStateLogStore() != null) {       //记录状态机开始状态         stateMachineConfig.getStateLogStore().recordStateMachineStarted(instance, processContext);     }     if (StringUtils.isEmpty(instance.getId())) {         instance.setId(             stateMachineConfig.getSeqGenerator().generate(DomainConstants.SEQ_ENTITY_STATE_MACHINE_INST));     }      if (async) {         stateMachineConfig.getAsyncProcessCtrlEventPublisher().publish(processContext);     } else {       //发送消息到EventBus，这里的消费者是ProcessCtrlEventConsumer，在DefaultStateMachineConfig初始化时设置         stateMachineConfig.getProcessCtrlEventPublisher().publish(processContext);     }      return instance; }

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

开启全局事务

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

stateMachineConfig.getStateLogStore().recordStateMachineStarted(instance, processContext);

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

public void recordStateMachineStarted(StateMachineInstance machineInstance, ProcessContext context) {      if (machineInstance != null) {         //if parentId is not null, machineInstance is a SubStateMachine, do not start a new global transaction,         //use parent transaction instead.         String parentId = machineInstance.getParentId();         if (StringUtils.hasLength(parentId)) {             if (StringUtils.isEmpty(machineInstance.getId())) {                 machineInstance.setId(parentId);             }         } else {         //走这个分支，因为没有配置子状态机         /**              * 这里的beginTransaction就是开启全局事务，        * 这里是调用TC开启全局事务              */             beginTransaction(machineInstance, context);         }           if (StringUtils.isEmpty(machineInstance.getId()) && seqGenerator != null) {             machineInstance.setId(seqGenerator.generate(DomainConstants.SEQ_ENTITY_STATE_MACHINE_INST));         }          // save to db     //dbType = "MySQL"         machineInstance.setSerializedStartParams(paramsSerializer.serialize(machineInstance.getStartParams()));         executeUpdate(stateLogStoreSqls.getRecordStateMachineStartedSql(dbType),                 STATE_MACHINE_INSTANCE_TO_STATEMENT_FOR_INSERT, machineInstance);     } }

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

INSERT INTO seata_state_machine_inst (id, machine_id, tenant_id, parent_id, gmt_started, business_key, start_params, is_running, status, gmt_updated) VALUES ('192.168.59.146:8091:65853497147990016', '06a098cab53241ca7ed09433342e9f07', '000001', null, '2020-10-31 17:18:24.773',  '1604135904773', '{"@type":"java.util.HashMap","money":50.,"productId":1L,"_business_key_":"1604135904773","businessKey":"1604135904773", "count":1,"mockReduceAccountFail":"true","userId":1L,"order":{"@type":"io.seata.sample.entity.Order","count":1,"payAmount":50, "productId":1,"userId":1}}', 1, 'RU', '2020-10-31 17:18:24.773')

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

分支事务处理

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

public class ProcessCtrlEventConsumer implements EventConsumer {      private ProcessController processController;      @Override     public void process(ProcessContext event) throws FrameworkException {         //这里的processController是ProcessControllerImpl         processController.process(event);     }      @Override     public boolean accept(Class clazz) {         return ProcessContext.class.isAssignableFrom(clazz);     }      public void setProcessController(ProcessController processController) {         this.processController = processController;     } }

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

public void process(ProcessContext context) throws FrameworkException {      try {         //这里的businessProcessor是CustomizeBusinessProcessor         businessProcessor.process(context);          businessProcessor.route(context);      } catch (FrameworkException fex) {         throw fex;     } catch (Exception ex) {         LOGGER.error("Unknown exception occurred, context = {}", context, ex);         throw new FrameworkException(ex, "Unknown exception occurred", FrameworkErrorCode.UnknownAppError);     } }

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

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

public void process(ProcessContext context) throws FrameworkException {      /**     *processType = {ProcessType@10310} "STATE_LANG"     *code = "STATE_LANG"     *message = "SEATA State Language"     *name = "STATE_LANG"     *ordinal = 0     */     ProcessType processType = matchProcessType(context);     if (processType == null) {         if (LOGGER.isWarnEnabled()) {             LOGGER.warn("Process type not found, context= {}", context);         }         throw new FrameworkException(FrameworkErrorCode.ProcessTypeNotFound);     }      ProcessHandler processor = processHandlers.get(processType.getCode());     if (processor == null) {         LOGGER.error("Cannot find process handler by type {}, context= {}", processType.getCode(), context);         throw new FrameworkException(FrameworkErrorCode.ProcessHandlerNotFound);     }     //这里的是StateMachineProcessHandler     processor.process(context); }

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

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

public void process(ProcessContext context) throws FrameworkException {     /**    * instruction = {StateInstruction@11057}     * stateName = null    * stateMachineName = "buyGoodsOnline"    * tenantId = "000001"    * end = false    * temporaryState = null     */     StateInstruction instruction = context.getInstruction(StateInstruction.class);   //这里的state实现类是ServiceTaskStateImpl     State state = instruction.getState(context);     String stateType = state.getType();   //这里stateHandler实现类是ServiceTaskStateHandler     StateHandler stateHandler = stateHandlers.get(stateType);      List interceptors = null;     if (stateHandler instanceof InterceptableStateHandler) {       //list上有1个元素ServiceTaskHandlerInterceptor         interceptors = ((InterceptableStateHandler)stateHandler).getInterceptors();     }      List executedInterceptors = null;     Exception exception = null;     try {         if (interceptors != null && interceptors.size() > 0) {             executedInterceptors = new ArrayList<>(interceptors.size());             for (StateHandlerInterceptor interceptor : interceptors) {                 executedInterceptors.add(interceptor);                 interceptor.preProcess(context);             }         }          stateHandler.process(context);      } catch (Exception e) {         exception = e;         throw e;     } finally {          if (executedInterceptors != null && executedInterceptors.size() > 0) {             for (int i = executedInterceptors.size() - 1; i >= 0; i--) {                 StateHandlerInterceptor interceptor = executedInterceptors.get(i);                 interceptor.postProcess(context, exception);             }         }     } }

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

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

public class ServiceTaskHandlerInterceptor implements StateHandlerInterceptor {     //省略部分代码     @Override     public void preProcess(ProcessContext context) throws EngineExecutionException {          StateInstruction instruction = context.getInstruction(StateInstruction.class);          StateMachineInstance stateMachineInstance = (StateMachineInstance)context.getVariable(             DomainConstants.VAR_NAME_STATEMACHINE_INST);         StateMachineConfig stateMachineConfig = (StateMachineConfig)context.getVariable(             DomainConstants.VAR_NAME_STATEMACHINE_CONFIG);          //如果超时,修改状态机状态为FA         if (EngineUtils.isTimeout(stateMachineInstance.getGmtUpdated(), stateMachineConfig.getTransOperationTimeout())) {             String message = "Saga Transaction [stateMachineInstanceId:" + stateMachineInstance.getId()                     + "] has timed out, stop execution now.";             EngineUtils.failStateMachine(context, exception);             throw exception;         }          StateInstanceImpl stateInstance = new StateInstanceImpl();          Map contextVariables = (Map)context.getVariable(             DomainConstants.VAR_NAME_STATEMACHINE_CONTEXT);         ServiceTaskStateImpl state = (ServiceTaskStateImpl)instruction.getState(context);         List