Seata源码—8.Seata Saga模式的事务处理

大纲
1.Seata Saga案例简介
2.Seata Saga案例的状态机定义分析
3.Seata Saga分布式事务与状态机关系
4.Seata Saga案例的Dubbo服务调用配置分析
5.Seata Saga案例的状态机数据库和工程启动
6.基于数据库的状态机配置实例的初始化
7.状态机配置实例中包含的一些关键组件
8.默认的状态机配置类的初始化
9.状态机定义的仓储组件解析状态机定义文件
10.状态机定义的仓储组件注册StateMachine
11.状态机引擎接口的源码
12.状态机引擎创建状态机实例的源码
13.ProcessContextBuilder构建流程上下文
14.StateLogStore记录日志和开启Saga全局事务
15.状态操作组件获取状态和State状态类继承体系
16.启动状态机实例时发布流程上下文到事件总线
17.通过业务处理器处理状态机当前需要执行的状态
18.通过服务调用组件来执行State指定的服务方法
19.将业务处理器路由到状态机下一个要执行的状态
 
1.Seata Saga案例简介
seata-samples项目下的saga模块，会基于Seata Saga模式，实现了分布式事务的提交和回滚。
 
在dubbo-saga-sample模块中，一个分布式事务内会有两个Saga事务参与者，这两个Saga事务参与者分别是InventoryAction和BalanceAction。当分布式事务提交时，两者均提交。当分布式事务回滚时，两者均回滚。
 
这两个Saga事务参与者均是Dubbo服务。这两个Saga事务参与者都有一个reduce方法表示库存扣减或余额扣减，还有一个compensateReduce方法表示补偿扣减操作。
 
InventoryAction接口定义如下：

1. public interface InventoryAction {
2.     boolean reduce(String businessKey, int count);
3.     boolean compensateReduce(String businessKey);
4. }

复制代码

BalanceAction接口定义如下：

1. public interface BalanceAction {
2.     boolean reduce(String businessKey, BigDecimal amount, Map<String, Object> params);
3.     boolean compensateReduce(String businessKey, Map<String, Object> params);
4. }

复制代码

运行dubbo-saga-sample模块的Demo工程：
步骤一：启动Seata Server
步骤二：运行DubboSagaProviderStarter
步骤三：运行DubboSagaTransactionStarter
 
2.Seata Saga案例的状态机定义分析
(1)Saga状态机定义文件详情
(2)状态机定义的全流程输入输出和流转分析
 
(1)Saga状态机定义文件详情
Saga的本质就是一个状态机。
 
reduce_inventory_and_balance.json

1. {
2.     "Name": "reduceInventoryAndBalance",
3.     "Comment": "reduce inventory then reduce balance in a transaction",
4.     "StartState": "ReduceInventory",
5.     "Version": "0.0.1",
6.     "States": {
7.         "ReduceInventory": {
8.             "Type": "ServiceTask",
9.             "ServiceName": "inventoryAction",
10.             "ServiceMethod": "reduce",
11.             "CompensateState": "CompensateReduceInventory",
12.             "Next": "ChoiceState",
13.             "Input": [
14.                 "$.[businessKey]",
15.                 "$.[count]"
16.             ],
17.             "Output": {
18.                 "reduceInventoryResult": "$.#root"
19.             },
20.             "Status": {
21.                 "#root == true": "SU",
22.                 "#root == false": "FA",
23.                 "$Exception{java.lang.Throwable}": "UN"
24.             }
25.         },
26.         "ChoiceState": {
27.             "Type": "Choice",
28.             "Choices": [{
29.                 "Expression": "[reduceInventoryResult] == true",
30.                 "Next": "ReduceBalance"
31.             }],
32.             "Default": "Fail"
33.         },
34.         "ReduceBalance": {
35.             "Type": "ServiceTask",
36.             "ServiceName": "balanceAction",
37.             "ServiceMethod": "reduce",
38.             "CompensateState": "CompensateReduceBalance",
39.             "Input": [
40.                 "$.[businessKey]",
41.                 "$.[amount]",
42.                 {
43.                     "throwException": "$.[mockReduceBalanceFail]"
44.                 }
45.             ],
46.             "Output": {
47.                 "compensateReduceBalanceResult": "$.#root"
48.             },
49.             "Status": {
50.                 "#root == true": "SU",
51.                 "#root == false": "FA",
52.                 "$Exception{java.lang.Throwable}": "UN"
53.             },
54.             "Catch": [{
55.                 "Exceptions": [
56.                     "java.lang.Throwable"
57.                 ],
58.                 "Next": "CompensationTrigger"
59.             }],
60.             "Next": "Succeed"
61.         },
62.         "CompensateReduceInventory": {
63.             "Type": "ServiceTask",
64.             "ServiceName": "inventoryAction",
65.             "ServiceMethod": "compensateReduce",
66.             "Input": [
67.                 "$.[businessKey]"
68.             ]
69.         },
70.         "CompensateReduceBalance": {
71.             "Type": "ServiceTask",
72.             "ServiceName": "balanceAction",
73.             "ServiceMethod": "compensateReduce",
74.             "Input": [
75.                 "$.[businessKey]"
76.             ]
77.         },
78.         "CompensationTrigger": {
79.             "Type": "CompensationTrigger",
80.             "Next": "Fail"
81.         },
82.         "Succeed": {
83.             "Type": "Succeed"
84.         },
85.         "Fail": {
86.             "Type": "Fail",
87.             "ErrorCode": "PURCHASE_FAILED",
88.             "Message": "purchase failed"
89.         }
90.     }
91. }

复制代码

(2)状态机定义的全流程输入输出和流转分析

 
3.Seata Saga分布式事务与状态机关系
Seata Saga分布式事务基本都是通过状态机一起配合起来使用的，通过状态机来调度编排每一个分布式事务的执行。编排的每一个服务执行时，可能会执行本地事务，也可能会调用远程服务。触发每一个服务的补偿时，可能会回滚本地事务，也可能会补偿远程服务。

 
4.Seata Saga案例的Dubbo服务调用配置分析
(1)Dubbo服务提供者的配置
(2)Dubbo服务调用者的配置
(3)Seata Saga状态机的Bean的配置
 
(1)Dubbo服务提供者的配置
seata-dubbo-provider.xml

1. <beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
2.        xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"
3.        xmlns="http://www.springframework.org/schema/beans"
4.        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
5.        http://code.alibabatech.com/schema/dubbo
6.        http://code.alibabatech.com/schema/dubbo/dubbo.xsd" default-autowire="byName">
7.     <dubbo:application name="saga-sample">
8.         <dubbo:parameter key="qos.enable" value="true"/>
9.         <dubbo:parameter key="qos.accept.foreign.ip" value="false"/>
10.         <dubbo:parameter key="qos.port" value="33333"/>
11.     </dubbo:application>
13.     <dubbo:registry address="zookeeper://127.0.0.1:2181"/>
14.     <dubbo:protocol name="dubbo" port="-1"/>
15.     <dubbo:provider timeout="10000" threads="10" threadpool="fixed" loadbalance="roundrobin"/>
16.    
17.     <dubbo:service ref="inventoryActionImpl" interface="io.seata.samples.saga.action.InventoryAction"/>
18.     <dubbo:service ref="balanceActionImpl" interface="io.seata.samples.saga.action.BalanceAction"/>
20.     <bean id="inventoryActionImpl" />
21.     <bean id="balanceActionImpl" />
22. </beans>

复制代码

(2)Dubbo服务调用者的配置
seata-dubbo-reference.xml

1. <beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
2.        xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"
3.        xmlns="http://www.springframework.org/schema/beans"
4.        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
5.        http://code.alibabatech.com/schema/dubbo
6.        http://code.alibabatech.com/schema/dubbo/dubbo.xsd" default-autowire="byName">
7.     <dubbo:application name="saga-sample-reference">
8.         <dubbo:parameter key="qos.enable" value="true"/>
9.         <dubbo:parameter key="qos.accept.foreign.ip" value="false"/>
10.         <dubbo:parameter key="qos.port" value="22222"/>
11.     </dubbo:application>
13.     <dubbo:registry address="zookeeper://127.0.0.1:2181"/>
14.     <dubbo:protocol name="dubbo" port="-1"/>
16.     <dubbo:reference id="inventoryAction" interface="io.seata.samples.saga.action.InventoryAction" check="false" lazy="true"/>
17.     <dubbo:reference id="balanceAction" interface="io.seata.samples.saga.action.BalanceAction" check="false" lazy="true"/>
18. </beans>

复制代码

(3)Seata Saga状态机的Bean的配置
seata-saga.xml

1. <beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
2.        xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"
3.        xmlns="http://www.springframework.org/schema/beans"
4.        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
5.        http://code.alibabatech.com/schema/dubbo
6.        http://code.alibabatech.com/schema/dubbo/dubbo.xsd" default-autowire="byName">
7.     <dubbo:application name="saga-sample">
8.         <dubbo:parameter key="qos.enable" value="true"/>
9.         <dubbo:parameter key="qos.accept.foreign.ip" value="false"/>
10.         <dubbo:parameter key="qos.port" value="33333"/>
11.     </dubbo:application>
13.     <dubbo:registry address="zookeeper://127.0.0.1:2181"/>
14.     <dubbo:protocol name="dubbo" port="-1"/>
15.     <dubbo:provider timeout="10000" threads="10" threadpool="fixed" loadbalance="roundrobin"/>
16.    
17.     <dubbo:service ref="inventoryActionImpl" interface="io.seata.samples.saga.action.InventoryAction"/>
18.     <dubbo:service ref="balanceActionImpl" interface="io.seata.samples.saga.action.BalanceAction"/>
20.     <bean id="inventoryActionImpl" />
21.     <bean id="balanceActionImpl" />
22. </beans><beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
23.        xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"
24.        xmlns="http://www.springframework.org/schema/beans"
25.        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
26.        http://code.alibabatech.com/schema/dubbo
27.        http://code.alibabatech.com/schema/dubbo/dubbo.xsd" default-autowire="byName">
28.     <dubbo:application name="saga-sample">
29.         <dubbo:parameter key="qos.enable" value="true"/>
30.         <dubbo:parameter key="qos.accept.foreign.ip" value="false"/>
31.         <dubbo:parameter key="qos.port" value="33333"/>
32.     </dubbo:application>
34.     <dubbo:registry address="zookeeper://127.0.0.1:2181"/>
35.     <dubbo:protocol name="dubbo" port="-1"/>
36.     <dubbo:provider timeout="10000" threads="10" threadpool="fixed" loadbalance="roundrobin"/>
37.    
38.     <dubbo:service ref="inventoryActionImpl" interface="io.seata.samples.saga.action.InventoryAction"/>
39.     <dubbo:service ref="balanceActionImpl" interface="io.seata.samples.saga.action.BalanceAction"/>
41.     <bean id="inventoryActionImpl" />
42.     <bean id="balanceActionImpl" />
43. </beans><beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
44.        xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"
45.        xmlns="http://www.springframework.org/schema/beans"
46.        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
47.        http://code.alibabatech.com/schema/dubbo
48.        http://code.alibabatech.com/schema/dubbo/dubbo.xsd" default-autowire="byName">
49.     <dubbo:application name="saga-sample">
50.         <dubbo:parameter key="qos.enable" value="true"/>
51.         <dubbo:parameter key="qos.accept.foreign.ip" value="false"/>
52.         <dubbo:parameter key="qos.port" value="33333"/>
53.     </dubbo:application>
55.     <dubbo:registry address="zookeeper://127.0.0.1:2181"/>
56.     <dubbo:protocol name="dubbo" port="-1"/>
57.     <dubbo:provider timeout="10000" threads="10" threadpool="fixed" loadbalance="roundrobin"/>
58.    
59.     <dubbo:service ref="inventoryActionImpl" interface="io.seata.samples.saga.action.InventoryAction"/>
60.     <dubbo:service ref="balanceActionImpl" interface="io.seata.samples.saga.action.BalanceAction"/>
62.     <bean id="inventoryActionImpl" />
63.     <bean id="balanceActionImpl" />
64. </beans><beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
65.        xmlns:dubbo="http://code.alibabatech.com/schema/dubbo"
66.        xmlns="http://www.springframework.org/schema/beans"
67.        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
68.        http://code.alibabatech.com/schema/dubbo
69.        http://code.alibabatech.com/schema/dubbo/dubbo.xsd" default-autowire="byName">
70.     <dubbo:application name="saga-sample-reference">
71.         <dubbo:parameter key="qos.enable" value="true"/>
72.         <dubbo:parameter key="qos.accept.foreign.ip" value="false"/>
73.         <dubbo:parameter key="qos.port" value="22222"/>
74.     </dubbo:application>
76.     <dubbo:registry address="zookeeper://127.0.0.1:2181"/>
77.     <dubbo:protocol name="dubbo" port="-1"/>
79.     <dubbo:reference id="inventoryAction" interface="io.seata.samples.saga.action.InventoryAction" check="false" lazy="true"/>
80.     <dubbo:reference id="balanceAction" interface="io.seata.samples.saga.action.BalanceAction" check="false" lazy="true"/>
81. </beans>        

复制代码

 
5.Seata Saga案例的状态机数据库和工程启动
(1)状态机相关的数据库表
(2)Seata Saga案例的工程启动
 
(1)状态机相关的数据库表
状态机定义 -> 状态机实例 -> 状态实例

1. -- PUBLIC.SEATA_STATE_INST definition
3. -- 状态实例，每个状态机实例里会有多个状态实例
4. CREATE CACHED TABLE "PUBLIC"."SEATA_STATE_INST"(
5.     "ID" VARCHAR NOT NULL COMMENT 'id',
6.     "MACHINE_INST_ID" VARCHAR NOT NULL COMMENT 'state machine instance id，机器实例ID',
7.     "NAME" VARCHAR NOT NULL COMMENT 'state name',
8.     "TYPE" VARCHAR COMMENT 'state type',
9.     "SERVICE_NAME" VARCHAR COMMENT 'service name',
10.     "SERVICE_METHOD" VARCHAR COMMENT 'method name',
11.     "SERVICE_TYPE" VARCHAR COMMENT 'service type',
12.     "BUSINESS_KEY" VARCHAR COMMENT 'business key',
13.     "STATE_ID_COMPENSATED_FOR" VARCHAR COMMENT 'state compensated for',
14.     "STATE_ID_RETRIED_FOR" VARCHAR COMMENT 'state retried for',
15.     "GMT_STARTED" TIMESTAMP NOT NULL COMMENT 'start time',
16.     "IS_FOR_UPDATE" TINYINT COMMENT 'is service for update',
17.     "INPUT_PARAMS" CLOB COMMENT 'input parameters',
18.     "OUTPUT_PARAMS" CLOB COMMENT 'output parameters',
19.     "STATUS" VARCHAR NOT NULL COMMENT 'status(SU succeed|FA failed|UN unknown|SK skipped|RU running)',
20.     "EXCEP" BLOB COMMENT 'exception',
21.     "GMT_UPDATED" TIMESTAMP COMMENT 'update time',
22.     "GMT_END" TIMESTAMP COMMENT 'end time'
23. );
25. -- PUBLIC.SEATA_STATE_MACHINE_DEF definition
26. -- 状态机定义，在json文件中定义好的状态流程会存放到这里
27. CREATE CACHED TABLE "PUBLIC"."SEATA_STATE_MACHINE_DEF"(
28.     "ID" VARCHAR NOT NULL COMMENT 'id',
29.     "NAME" VARCHAR NOT NULL COMMENT 'name',
30.     "TENANT_ID" VARCHAR NOT NULL COMMENT 'tenant id',
31.     "APP_NAME" VARCHAR NOT NULL COMMENT 'application name',
32.     "TYPE" VARCHAR COMMENT 'state language type',
33.     "COMMENT_" VARCHAR COMMENT 'comment',
34.     "VER" VARCHAR NOT NULL COMMENT 'version',
35.     "GMT_CREATE" TIMESTAMP NOT NULL COMMENT 'create time',
36.     "STATUS" VARCHAR NOT NULL COMMENT 'status(AC:active|IN:inactive)',
37.     "CONTENT" CLOB COMMENT 'content',
38.     "RECOVER_STRATEGY" VARCHAR COMMENT 'transaction recover strategy(compensate|retry)'
39. );
41. -- PUBLIC.SEATA_STATE_MACHINE_INST definition
42. -- 状态机实例，每执行一次完整的分布式事务就会对应一个状态机实例，每个状态机实例里会有多个状态实例
43. CREATE CACHED TABLE "PUBLIC"."SEATA_STATE_MACHINE_INST"(
44.     "ID" VARCHAR NOT NULL COMMENT 'id',
45.     "MACHINE_ID" VARCHAR NOT NULL COMMENT 'state machine definition id',
46.     "TENANT_ID" VARCHAR NOT NULL COMMENT 'tenant id',
47.     "PARENT_ID" VARCHAR COMMENT 'parent id',
48.     "GMT_STARTED" TIMESTAMP NOT NULL COMMENT 'start time',
49.     "BUSINESS_KEY" VARCHAR COMMENT 'business key',
50.     "START_PARAMS" CLOB COMMENT 'start parameters',
51.     "GMT_END" TIMESTAMP COMMENT 'end time',
52.     "EXCEP" BLOB COMMENT 'exception',
53.     "END_PARAMS" CLOB COMMENT 'end parameters',
54.     "STATUS" VARCHAR COMMENT 'status(SU succeed|FA failed|UN unknown|SK skipped|RU running)',
55.     "COMPENSATION_STATUS" VARCHAR COMMENT 'compensation status(SU succeed|FA failed|UN unknown|SK skipped|RU running)',
56.     "IS_RUNNING" TINYINT COMMENT 'is running(0 no|1 yes)',
57.     "GMT_UPDATED" TIMESTAMP NOT NULL
58. );

复制代码

(2)Seata Saga案例的工程启动
ApplicationKeeper启动完之后就会被阻塞住。

1. public class ApplicationKeeper {
2.     private static final Logger LOGGER = LoggerFactory.getLogger(ApplicationKeeper.class);
3.     private final ReentrantLock LOCK = new ReentrantLock();
4.     private final Condition STOP = LOCK.newCondition();
6.     //Instantiates a new Application keeper.
7.     public ApplicationKeeper(AbstractApplicationContext applicationContext) {
8.         addShutdownHook(applicationContext);
9.     }
11.     private void addShutdownHook(final AbstractApplicationContext applicationContext) {
12.         Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
13.             @Override
14.             public void run() {
15.                 try {
16.                     applicationContext.close();
17.                     LOGGER.info("ApplicationContext " + applicationContext + " is closed.");
18.                 } catch (Exception e) {
19.                     LOGGER.error("Failed to close ApplicationContext", e);
20.                 }
21.                 try {
22.                     LOCK.lock();
23.                     STOP.signal();
24.                 } finally {
25.                     LOCK.unlock();
26.                 }
27.             }
28.         }));
29.     }
31.     public void keep() {
32.         synchronized (LOCK) {
33.             try {
34.                 LOGGER.info("Application is keep running ... ");
35.                 LOCK.wait();
36.             } catch (InterruptedException e) {
37.                 e.printStackTrace();
38.             }
39.         }
40.     }
41. }

复制代码

步骤一：运行DubboSagaProviderStarter，启动Dubbo Provider。

1. public class DubboSagaProviderStarter {
2.     private static TestingServer server;
4.     //The entry point of application.
5.     public static void main(String[] args) throws Exception {
6.         //mock zk server
7.         mockZKServer();
8.         ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext(
9.             new String[]{"spring/seata-dubbo-provider.xml"}
10.         );
11.         new ApplicationKeeper(applicationContext).keep();
12.     }
14.     private static void mockZKServer() throws Exception {
15.         //Mock zk server，作为Dubbo配置中心
16.         server = new TestingServer(2181, true);
17.         server.start();
18.     }
19. }
21. public class InventoryActionImpl implements InventoryAction {
22.     private static final Logger LOGGER = LoggerFactory.getLogger(InventoryActionImpl.class);
23.    
24.     @Override
25.     public boolean reduce(String businessKey, int count) {
26.         LOGGER.info("reduce inventory succeed, count: " + count + ", businessKey:" + businessKey);
27.         return true;
28.     }
29.    
30.     @Override
31.     public boolean compensateReduce(String businessKey) {
32.         LOGGER.info("compensate reduce inventory succeed, businessKey:" + businessKey);
33.         return true;
34.     }
35. }
37. public class BalanceActionImpl implements BalanceAction {
38.     private static final Logger LOGGER = LoggerFactory.getLogger(BalanceActionImpl.class);
39.    
40.     @Override
41.     public boolean reduce(String businessKey, BigDecimal amount, Map<String, Object> params) {
42.         if (params != null) {
43.             Object throwException = params.get("throwException");
44.             if (throwException != null && "true".equals(throwException.toString())) {
45.                 throw new RuntimeException("reduce balance failed");
46.             }
47.         }
48.         LOGGER.info("reduce balance succeed, amount: " + amount + ", businessKey:" + businessKey);
49.         return true;
50.     }
52.     @Override
53.     public boolean compensateReduce(String businessKey, Map<String, Object> params) {
54.         if (params != null) {
55.             Object throwException = params.get("throwException");
56.             if (throwException != null && "true".equals(throwException.toString())) {
57.                 throw new RuntimeException("compensate reduce balance failed");
58.             }
59.         }
60.         LOGGER.info("compensate reduce balance succeed, businessKey:" + businessKey);
61.         return true;
62.     }
63. }

复制代码

步骤二：运行DubboSagaTransactionStarter，启动Demo工程。

1. public class DubboSagaTransactionStarter {
2.     private static volatile Object lock = new Object();
3.    
4.     private static AsyncCallback CALL_BACK = new AsyncCallback() {
5.         @Override
6.         public void onFinished(ProcessContext context, StateMachineInstance stateMachineInstance) {
7.             synchronized (lock) {
8.                 lock.notifyAll();
9.             }
10.         }
11.         
12.         @Override
13.         public void onError(ProcessContext context, StateMachineInstance stateMachineInstance, Exception exp) {
14.             synchronized (lock) {
15.                 lock.notifyAll();
16.             }
17.         }
18.     };
20.     public static void main(String[] args) {
21.         AbstractApplicationContext applicationContext = new ClassPathXmlApplicationContext(
22.             new String[]{"spring/seata-saga.xml", "spring/seata-dubbo-reference.xml"}
23.         );
24.         StateMachineEngine stateMachineEngine = (StateMachineEngine) applicationContext.getBean("stateMachineEngine");
25.         transactionCommittedDemo(stateMachineEngine);
26.         transactionCompensatedDemo(stateMachineEngine);
27.         new ApplicationKeeper(applicationContext).keep();
28.     }
30.     private static void transactionCommittedDemo(StateMachineEngine stateMachineEngine) {
31.         //设置状态机上下文
32.         Map<String, Object> startParams = new HashMap<>(3);
33.         String businessKey = String.valueOf(System.currentTimeMillis());
34.         startParams.put("businessKey", businessKey);
35.         startParams.put("count", 10);
36.         startParams.put("amount", new BigDecimal("100"));
38.         //sync test，同步启动状态机
39.         StateMachineInstance inst = stateMachineEngine.startWithBusinessKey("reduceInventoryAndBalance", null, businessKey, startParams);
40.         Assert.isTrue(ExecutionStatus.SU.equals(inst.getStatus()), "saga transaction execute failed. XID: " + inst.getId());
41.         System.out.println("saga transaction commit succeed. XID: " + inst.getId());
43.         //async test，异步启动状态机
44.         businessKey = String.valueOf(System.currentTimeMillis());
45.         inst = stateMachineEngine.startWithBusinessKeyAsync("reduceInventoryAndBalance", null, businessKey, startParams, CALL_BACK);
47.         waittingForFinish(inst);
49.         Assert.isTrue(ExecutionStatus.SU.equals(inst.getStatus()), "saga transaction execute failed. XID: " + inst.getId());
50.         System.out.println("saga transaction commit succeed. XID: " + inst.getId());
51.     }
53.     private static void transactionCompensatedDemo(StateMachineEngine stateMachineEngine) {
54.         //设置状态机上下文
55.         Map<String, Object> startParams = new HashMap<>(4);
56.         String businessKey = String.valueOf(System.currentTimeMillis());
57.         startParams.put("businessKey", businessKey);
58.         startParams.put("count", 10);
59.         startParams.put("amount", new BigDecimal("100"));
60.         startParams.put("mockReduceBalanceFail", "true");
62.         //sync test，同步启动状态机
63.         StateMachineInstance inst = stateMachineEngine.startWithBusinessKey("reduceInventoryAndBalance", null, businessKey, startParams);
65.         //async test，异步启动状态机
66.         businessKey = String.valueOf(System.currentTimeMillis());
67.         inst = stateMachineEngine.startWithBusinessKeyAsync("reduceInventoryAndBalance", null, businessKey, startParams, CALL_BACK);
69.         waittingForFinish(inst);
71.         Assert.isTrue(ExecutionStatus.SU.equals(inst.getCompensationStatus()), "saga transaction compensate failed. XID: " + inst.getId());
72.         System.out.println("saga transaction compensate succeed. XID: " + inst.getId());
73.     }
75.     private static void waittingForFinish(StateMachineInstance inst) {
76.         synchronized (lock) {
77.             if (ExecutionStatus.RU.equals(inst.getStatus())) {
78.                 try {
79.                     lock.wait();
80.                 } catch (InterruptedException e) {
81.                     e.printStackTrace();
82.                 }
83.             }
84.         }
85.     }
86. }

复制代码

 
6.基于数据库的状态机配置实例的初始化
基于数据库的状态机配置实例DbStateMachineConfig会将状态机定义、状态机实例、状态实例放在数据库中。由seata-saga.xml配置文件可知，它会被注入到状态机引擎实例StateMachineEngine里。DbStateMachineConfig的初始化源码如下：

1. //DbStateMachineConfig表示状态机定义、状态机实例、状态实例会放在DB中
2. public class DbStateMachineConfig extends DefaultStateMachineConfig implements DisposableBean {
3.     //会通过seata-saga.xml配置文件，注入一个数据库连接池进来，这个DB会用来存放状态机定义、状态机实例、状态实例
4.     private DataSource dataSource;
5.     //应用程序ID
6.     private String applicationId;
7.     //分布式事务服务的分组
8.     private String txServiceGroup;
9.     //DB里存放状态机数据的表的前缀
10.     private String tablePrefix = "seata_";
11.     //DB的类型，seata-samples里用的是h2，一般都用MySQL
12.     private String dbType;
13.     //Saga分布式事务模版
14.     private SagaTransactionalTemplate sagaTransactionalTemplate;
15.     //是否启用RM资源管理器上报成功的机制，默认是不启用的
16.     private boolean rmReportSuccessEnable = DEFAULT_CLIENT_REPORT_SUCCESS_ENABLE;
17.     //是否启用Saga分支事务的注册机制，默认是不启用的
18.     private boolean sagaBranchRegisterEnable = DEFAULT_CLIENT_SAGA_BRANCH_REGISTER_ENABLE;
20.     //初始化
21.     public DbStateMachineConfig() {
22.         try {
23.             Configuration configuration = ConfigurationFactory.getInstance();
24.             if (configuration != null) {
25.                 this.rmReportSuccessEnable = configuration.getBoolean(ConfigurationKeys.CLIENT_REPORT_SUCCESS_ENABLE, DEFAULT_CLIENT_REPORT_SUCCESS_ENABLE);
26.                 this.sagaBranchRegisterEnable = configuration.getBoolean(ConfigurationKeys.CLIENT_SAGA_BRANCH_REGISTER_ENABLE, DEFAULT_CLIENT_SAGA_BRANCH_REGISTER_ENABLE);
27.                 //设置Saga状态机定义文件的解析器
28.                 setSagaJsonParser(configuration.getConfig(ConfigurationKeys.CLIENT_SAGA_JSON_PARSER, DEFAULT_SAGA_JSON_PARSER));
29.                 this.applicationId = configuration.getConfig(ConfigurationKeys.APPLICATION_ID);
30.                 this.txServiceGroup = configuration.getConfig(ConfigurationKeys.TX_SERVICE_GROUP);
31.                 //设置Saga重试持久化模式是否为更新模式
32.                 setSagaRetryPersistModeUpdate(configuration.getBoolean(ConfigurationKeys.CLIENT_SAGA_RETRY_PERSIST_MODE_UPDATE, DEFAULT_CLIENT_SAGA_RETRY_PERSIST_MODE_UPDATE));
33.                 //设置Saga补偿持久化模式是否为更新模式
34.                 setSagaCompensatePersistModeUpdate(configuration.getBoolean(ConfigurationKeys.CLIENT_SAGA_COMPENSATE_PERSIST_MODE_UPDATE, DEFAULT_CLIENT_SAGA_COMPENSATE_PERSIST_MODE_UPDATE));
35.             }
36.         } catch (Exception e) {
37.             LOGGER.warn("Load SEATA configuration failed, use default configuration instead.", e);
38.         }
39.     }
41.     //根据数据库连接池获取DB类型
42.     public static String getDbTypeFromDataSource(DataSource dataSource) throws SQLException {
43.         try (Connection con = dataSource.getConnection()) {
44.             DatabaseMetaData metaData = con.getMetaData();
45.             return metaData.getDatabaseProductName();
46.         }
47.     }
49.     //初始化DB状态机配置DbStateMachineConfig
50.     //因为DbStateMachineConfig本身就是Spring Bean，当该Bean初始化完毕后，就会执行afterPropertiesSet()方法
51.     @Override
52.     public void afterPropertiesSet() throws Exception {
53.         //先根据数据库连接池获取DB类型
54.         dbType = getDbTypeFromDataSource(dataSource);
55.         //如果状态机实例日志和状态实例日志存储组件为null
56.         if (getStateLogStore() == null) {
57.             DbAndReportTcStateLogStore dbStateLogStore = new DbAndReportTcStateLogStore();
58.             dbStateLogStore.setDataSource(dataSource);
59.             dbStateLogStore.setTablePrefix(tablePrefix);
60.             dbStateLogStore.setDbType(dbType);
61.             dbStateLogStore.setDefaultTenantId(getDefaultTenantId());
62.             dbStateLogStore.setSeqGenerator(getSeqGenerator());
63.    
64.             if (StringUtils.hasLength(getSagaJsonParser())) {
65.                 ParamsSerializer paramsSerializer = new ParamsSerializer();
66.                 paramsSerializer.setJsonParserName(getSagaJsonParser());
67.                 dbStateLogStore.setParamsSerializer(paramsSerializer);
68.             }
70.             if (sagaTransactionalTemplate == null) {
71.                 DefaultSagaTransactionalTemplate defaultSagaTransactionalTemplate = new DefaultSagaTransactionalTemplate();
72.                 defaultSagaTransactionalTemplate.setApplicationContext(getApplicationContext());
73.                 defaultSagaTransactionalTemplate.setApplicationId(applicationId);
74.                 defaultSagaTransactionalTemplate.setTxServiceGroup(txServiceGroup);
75.                 defaultSagaTransactionalTemplate.afterPropertiesSet();
76.                 sagaTransactionalTemplate = defaultSagaTransactionalTemplate;
77.             }
79.             dbStateLogStore.setSagaTransactionalTemplate(sagaTransactionalTemplate);
80.             setStateLogStore(dbStateLogStore);
81.         }
83.         if (getStateLangStore() == null) {
84.             DbStateLangStore dbStateLangStore = new DbStateLangStore();
85.             dbStateLangStore.setDataSource(dataSource);
86.             dbStateLangStore.setTablePrefix(tablePrefix);
87.             dbStateLangStore.setDbType(dbType);
88.             setStateLangStore(dbStateLangStore);
89.         }
91.         super.afterPropertiesSet();//must execute after StateLangStore initialized
92.     }
94.     @Override
95.     public void destroy() throws Exception {
96.         if ((sagaTransactionalTemplate != null) && (sagaTransactionalTemplate instanceof DisposableBean)) {
97.             ((DisposableBean) sagaTransactionalTemplate).destroy();
98.         }
99.     }
100.     ...
101. }
103. //Default state machine configuration
104. //默认的状态机配置，DB状态机配置DbStateMachineConfig是它的子类
105. //DefaultStateMachineConfig可以支持不同种类的状态机配置
106. //即状态机定义、状态机实例、状态实例，既可以放在DB里，也可以放在其他存储里，默认是放在DB里的
108. //DefaultStateMachineConfig会封装(注入)状态机运行时需要的所有组件
109. //StateMachineConfig可以获取到状态机需要的各种组件
110. //ApplicationContextAware可以感知Spring容器上下文
111. //InitializingBean可以对Spring Bean进行初始化
112. public class DefaultStateMachineConfig implements StateMachineConfig, ApplicationContextAware, InitializingBean {
113.     ...
114.     ...
115. }

复制代码

 
7.状态机配置实例中包含的一些关键组件
(1)状态机配置组件
(2)状态机实例和状态实例的仓储组件
(3)状态机实例日志和状态实例日志的存储组件
(4)状态机定义的存储组件
(5)表达式工厂管理器
(6)状态机定义的仓储组件
 
(1)状态机配置组件
StateMachineConfig表示的是状态机配置组件，可以从中获取状态机需要的各种组件。

1. //StateMachineConfig
2. //状态机配置组件接口，可以获取到状态机需要的各种组件
3. public interface StateMachineConfig {
4.     //Gets state log store.
5.     //获取状态机实例和状态实例的仓储组件
6.     //StateLogRepository可以根据各种条件查询状态机实例和状态实例
7.     StateLogRepository getStateLogRepository();
9.     //Gets get state log store.
10.     //获取状态机实例日志和状态实例日志的存储组件
11.     //StateLogStore可以对状态机日志和状态日志进行读写、也可以根据各种条件查询状态机实例和状态实例
12.     StateLogStore getStateLogStore();
14.     //Gets get state language definition store.
15.     //获取状态机定义的存储组件
16.     StateLangStore getStateLangStore();
18.     //Gets get expression factory manager.
19.     //获取表达式工厂管理器
20.     ExpressionFactoryManager getExpressionFactoryManager();
22.     //Gets get evaluator factory manager.
23.     //获取表达式计算工厂管理器
24.     EvaluatorFactoryManager getEvaluatorFactoryManager();
26.     //Gets get charset.
27.     //获取字符集编码
28.     String getCharset();
30.     //Gets get default tenant id.
31.     //获取默认的租户ID
32.     String getDefaultTenantId();
34.     //Gets get state machine repository.
35.     //获取状态机定义的仓储组件
36.     StateMachineRepository getStateMachineRepository();
38.     //Gets get status decision strategy.
39.     //获取State执行结果的判定策略组件
40.     StatusDecisionStrategy getStatusDecisionStrategy();
42.     //Gets get seq generator.
43.     //获取序号生成器
44.     SeqGenerator getSeqGenerator();
46.     //Gets get process ctrl event publisher.
47.     //获取流程控制事件的发布器(同步发布)
48.     ProcessCtrlEventPublisher getProcessCtrlEventPublisher();
50.     //Gets get async process ctrl event publisher.
51.     //获取流程控制事件的发布器(异步发布)
52.     ProcessCtrlEventPublisher getAsyncProcessCtrlEventPublisher();
54.     //Gets get application context.
55.     //获取Spring容器上下文
56.     ApplicationContext getApplicationContext();
58.     //Gets get thread pool executor.
59.     //获取异步执行的线程池
60.     ThreadPoolExecutor getThreadPoolExecutor();
62.     //Is enable async boolean.
63.     //是否启用异步
64.     boolean isEnableAsync();
66.     //get ServiceInvokerManager
67.     //获取服务调用的管理器
68.     ServiceInvokerManager getServiceInvokerManager();
70.     //get trans operation timeout
71.     //获取事务操作的超时时间
72.     int getTransOperationTimeout();
74.     //get service invoke timeout
75.     //获取服务调用的超时时间
76.     int getServiceInvokeTimeout();
78.     //get ScriptEngineManager
79.     //获取脚本引擎管理器
80.     ScriptEngineManager getScriptEngineManager();
81. }

复制代码

(2)状态机实例和状态实例的仓储组件
StateLogRepository表示的是状态机实例和状态实例的仓储组件，可以根据不同的条件查询状态机实例和状态实例。

1. //State Log Repository
2. //状态机实例和状态实例的仓储组件，可以根据各种条件查询状态机实例和状态实例
3. public interface StateLogRepository {
4.     //Get state machine instance
5.     //根据状态机实例ID，获取到具体的状态机实例
6.     StateMachineInstance getStateMachineInstance(String stateMachineInstanceId);
8.     //Get state machine instance by businessKey
9.     //根据业务key获取状态机实例
10.     StateMachineInstance getStateMachineInstanceByBusinessKey(String businessKey, String tenantId);
12.     //Query the list of state machine instances by parent id
13.     //根据父ID查询状态机实例
14.     List<StateMachineInstance> queryStateMachineInstanceByParentId(String parentId);
16.     //Get state instance
17.     //根据状态实例ID查询状态实例
18.     StateInstance getStateInstance(String stateInstanceId, String machineInstId);
20.     //Get a list of state instances by state machine instance id
21.     //根据状态机实例ID查询所有状态实例
22.     List<StateInstance> queryStateInstanceListByMachineInstanceId(String stateMachineInstanceId);
23. }

复制代码

(3)状态机实例日志和状态实例日志的存储组件接口
StateLogStore表示的是状态机实例日志和状态实例日志的存储组件，可以记录各种状态机实例日志和状态实例日志。

1. //StateMachine engine log store
2. //状态机实例日志和状态实例日志的存储组件
3. public interface StateLogStore {
4.     //Record state machine startup events
5.     //记录状态机实例的启动事件日志
6.     void recordStateMachineStarted(StateMachineInstance machineInstance, ProcessContext context);
8.     //Record status end event
9.     //记录状态机实例的完成事件日志
10.     void recordStateMachineFinished(StateMachineInstance machineInstance, ProcessContext context);
12.     //Record state machine restarted
13.     //记录状态机实例的重启事件日志
14.     void recordStateMachineRestarted(StateMachineInstance machineInstance, ProcessContext context);
16.     //Record state start execution event
17.     //记录状态实例的启动事件日志
18.     void recordStateStarted(StateInstance stateInstance, ProcessContext context);
20.     //Record state execution end event
21.     //记录状态实例的完成事件日志
22.     void recordStateFinished(StateInstance stateInstance, ProcessContext context);
24.     //Get state machine instance
25.     //根据状态机实例ID获取状态机实例
26.     StateMachineInstance getStateMachineInstance(String stateMachineInstanceId);
28.     //Get state machine instance by businessKey
29.     //根据业务key获取状态机实例
30.     StateMachineInstance getStateMachineInstanceByBusinessKey(String businessKey, String tenantId);
32.     //Query the list of state machine instances by parent id
33.     //根据父ID查询状态机实例
34.     List<StateMachineInstance> queryStateMachineInstanceByParentId(String parentId);
36.     //Get state instance
37.     //根据状态实例ID和状态机ID获取状态实例
38.     StateInstance getStateInstance(String stateInstanceId, String machineInstId);
40.     //Get a list of state instances by state machine instance id
41.     //根据状态机实例ID查询状态实例
42.     List<StateInstance> queryStateInstanceListByMachineInstanceId(String stateMachineInstanceId);
43. }

复制代码

(4)状态机定义的存储组件接口
StateLangStore表示的是状态机定义的存储组件，可以存储和获取不同的状态机定义。

1. //State language definition store
2. //状态机定义的存储组件
3. public interface StateLangStore {
4.     //Query the state machine definition by id
5.     //根据状态机ID获取状态机定义
6.     StateMachine getStateMachineById(String stateMachineId);
8.     //Get the latest version of the state machine by state machine name
9.     //根据状态机名称和租户ID，可以获取最新版本的状态机定义
10.     StateMachine getLastVersionStateMachine(String stateMachineName, String tenantId);
12.     //Storage state machine definition
13.     //存储状态机定义
14.     boolean storeStateMachine(StateMachine stateMachine);
15. }

复制代码

(5)表达式工厂管理器
ExpressionFactoryManager表示的是表达式工厂管理器，可以根据不同的表达式类型获取不同的表达式工厂组件。

1. //Expression factory manager
2. //表达式工厂管理器
3. public class ExpressionFactoryManager {
4.     public static final String DEFAULT_EXPRESSION_TYPE = "Default";
5.     //表达式类型 -> 表达式工厂
6.     private Map<String, ExpressionFactory> expressionFactoryMap = new ConcurrentHashMap<>();
7.    
8.     //根据不同的表达式类型，去获取不同的表达式工厂组件
9.     public ExpressionFactory getExpressionFactory(String expressionType) {
10.         if (StringUtils.isBlank(expressionType)) {
11.             expressionType = DEFAULT_EXPRESSION_TYPE;
12.         }
13.         return expressionFactoryMap.get(expressionType);
14.     }
15.    
16.     public void setExpressionFactoryMap(Map<String, ExpressionFactory> expressionFactoryMap) {
17.         this.expressionFactoryMap.putAll(expressionFactoryMap);
18.     }
19.    
20.     public void putExpressionFactory(String type, ExpressionFactory factory) {
21.         this.expressionFactoryMap.put(type, factory);
22.     }
23. }

复制代码

(6)状态机定义的仓储组件
StateMachineRepository表示的是状态机定义的仓储组件，可以注册和获取一个状态机定义。

1. //StateMachineRepository
2. //状态机定义的仓储组件
3. public interface StateMachineRepository {
4.     //Gets get state machine by id.
5.     //根据状态机ID获取一个状态机定义StateMachine
6.     StateMachine getStateMachineById(String stateMachineId);
8.     //Gets get state machine.
9.     //根据状态机名字、租户ID获取一个状态机定义StateMachine
10.     StateMachine getStateMachine(String stateMachineName, String tenantId);
12.     //Gets get state machine.
13.     //根据状态机名字、租户ID、版本号获取一个状态机定义StateMachine
14.     StateMachine getStateMachine(String stateMachineName, String tenantId, String version);
16.     //Register the state machine to the repository (if the same version already exists, return the existing version)
17.     //向状态机定义的仓储组件注册一个状态机定义
18.     StateMachine registryStateMachine(StateMachine stateMachine);
20.     //registry by resources
21.     //向状态机定义的仓储组件注册资源
22.     void registryByResources(Resource[] resources, String tenantId) throws IOException;
23. }

复制代码

 
8.默认的状态机配置类的初始化
DefaultStateMachineConfig是默认的状态机配置类，基于数据库的状态机配置组件DbStateMachineConfig是它的子类。它可以支持不同种类的状态机配置，也就是状态机定义、状态机实例、状态实例，既可以放在数据库里，也可以放在其他存储里，默认是放在数据库里的。

1. //Default state machine configuration
2. //DefaultStateMachineConfig会封装(注入)状态机运行时需要的所有组件
3. //StateMachineConfig可以获取到状态机需要的各种组件
4. //ApplicationContextAware可以感知Spring容器上下文
5. //InitializingBean可以对Spring Bean进行初始化
6. public class DefaultStateMachineConfig implements StateMachineConfig, ApplicationContextAware, InitializingBean {
7.     private static final int DEFAULT_TRANS_OPER_TIMEOUT     = 60000 * 30;
8.     private static final int DEFAULT_SERVICE_INVOKE_TIMEOUT = 60000 * 5;
10.     //事务操作的超时时间，默认是30分钟
11.     private int transOperationTimeout = DEFAULT_TRANS_OPER_TIMEOUT;
12.     //服务调用的超时时间，默认是5分钟
13.     private int serviceInvokeTimeout  = DEFAULT_SERVICE_INVOKE_TIMEOUT;
15.     //状态机实例和状态实例的仓储组件
16.     private StateLogRepository stateLogRepository;
17.     //状态机实例日志和状态实例日志的存储组件
18.     private StateLogStore stateLogStore;
19.     //状态机定义的存储组件
20.     private StateLangStore stateLangStore;
21.     //表达式工厂管理器
22.     private ExpressionFactoryManager expressionFactoryManager;
23.     //表达式计算的工厂管理器
24.     private EvaluatorFactoryManager evaluatorFactoryManager;
25.     //状态机定义的仓储组件
26.     private StateMachineRepository stateMachineRepository;
27.     //State执行结果的判定策略组件
28.     private StatusDecisionStrategy statusDecisionStrategy;
29.     //序号生成器
30.     private SeqGenerator seqGenerator;
32.     //流程控制的事件发布器(同步发布)
33.     private ProcessCtrlEventPublisher syncProcessCtrlEventPublisher;
34.     //流程控制的事件发布器(异步发布)
35.     private ProcessCtrlEventPublisher asyncProcessCtrlEventPublisher;
36.     //Spring容器上下文
37.     private ApplicationContext applicationContext;
38.     //异步执行的线程池
39.     private ThreadPoolExecutor threadPoolExecutor;
40.     //是否启用异步执行，默认是false
41.     private boolean enableAsync = false;
42.     //服务调用的管理器
43.     private ServiceInvokerManager serviceInvokerManager;
45.     //是否自动注册资源(自动注册流程定义)，默认是true
46.     private boolean autoRegisterResources = true;
47.     //默认的状态机定义文件(json文件)
48.     private String[] resources = new String[]{"classpath*:seata/saga/statelang/**/*.json"};
49.     //字符集编码
50.     private String charset = "UTF-8";
51.     //默认租户ID
52.     private String defaultTenantId = "000001";
53.     //脚本引擎管理器
54.     private ScriptEngineManager scriptEngineManager;
55.     //状态机定义文件(json文件)解析器
56.     private String sagaJsonParser = DEFAULT_SAGA_JSON_PARSER;
57.     //是否更新Saga重试持久化的模式
58.     private boolean sagaRetryPersistModeUpdate = DEFAULT_CLIENT_SAGA_RETRY_PERSIST_MODE_UPDATE;
59.     //是否更新Saga补偿持久化的模式
60.     private boolean sagaCompensatePersistModeUpdate = DEFAULT_CLIENT_SAGA_COMPENSATE_PERSIST_MODE_UPDATE;
62.     //因为DefaultStateMachineConfig实现了InitializingBean接口
63.     //所以继承DefaultStateMachineConfig的Spring Bean初始化之后，就会回调afterPropertiesSet()方法
64.     @Override
65.     public void afterPropertiesSet() throws Exception {
66.         init();
67.     }
69.     //初始化方法
70.     protected void init() throws Exception {
71.         //创建表达式工厂管理器
72.         if (expressionFactoryManager == null) {
73.             //创建一个表达式工厂管理器
74.             expressionFactoryManager = new ExpressionFactoryManager();
76.             //创建一个Spring EL表达式工厂
77.             SpringELExpressionFactory springELExpressionFactory = new SpringELExpressionFactory();
78.             //注入(设置)Spring容器上下文到Spring EL表达式工厂中
79.             springELExpressionFactory.setApplicationContext(getApplicationContext());
80.             //将Spring EL表达式工厂放入表达式工厂管理器中
81.             expressionFactoryManager.putExpressionFactory(ExpressionFactoryManager.DEFAULT_EXPRESSION_TYPE, springELExpressionFactory);
83.             //创建一个序号表达式工厂
84.             SequenceExpressionFactory sequenceExpressionFactory = new SequenceExpressionFactory();
85.             //注入(设置)序号生成器到序号表达式工厂中
86.             sequenceExpressionFactory.setSeqGenerator(getSeqGenerator());
87.             //将序号表达式工厂放入表达式工厂管理器中
88.             expressionFactoryManager.putExpressionFactory(DomainConstants.EXPRESSION_TYPE_SEQUENCE, sequenceExpressionFactory);
89.         }
91.         //创建表达式计算工厂管理器
92.         if (evaluatorFactoryManager == null) {
93.             //创建一个表达式计算工厂管理器
94.             evaluatorFactoryManager = new EvaluatorFactoryManager();
96.             //创建一个表达式计算工厂，并将该表达式计算工厂放入表达式计算工厂管理器中
97.             ExpressionEvaluatorFactory expressionEvaluatorFactory = new ExpressionEvaluatorFactory();
98.             expressionEvaluatorFactory.setExpressionFactory(expressionFactoryManager.getExpressionFactory(ExpressionFactoryManager.DEFAULT_EXPRESSION_TYPE));
99.             evaluatorFactoryManager.putEvaluatorFactory(EvaluatorFactoryManager.EVALUATOR_TYPE_DEFAULT, expressionEvaluatorFactory);
100.             //创建一个异常匹配计算工厂，并将该异常匹配计算工厂放入表达式计算工厂管理器中
101.             evaluatorFactoryManager.putEvaluatorFactory(DomainConstants.EVALUATOR_TYPE_EXCEPTION, new ExceptionMatchEvaluatorFactory());
102.         }
104.         //创建状态机定义的仓储组件
105.         if (stateMachineRepository == null) {
106.             StateMachineRepositoryImpl stateMachineRepository = new StateMachineRepositoryImpl();
107.             stateMachineRepository.setCharset(charset);//设置字符集编码
108.             stateMachineRepository.setSeqGenerator(seqGenerator);//设置序号生成器
109.             stateMachineRepository.setStateLangStore(stateLangStore);//设置状态机定义的存储组件
110.             stateMachineRepository.setDefaultTenantId(defaultTenantId);//设置默认租户ID
111.             stateMachineRepository.setJsonParserName(sagaJsonParser);//设置状态机定义文件(json文件)解析器
112.             this.stateMachineRepository = stateMachineRepository;
113.         }
115.         //stateMachineRepository may be overridden, so move `stateMachineRepository.registryByResources()` here.
116.         //如果需要自动注册资源(比如数据库等)，则获取资源并进行注册
117.         if (autoRegisterResources && ArrayUtils.isNotEmpty(resources)) {
118.             try {
119.                 //读取默认的状态机定义文件(json文件)，把这个状态机定义注册到状态机定义的仓储组件中
120.                 Resource[] resources = ResourceUtil.getResources(this.resources);
121.                 stateMachineRepository.registryByResources(resources, defaultTenantId);
122.             } catch (IOException e) {
123.                 LOGGER.error("Load State Language Resources failed.", e);
124.             }
125.         }
127.         //创建状态机实例和状态实例的仓储组件
128.         if (stateLogRepository == null) {
129.             StateLogRepositoryImpl stateLogRepositoryImpl = new StateLogRepositoryImpl();
130.             stateLogRepositoryImpl.setStateLogStore(stateLogStore);
131.             this.stateLogRepository = stateLogRepositoryImpl;
132.         }
134.         //创建State执行结果的判定策略组件
135.         if (statusDecisionStrategy == null) {
136.             statusDecisionStrategy = new DefaultStatusDecisionStrategy();
137.         }
139.         //创建流程控制的事件发布器(同步发布)
140.         if (syncProcessCtrlEventPublisher == null) {
141.             //创建流程控制的事件发布器
142.             ProcessCtrlEventPublisher syncEventPublisher = new ProcessCtrlEventPublisher();
143.             //创建流程控制器
144.             ProcessControllerImpl processorController = createProcessorController(syncEventPublisher);
146.             //创建流程控制事件的消费者
147.             ProcessCtrlEventConsumer processCtrlEventConsumer = new ProcessCtrlEventConsumer();
148.             processCtrlEventConsumer.setProcessController(processorController);
150.             //创建事件总线
151.             DirectEventBus directEventBus = new DirectEventBus();
152.             syncEventPublisher.setEventBus(directEventBus);
154.             directEventBus.registerEventConsumer(processCtrlEventConsumer);
155.             syncProcessCtrlEventPublisher = syncEventPublisher;
156.         }
158.         //如果启用了异步化执行 且 流程控制的事件发布器(异步发布)为null
159.         if (enableAsync && asyncProcessCtrlEventPublisher == null) {
160.             ProcessCtrlEventPublisher asyncEventPublisher = new ProcessCtrlEventPublisher();
161.             ProcessControllerImpl processorController = createProcessorController(asyncEventPublisher);
163.             ProcessCtrlEventConsumer processCtrlEventConsumer = new ProcessCtrlEventConsumer();
164.             processCtrlEventConsumer.setProcessController(processorController);
166.             AsyncEventBus asyncEventBus = new AsyncEventBus();
167.             asyncEventBus.setThreadPoolExecutor(getThreadPoolExecutor());
168.             asyncEventPublisher.setEventBus(asyncEventBus);
170.             asyncEventBus.registerEventConsumer(processCtrlEventConsumer);
171.             asyncProcessCtrlEventPublisher = asyncEventPublisher;
172.         }
174.         //创建服务调用管理器
175.         if (this.serviceInvokerManager == null) {
176.             this.serviceInvokerManager = new ServiceInvokerManager();
178.             //创建Spring Bean的服务调用组件
179.             SpringBeanServiceInvoker springBeanServiceInvoker = new SpringBeanServiceInvoker();
180.             springBeanServiceInvoker.setApplicationContext(getApplicationContext());
181.             springBeanServiceInvoker.setThreadPoolExecutor(threadPoolExecutor);
182.             springBeanServiceInvoker.setSagaJsonParser(getSagaJsonParser());
183.             //将Spring Bean的服务调用组件放入到服务调用管理器中
184.             this.serviceInvokerManager.putServiceInvoker(DomainConstants.SERVICE_TYPE_SPRING_BEAN, springBeanServiceInvoker);
185.         }
187.         //创建脚本引擎管理器
188.         if (this.scriptEngineManager == null) {
189.             this.scriptEngineManager = new ScriptEngineManager();
190.         }
191.     }
193.     protected ProcessControllerImpl createProcessorController(ProcessCtrlEventPublisher eventPublisher) throws Exception {
194.         //创建状态机流程处理的路由器
195.         StateMachineProcessRouter stateMachineProcessRouter = new StateMachineProcessRouter();
196.         stateMachineProcessRouter.initDefaultStateRouters();
197.         //通过SPI机制加载状态机流程处理路由的拦截器
198.         loadStateRouterInterceptors(stateMachineProcessRouter.getStateRouters());
200.         //创建状态机流程处理器
201.         StateMachineProcessHandler stateMachineProcessHandler = new StateMachineProcessHandler();
202.         stateMachineProcessHandler.initDefaultHandlers();
203.         //通过SPI机制加载状态机流程处理器的拦截器
204.         loadStateHandlerInterceptors(stateMachineProcessHandler.getStateHandlers());
206.         //创建默认的路由处理器，并设置事件发布器
207.         DefaultRouterHandler defaultRouterHandler = new DefaultRouterHandler();
208.         defaultRouterHandler.setEventPublisher(eventPublisher);
210.         //创建状态机流程处理路由器对应的Map
211.         Map<String, ProcessRouter> processRouterMap = new HashMap<>(1);
212.         processRouterMap.put(ProcessType.STATE_LANG.getCode(), stateMachineProcessRouter);
213.         defaultRouterHandler.setProcessRouters(processRouterMap);
215.         //创建自定义的业务处理器
216.         CustomizeBusinessProcessor customizeBusinessProcessor = new CustomizeBusinessProcessor();
218.         //创建状态机流程处理器对应的Map
219.         Map<String, ProcessHandler> processHandlerMap = new HashMap<>(1);
220.         processHandlerMap.put(ProcessType.STATE_LANG.getCode(), stateMachineProcessHandler);
221.         customizeBusinessProcessor.setProcessHandlers(processHandlerMap);
223.         //创建路由处理器对应的Map
224.         Map<String, RouterHandler> routerHandlerMap = new HashMap<>(1);
225.         routerHandlerMap.put(ProcessType.STATE_LANG.getCode(), defaultRouterHandler);
226.         customizeBusinessProcessor.setRouterHandlers(routerHandlerMap);
228.         //创建流程控制器
229.         ProcessControllerImpl processorController = new ProcessControllerImpl();
230.         processorController.setBusinessProcessor(customizeBusinessProcessor);
232.         return processorController;
233.     }
235.     protected void loadStateHandlerInterceptors(Map<String, StateHandler> stateHandlerMap) {
236.         for (StateHandler stateHandler : stateHandlerMap.values()) {
237.             if (stateHandler instanceof InterceptableStateHandler) {
238.                 InterceptableStateHandler interceptableStateHandler = (InterceptableStateHandler) stateHandler;
239.                 List<StateHandlerInterceptor> interceptorList = EnhancedServiceLoader.loadAll(StateHandlerInterceptor.class);
240.                 for (StateHandlerInterceptor interceptor : interceptorList) {
241.                     if (interceptor.match(interceptableStateHandler.getClass())) {
242.                         interceptableStateHandler.addInterceptor(interceptor);
243.                     }
244.                     if (interceptor instanceof ApplicationContextAware) {
245.                         ((ApplicationContextAware) interceptor).setApplicationContext(getApplicationContext());
246.                     }
247.                 }
248.             }
249.         }
250.     }
252.     protected void loadStateRouterInterceptors(Map<String, StateRouter> stateRouterMap) {
253.         for (StateRouter stateRouter : stateRouterMap.values()) {
254.             if (stateRouter instanceof InterceptableStateRouter) {
255.                 InterceptableStateRouter interceptableStateRouter = (InterceptableStateRouter) stateRouter;
256.                 List<StateRouterInterceptor> interceptorList = EnhancedServiceLoader.loadAll(StateRouterInterceptor.class);
257.                 for (StateRouterInterceptor interceptor : interceptorList) {
258.                     if (interceptor.match(interceptableStateRouter.getClass())) {
259.                         interceptableStateRouter.addInterceptor(interceptor);
260.                     }
261.                     if (interceptor instanceof ApplicationContextAware) {
262.                         ((ApplicationContextAware) interceptor).setApplicationContext(getApplicationContext());
263.                     }
264.                 }
265.             }
266.         }
267.     }
268.     ...
269. }

复制代码

 
9.状态机定义的仓储组件解析状态机定义文件
在DefaultStateMachineConfig的init()方法初始化逻辑中，会通过调用状态机定义的仓储组件StateMachineRepository.registryByResources()方法来读取和解析状态机定义文件，并把解析出来的状态机定义StateMachine注册到状态机定义的仓储组件中，也就是存储在stateMachineRepository变量里。
 
注意：StateMachine就是状态机定义。

1. public class StateMachineRepositoryImpl implements StateMachineRepository {
2.     ...
3.     @Override
4.     public void registryByResources(Resource[] resources, String tenantId) throws IOException {
5.         if (resources != null) {
6.             for (Resource resource : resources) {
7.                 //1.基于IO流把状态机定义文件(json文件)读取出来，放入json字符串
8.                 String json;
9.                 try (InputStream is = resource.getInputStream()) {
10.                     json = IOUtils.toString(is, charset);
11.                 }
13.                 //2.从状态机解析器工厂中获取一个状态机解析器，然后解析json字符串得到一个状态机定义StateMachine
14.                 StateMachine stateMachine = StateMachineParserFactory.getStateMachineParser(jsonParserName).parse(json);
15.                 if (stateMachine != null) {
16.                     stateMachine.setContent(json);
17.                     if (StringUtils.isBlank(stateMachine.getTenantId())) {
18.                         stateMachine.setTenantId(tenantId);
19.                     }
20.                     //3.注册状态机定义StateMachine
21.                     registryStateMachine(stateMachine);
22.                     if (LOGGER.isDebugEnabled()) {
23.                         LOGGER.debug("===== StateMachine Loaded: \n{}", json);
24.                     }
25.                 }
26.             }
27.         }
28.     }
29.     ...
30. }

复制代码

 
10.状态机定义的仓储组件注册StateMachine

1. public class StateMachineRepositoryImpl implements StateMachineRepository {
2.     private Map<String/** Name_Tenant **/, Item> stateMachineMapByNameAndTenant = new ConcurrentHashMap<>();
3.     private Map<String/** Id **/, Item> stateMachineMapById = new ConcurrentHashMap<>();
4.     private StateLangStore stateLangStore;//状态机定义的存储组件
5.     ...
6.    
7.     @Override
8.     public StateMachine registryStateMachine(StateMachine stateMachine) {
9.         String stateMachineName = stateMachine.getName();
10.         String tenantId = stateMachine.getTenantId();
12.         if (stateLangStore != null) {
13.             //1.从状态机定义的存储组件中，根据状态机名称和租户ID，获取最新版本的状态机定义StateMachine
14.             StateMachine oldStateMachine = stateLangStore.getLastVersionStateMachine(stateMachineName, tenantId);
16.             if (oldStateMachine != null) {
17.                 byte[] oldBytesContent = null;
18.                 byte[] bytesContent = null;
19.                 try {
20.                     oldBytesContent = oldStateMachine.getContent().getBytes(charset);
21.                     bytesContent = stateMachine.getContent().getBytes(charset);
22.                 } catch (UnsupportedEncodingException e) {
23.                     LOGGER.error(e.getMessage(), e);
24.                 }
25.                 if (Arrays.equals(bytesContent, oldBytesContent) && stateMachine.getVersion() != null && stateMachine.getVersion().equals(oldStateMachine.getVersion())) {
26.                     LOGGER.info("StateMachine[{}] is already exist a same version", stateMachineName);
28.                     stateMachine.setId(oldStateMachine.getId());
29.                     stateMachine.setGmtCreate(oldStateMachine.getGmtCreate());
31.                     Item item = new Item(stateMachine);
32.                     stateMachineMapByNameAndTenant.put(stateMachineName + "_" + tenantId, item);
33.                     stateMachineMapById.put(stateMachine.getId(), item);
34.                     return stateMachine;
35.                 }
36.             }
37.             if (StringUtils.isBlank(stateMachine.getId())) {
38.                 //2.生成新的状态机定义StateMachine的ID
39.                 stateMachine.setId(seqGenerator.generate(DomainConstants.SEQ_ENTITY_STATE_MACHINE));
40.             }
41.             stateMachine.setGmtCreate(new Date());
42.             //3.通过状态机定义的存储组件，把状态机定义StateMachine保存起来，默认就是把状态机定义StateMachine存储到DB中
43.             stateLangStore.storeStateMachine(stateMachine);
44.         }
46.         if (StringUtils.isBlank(stateMachine.getId())) {
47.             stateMachine.setId(seqGenerator.generate(DomainConstants.SEQ_ENTITY_STATE_MACHINE));
48.         }
50.         Item item = new Item(stateMachine);
51.         //4.把状态机定义放入内存中
52.         stateMachineMapByNameAndTenant.put(stateMachineName + "_" + tenantId, item);
53.         stateMachineMapById.put(stateMachine.getId(), item);
54.         return stateMachine;
55.     }
56.     ...
57. }
59. public class DbStateLangStore extends AbstractStore implements StateLangStore {
60.     ...
61.     @Override
62.     public boolean storeStateMachine(StateMachine stateMachine) {
63.         //把状态机定义StateMachine写入到DB中
64.         return executeUpdate(stateLangStoreSqls.getInsertStateMachineSql(dbType), STATE_MACHINE_TO_STATEMENT, stateMachine) > 0;
65.     }
66.     ...
67. }

复制代码

 
11.状态机引擎接口的源码
从seata-saga.xml配置文件可知，基于数据库的状态机配置实例DbStateMachineConfig会被注入到基于流程控制的状态机引擎实例ProcessCtrlStateMachineEngine中。

1. <beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
2.        xmlns:jdbc="http://www.springframework.org/schema/jdbc"
3.        xmlns="http://www.springframework.org/schema/beans"
4.        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
5.        http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd"
6.        default-autowire="byName">
7.     ...
8.     <bean id="stateMachineEngine" >
9.         <property name="stateMachineConfig" ref="dbStateMachineConfig"></property>
10.     </bean>
11.     ...
12. </beans>

复制代码

基于流程控制的状态机引擎ProcessCtrlStateMachineEngine，会实现状态机引擎接口StateMachineEngine。

1. //ProcessCtrl-based state machine engine
2. //基于流程控制的状态机引擎
3. public class ProcessCtrlStateMachineEngine implements StateMachineEngine {
4.     //需要在seata-saga.xml配置文件中注入状态机配置
5.     private StateMachineConfig stateMachineConfig;
6.     ...
7. }
9. //State machine engine
10. //状态机引擎接口
11. public interface StateMachineEngine {
12.     //start a state machine instance
13.     //根据状态机的名称、租户ID、启动参数来启动一个状态机实例
14.     StateMachineInstance start(String stateMachineName, String tenantId, Map<String, Object> startParams) throws EngineExecutionException;
16.     //start a state machine instance with businessKey
17.     //根据状态机的名称、租户ID、业务key、启动参数来启动一个状态机实例
18.     StateMachineInstance startWithBusinessKey(String stateMachineName, String tenantId, String businessKey, Map<String, Object> startParams) throws EngineExecutionException;
20.     //start a state machine instance asynchronously
21.     //根据状态机的名称、租户ID、启动参数来启动一个状态机实例
22.     //也就是状态机实例跑完之后，会回调传入的callback()方法
23.     StateMachineInstance startAsync(String stateMachineName, String tenantId, Map<String, Object> startParams, AsyncCallback callback) throws EngineExecutionException;
25.     //start a state machine instance asynchronously with businessKey
26.     //根据状态机的名称、租户ID、业务key、启动参数来异步化启动一个状态机实例
27.     //也就是状态机实例跑完之后，会回调传入的callback()方法
28.     StateMachineInstance startWithBusinessKeyAsync(String stateMachineName, String tenantId, String businessKey, Map<String, Object> startParams, AsyncCallback callback) throws EngineExecutionException;
30.     //forward restart a failed state machine instance
31.     //重启一个失败的状态机实例
32.     StateMachineInstance forward(String stateMachineInstId, Map<String, Object> replaceParams) throws ForwardInvalidException;
34.     //forward restart a failed state machine instance asynchronously
35.     //异步化重启一个失败的状态机实例
36.     StateMachineInstance forwardAsync(String stateMachineInstId, Map<String, Object> replaceParams, AsyncCallback callback) throws ForwardInvalidException;
38.     //compensate a state machine instance
39.     //对一个状态机实例进行补偿
40.     StateMachineInstance compensate(String stateMachineInstId, Map<String, Object> replaceParams) throws EngineExecutionException;
42.     //compensate a state machine instance asynchronously
43.     //对一个状态机实例进行异步化补偿
44.     StateMachineInstance compensateAsync(String stateMachineInstId, Map<String, Object> replaceParams, AsyncCallback callback) throws EngineExecutionException;
46.     //skip current failed state instance and forward restart state machine instance
47.     //跳过当前失败的状态机实例，同时重启一个状态机实例
48.     StateMachineInstance skipAndForward(String stateMachineInstId, Map<String, Object> replaceParams) throws EngineExecutionException;
50.     //skip current failed state instance and forward restart state machine instance asynchronously
51.     StateMachineInstance skipAndForwardAsync(String stateMachineInstId, AsyncCallback callback) throws EngineExecutionException;
53.     //get state machine configurations
54.     StateMachineConfig getStateMachineConfig();
56.     //Reload StateMachine Instance
57.     StateMachineInstance reloadStateMachineInstance(String instId);
58. }

复制代码

 
12.状态机引擎创建状态机实例的源码
基于流程控制的状态机引擎实例ProcessCtrlStateMachineEngine的start()方法调用其startInternal()方法启动一个状态机实例时，会先创建一个状态机实例StateMachineInstance。
 
在调用ProcessCtrlStateMachineEngine的createMachineInstance()方法创建一个状态机实例StateMachineInstance的过程中，会先通过状态机定义的仓储组件StateMachineRepository来获取一个状态机定义StateMachine，然后将状态机定义StateMachine注入到状态机实例对象中，以此来完成状态机实例对象的实例化。

1. //ProcessCtrl-based state machine engine
2. //基于流程控制的状态机引擎
3. public class ProcessCtrlStateMachineEngine implements StateMachineEngine {
4.     //需要在seata-saga.xml配置文件中注入状态机配置实例
5.     private StateMachineConfig stateMachineConfig;
6.     ...
7.    
8.     @Override
9.     public StateMachineInstance start(String stateMachineName, String tenantId, Map<String, Object> startParams) throws EngineExecutionException {
10.         return startInternal(stateMachineName, tenantId, null, startParams, false, null);
11.     }
12.    
13.     @Override
14.     public StateMachineInstance startAsync(String stateMachineName, String tenantId, Map<String, Object> startParams, AsyncCallback callback) throws EngineExecutionException {
15.         return startInternal(stateMachineName, tenantId, null, startParams, true, callback);
16.     }
17.    
18.     @Override
19.     public StateMachineInstance startWithBusinessKey(String stateMachineName, String tenantId, String businessKey, Map<String, Object> startParams) throws EngineExecutionException {
20.         return startInternal(stateMachineName, tenantId, businessKey, startParams, false, null);
21.     }
22.    
23.     @Override
24.     public StateMachineInstance startWithBusinessKeyAsync(String stateMachineName, String tenantId, String businessKey, Map<String, Object> startParams, AsyncCallback callback) throws EngineExecutionException {
25.         return startInternal(stateMachineName, tenantId, businessKey, startParams, true, callback);
26.     }
27.     ...
28.    
29.     //启动状态机实例StateMachineInstance
30.     //@param stateMachineName 状态机名称
31.     //@param tenantId 租户ID
32.     //@param businessKey 业务key
33.     //@param startParams 状态机实例的启动参数
34.     //@param async 是否异步化运行
35.     //@param callback 异步化运行时的回调接口
36.     private StateMachineInstance startInternal(String stateMachineName, String tenantId, String businessKey, Map<String, Object> startParams, boolean async, AsyncCallback callback) throws EngineExecutionException {
37.         try {
38.             //如果指定需要异步运行，但是状态机配置里是不允许异步运行的，则会抛异常
39.             if (async && !stateMachineConfig.isEnableAsync()) {
40.                 throw new EngineExecutionException("Asynchronous start is disabled. please set StateMachineConfig.enableAsync=true first.", FrameworkErrorCode.AsynchronousStartDisabled);
41.             }
42.             if (StringUtils.isEmpty(tenantId)) {
43.                 tenantId = stateMachineConfig.getDefaultTenantId();
44.             }
46.             //创建状态机实例StateMachineInstance
47.             StateMachineInstance instance = createMachineInstance(stateMachineName, tenantId, businessKey, startParams);
48.             ...
49.         }
50.         ...
51.     }
52.    
53.     private StateMachineInstance createMachineInstance(String stateMachineName, String tenantId, String businessKey, Map<String, Object> startParams) {
54.         //通过状态机定义的仓储组件StateMachineRepository，来获取一个状态机定义StateMachine
55.         StateMachine stateMachine = stateMachineConfig.getStateMachineRepository().getStateMachine(stateMachineName, tenantId);
56.         if (stateMachine == null) {
57.             throw new EngineExecutionException("StateMachine[" + stateMachineName + "] is not exists", FrameworkErrorCode.ObjectNotExists);
58.         }
60.         //根据状态机定义StateMachine，实例化一个状态机实例对象StateMachineInstanceImpl
61.         StateMachineInstanceImpl inst = new StateMachineInstanceImpl();
62.         inst.setStateMachine(stateMachine);
63.         inst.setMachineId(stateMachine.getId());
64.         inst.setTenantId(tenantId);
65.         inst.setBusinessKey(businessKey);
66.         //设置状态机实例的启动参数
67.         inst.setStartParams(startParams);
68.         if (startParams != null) {
69.             if (StringUtils.hasText(businessKey)) {
70.                 startParams.put(DomainConstants.VAR_NAME_BUSINESSKEY, businessKey);
71.             }
72.             String parentId = (String)startParams.get(DomainConstants.VAR_NAME_PARENT_ID);
73.             if (StringUtils.hasText(parentId)) {
74.                 inst.setParentId(parentId);
75.                 startParams.remove(DomainConstants.VAR_NAME_PARENT_ID);
76.             }
77.         }
79.         inst.setStatus(ExecutionStatus.RU);
80.         inst.setRunning(true);
81.         inst.setGmtStarted(new Date());
82.         inst.setGmtUpdated(inst.getGmtStarted());
84.         return inst;
85.     }
86.     ...
87. }

复制代码

 
13.ProcessContextBuilder构建流程上下文
ProcessCtrlStateMachineEngine.startInternal()方法在执行过程中会创建一个流程上下文构造器ProcessContextBuilder实例，然后根据这个流程上下文构造器ProcessContextBuilder构建出一个流程上下文ProcessContext。

1. public class ProcessCtrlStateMachineEngine implements StateMachineEngine {
2.     ...
3.     //启动状态机实例StateMachineInstance
4.     //@param stateMachineName 状态机名称
5.     //@param tenantId 租户ID
6.     //@param businessKey 业务key
7.     //@param startParams 状态机实例的启动参数
8.     //@param async 是否异步化运行
9.     //@param callback 异步化运行时的回调接口
10.     private StateMachineInstance startInternal(String stateMachineName, String tenantId, String businessKey, Map<String, Object> startParams, boolean async, AsyncCallback callback) throws EngineExecutionException {
11.         try {
12.             //如果指定需要异步运行，但是状态机配置里是不允许异步运行的，则会抛异常
13.             if (async && !stateMachineConfig.isEnableAsync()) {
14.                 throw new EngineExecutionException("Asynchronous start is disabled. please set StateMachineConfig.enableAsync=true first.", FrameworkErrorCode.AsynchronousStartDisabled);
15.             }
17.             if (StringUtils.isEmpty(tenantId)) {
18.                 tenantId = stateMachineConfig.getDefaultTenantId();
19.             }
21.             //创建状态机实例StateMachineInstance
22.             StateMachineInstance instance = createMachineInstance(stateMachineName, tenantId, businessKey, startParams);
24.             //创建一个流程上下文构造器ProcessContextBuilder实例，用来构造流程运行时的上下文
25.             ProcessContextBuilder contextBuilder = ProcessContextBuilder.create()
26.                 .withProcessType(ProcessType.STATE_LANG)//设置流程类型
27.                 .withOperationName(DomainConstants.OPERATION_NAME_START)//设置操作名称
28.                 .withAsyncCallback(callback)//设置异步化时的回调接口
29.                 .withInstruction(new StateInstruction(stateMachineName, tenantId))//设置状态获取组件
30.                 .withStateMachineInstance(instance)//设置状态机实例
31.                 .withStateMachineConfig(getStateMachineConfig())//设置状态机配置
32.                 .withStateMachineEngine(this);//设置状态机引擎
34.             //上下文变量Map
35.             Map<String, Object> contextVariables;
36.             if (startParams != null) {
37.                 contextVariables = new ConcurrentHashMap<>(startParams.size());
38.                 nullSafeCopy(startParams, contextVariables);
39.             } else {
40.                 contextVariables = new ConcurrentHashMap<>();
41.             }
42.             instance.setContext(contextVariables);
44.             //设置流程上下文构造器ProcessContextBuilder实例
45.             contextBuilder.withStateMachineContextVariables(contextVariables);
46.             contextBuilder.withIsAsyncExecution(async);
47.             //通过流程上下文构造器ProcessContextBuilder构建出一个流程上下文ProcessContext
48.             ProcessContext processContext = contextBuilder.build();
50.             //如果状态机定义StateMachine是支持持久化的 且 状态日志的存储组件不为null
51.             if (instance.getStateMachine().isPersist() && stateMachineConfig.getStateLogStore() != null) {
52.                 //通过状态机实例日志和状态实例日志的存储组件StateLogStore，记录状态机实例StateMachineInstance的启动事件日志 + 开启全局事务
53.                 //比如在DB中更新状态机实例StateMachineInstance的启动状态
54.                 stateMachineConfig.getStateLogStore().recordStateMachineStarted(instance, processContext);
55.             }
56.             if (StringUtils.isEmpty(instance.getId())) {
57.                 //生成状态机实例StateMachineInstance的序号
58.                 instance.setId(stateMachineConfig.getSeqGenerator().generate(DomainConstants.SEQ_ENTITY_STATE_MACHINE_INST));
59.             }
61.             StateInstruction stateInstruction = processContext.getInstruction(StateInstruction.class);
62.             Loop loop = LoopTaskUtils.getLoopConfig(processContext, stateInstruction.getState(processContext));
63.             if (null != loop) {
64.                 stateInstruction.setTemporaryState(new LoopStartStateImpl());
65.             }
67.             if (async) {
68.                 stateMachineConfig.getAsyncProcessCtrlEventPublisher().publish(processContext);
69.             } else {
70.                 stateMachineConfig.getProcessCtrlEventPublisher().publish(processContext);
71.             }
72.             return instance;
73.         } finally {
74.             if (stateMachineConfig.getStateLogStore() != null) {
75.                 stateMachineConfig.getStateLogStore().clearUp();
76.             }
77.         }
78.     }
79.     ...
80. }

复制代码

 
14.StateLogStore记录日志和开启Saga全局事务
在状态机引擎ProcessCtrlStateMachineEngine的startInternal()方法中，会通过调用StateLogStore的recordStateMachineStarted()方法，记录状态机实例StateMachineInstance的启动事件到DB，以及开启Saga全局事务。

1. public class DbAndReportTcStateLogStore extends AbstractStore implements StateLogStore {
2.     private static final Logger LOGGER = LoggerFactory.getLogger(DbAndReportTcStateLogStore.class);
3.    
4.                 //插入状态机实例到数据库的SQL语句
5.     private static final StateMachineInstanceToStatementForInsert STATE_MACHINE_INSTANCE_TO_STATEMENT_FOR_INSERT = new StateMachineInstanceToStatementForInsert();
6.     //更新数据库状态机实例的SQL语句
7.     private static final StateMachineInstanceToStatementForUpdate STATE_MACHINE_INSTANCE_TO_STATEMENT_FOR_UPDATE = new StateMachineInstanceToStatementForUpdate();
8.     //查询数据库中的状态机实例的SQL语句
9.     private static final ResultSetToStateMachineInstance RESULT_SET_TO_STATE_MACHINE_INSTANCE = new ResultSetToStateMachineInstance();
10.     //插入状态实例到数据库的SQL语句
11.     private static final StateInstanceToStatementForInsert STATE_INSTANCE_TO_STATEMENT_FOR_INSERT = new StateInstanceToStatementForInsert();
12.     //更新数据库中的状态实例的SQL语句
13.     private static final StateInstanceToStatementForUpdate STATE_INSTANCE_TO_STATEMENT_FOR_UPDATE = new StateInstanceToStatementForUpdate();
14.     //查询数据库中的状态实例的SQL语句
15.     private static final ResultSetToStateInstance RESULT_SET_TO_STATE_INSTANCE = new ResultSetToStateInstance();
17.     //Saga全局事务模版
18.     private SagaTransactionalTemplate sagaTransactionalTemplate;
19.     //参数序列化组件
20.     private Serializer<Object, String> paramsSerializer = new ParamsSerializer();
21.     //异常序列化组件
22.     private Serializer<Exception, byte[]> exceptionSerializer = new ExceptionSerializer();
23.     //状态日志的存储SQL语句
24.     private StateLogStoreSqls stateLogStoreSqls;
25.     //默认的租户ID
26.     private String defaultTenantId;
27.     //序号生成器
28.     private SeqGenerator seqGenerator;
30.     //记录状态机实例的启动事件日志到DB + 开启Saga全局事务
31.     @Override
32.     public void recordStateMachineStarted(StateMachineInstance machineInstance, ProcessContext context) {
33.         if (machineInstance != null) {
34.             //if parentId is not null, machineInstance is a SubStateMachine, do not start a new global transaction, use parent transaction instead.
35.             //如果parentId是空的，那么当前事务就是Saga分布式事务的启动入口；
36.             //如果parentId不是空的，则已经有服务使用状态机启动了Saga分布式事务；
37.             String parentId = machineInstance.getParentId();
38.             if (StringUtils.isEmpty(parentId)) {
39.                 //1.开启Saga全局事务
40.                 beginTransaction(machineInstance, context);
41.             }
43.             try {
44.                 if (StringUtils.isEmpty(machineInstance.getId()) && seqGenerator != null) {
45.                     machineInstance.setId(seqGenerator.generate(DomainConstants.SEQ_ENTITY_STATE_MACHINE_INST));
46.                 }
47.                 //bind SAGA branch type
48.                 RootContext.bindBranchType(BranchType.SAGA);
49.                 //save to db
50.                 machineInstance.setSerializedStartParams(paramsSerializer.serialize(machineInstance.getStartParams()));
51.                 //2.记录日志到DB
52.                 int effect = executeUpdate(stateLogStoreSqls.getRecordStateMachineStartedSql(dbType), STATE_MACHINE_INSTANCE_TO_STATEMENT_FOR_INSERT, machineInstance);
53.                 if (effect < 1) {
54.                     throw new StoreException("StateMachineInstance record start error, Xid: " + machineInstance.getId(), FrameworkErrorCode.OperationDenied);
55.                 }
56.             } catch (StoreException e) {
57.                 LOGGER.error("Record statemachine start error: {}, StateMachine: {}, XID: {}, Reason: {}", e.getErrcode(), machineInstance.getStateMachine().getName(), machineInstance.getId(), e.getMessage(), e);
58.                 this.clearUp();
59.                 throw e;
60.             }
61.         }
62.     }
64.     protected void beginTransaction(StateMachineInstance machineInstance, ProcessContext context) {
65.         if (sagaTransactionalTemplate != null) {
66.             //获取状态机配置
67.             StateMachineConfig stateMachineConfig = (StateMachineConfig) context.getVariable(DomainConstants.VAR_NAME_STATEMACHINE_CONFIG);
68.             //构建事务信息
69.             TransactionInfo transactionInfo = new TransactionInfo();
70.             transactionInfo.setTimeOut(stateMachineConfig.getTransOperationTimeout());
71.             transactionInfo.setName(Constants.SAGA_TRANS_NAME_PREFIX + machineInstance.getStateMachine().getName());
72.             try {
73.                 //通过Saga事务模版开启全局事务
74.                 GlobalTransaction globalTransaction = sagaTransactionalTemplate.beginTransaction(transactionInfo);
75.                 machineInstance.setId(globalTransaction.getXid());
77.                 context.setVariable(DomainConstants.VAR_NAME_GLOBAL_TX, globalTransaction);
78.                 Map<String, Object> machineContext = machineInstance.getContext();
79.                 if (machineContext != null) {
80.                     machineContext.put(DomainConstants.VAR_NAME_GLOBAL_TX, globalTransaction);
81.                 }
82.             } catch (ExecutionException e) {
83.                 String xid = null;
84.                 if (e.getTransaction() != null) {
85.                     xid = e.getTransaction().getXid();
86.                 }
87.                 throw new EngineExecutionException(e, e.getCode() + ", TransName:" + transactionInfo.getName() + ", XID: " + xid + ", Reason: " + e.getMessage(), FrameworkErrorCode.TransactionManagerError);
88.             } finally {
89.                 if (Boolean.TRUE.equals(context.getVariable(DomainConstants.VAR_NAME_IS_ASYNC_EXECUTION))) {
90.                     RootContext.unbind();
91.                     RootContext.unbindBranchType();
92.                 }
93.             }
94.         }
95.     }
96.     ...
97. }
99. public class StateLogStoreSqls {
100.     private String recordStateMachineStartedSql;
101.     private static final String RECORD_STATE_MACHINE_STARTED_SQL = "INSERT INTO ${TABLE_PREFIX}state_machine_inst\n"
102.         + "(id, machine_id, tenant_id, parent_id, gmt_started, business_key, start_params, is_running, status, "
103.         + "gmt_updated)\n"
104.         + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
105.     ...
106.    
107.     public StateLogStoreSqls(String tablePrefix) {
108.         this.tablePrefix = tablePrefix;
109.         init();
110.     }
111.    
112.     private void init() {
113.         recordStateMachineStartedSql = RECORD_STATE_MACHINE_STARTED_SQL.replaceAll(TABLE_PREFIX_REGEX, tablePrefix);
114.         ...
115.     }
116.    
117.     public String getRecordStateMachineStartedSql(String dbType) {
118.         return recordStateMachineStartedSql;
119.     }
120.     ...
121. }
123. public class AbstractStore {
124.     protected DataSource dataSource;
125.     ...
126.    
127.     protected <T> int executeUpdate(String sql, ObjectToStatement<T> objectToStatement, T o) {
128.         Connection connection = null;
129.         PreparedStatement stmt = null;
130.         try {
131.             connection = dataSource.getConnection();
132.             if (LOGGER.isDebugEnabled()) {
133.                 LOGGER.debug("Preparing SQL: {}", sql);
134.             }
135.             stmt = connection.prepareStatement(sql);
136.             if (LOGGER.isDebugEnabled()) {
137.                 LOGGER.debug("setting params to PreparedStatement: {}", BeanUtils.beanToString(o));
138.             }
139.             objectToStatement.toStatement(o, stmt);
140.             int count = stmt.executeUpdate();
141.             if (!connection.getAutoCommit()) {
142.                 connection.commit();
143.             }
144.             return count;
145.         } catch (SQLException e) {
146.             throw new StoreException(e);
147.         } finally {
148.             closeSilent(stmt);
149.             closeSilent(connection);
150.         }
151.     }
152.     ...
153. }
155. public class DefaultSagaTransactionalTemplate implements SagaTransactionalTemplate, ApplicationContextAware, DisposableBean, InitializingBean {
156.     ...
157.     @Override
158.     public GlobalTransaction beginTransaction(TransactionInfo txInfo) throws TransactionalExecutor.ExecutionException {
159.         GlobalTransaction tx = GlobalTransactionContext.getCurrentOrCreate();
160.         try {
161.             triggerBeforeBegin();
162.             //开启一个全局会话
163.             tx.begin(txInfo.getTimeOut(), txInfo.getName());
164.             triggerAfterBegin();
165.         } catch (TransactionException txe) {
166.             throw new TransactionalExecutor.ExecutionException(tx, txe, TransactionalExecutor.Code.BeginFailure);
167.         }
168.         return tx;
169.     }
170.     ...
171. }

复制代码

执行GlobalTransaction的begin()方法开启一个全局会话：

1. //The type Default global transaction.
2. //默认的全局事务
3. public class DefaultGlobalTransaction implements GlobalTransaction {
4.     private TransactionManager transactionManager;
5.     private String xid;
6.     private GlobalStatus status;
7.     private GlobalTransactionRole role;
8.     ...
9.    
10.     DefaultGlobalTransaction() {
11.         //全局事务角色是全局事务发起者
12.         this(null, GlobalStatus.UnKnown, GlobalTransactionRole.Launcher);
13.     }
14.    
15.     //Instantiates a new Default global transaction.
16.     DefaultGlobalTransaction(String xid, GlobalStatus status, GlobalTransactionRole role) {
17.         this.transactionManager = TransactionManagerHolder.get();//全局事务管理者
18.         this.xid = xid;
19.         this.status = status;
20.         this.role = role;
21.     }
22.     ...
23.    
24.     @Override
25.     public void begin(int timeout, String name) throws TransactionException {
26.         if (role != GlobalTransactionRole.Launcher) {
27.             assertXIDNotNull();
28.             if (LOGGER.isDebugEnabled()) {
29.                 LOGGER.debug("Ignore Begin(): just involved in global transaction [{}]", xid);
30.             }
31.             return;
32.         }
33.         assertXIDNull();
34.         String currentXid = RootContext.getXID();
35.         if (currentXid != null) {
36.             throw new IllegalStateException("Global transaction already exists," + " can't begin a new global transaction, currentXid = " + currentXid);
37.         }
38.         //通过全局事务管理器去真正开启全局事务，一旦开启成功，就可以获取到一个xid
39.         xid = transactionManager.begin(null, null, name, timeout);
40.         status = GlobalStatus.Begin;
41.         //把xid绑定到RootContext的线程本地变量副本里去
42.         RootContext.bind(xid);
43.         if (LOGGER.isInfoEnabled()) {
44.             LOGGER.info("Begin new global transaction [{}]", xid);
45.         }
46.     }
47.     ...
48. }
50. public class TransactionManagerHolder {
51.     private static final Logger LOGGER = LoggerFactory.getLogger(TransactionManagerHolder.class);
52.    
53.     private static class SingletonHolder {
54.         private static TransactionManager INSTANCE = null;
55.         static {
56.             try {
57.                 INSTANCE = EnhancedServiceLoader.load(TransactionManager.class);
58.                 LOGGER.info("TransactionManager Singleton {}", INSTANCE);
59.             } catch (Throwable anyEx) {
60.                 LOGGER.error("Failed to load TransactionManager Singleton! ", anyEx);
61.             }
62.         }
63.     }
64.    
65.     //Get transaction manager.
66.     public static TransactionManager get() {
67.         if (SingletonHolder.INSTANCE == null) {
68.             throw new ShouldNeverHappenException("TransactionManager is NOT ready!");
69.         }
70.         return SingletonHolder.INSTANCE;
71.     }
72.    
73.     private TransactionManagerHolder() {
74.     }
75. }
77. //The type Default transaction manager.
78. //默认的全局事务管理器
79. public class DefaultTransactionManager implements TransactionManager {
80.     @Override
81.     public String begin(String applicationId, String transactionServiceGroup, String name, int timeout) throws TransactionException {
82.         //构建一个全局事务开启请求GlobalBeginRequest
83.         GlobalBeginRequest request = new GlobalBeginRequest();
84.         request.setTransactionName(name);
85.         request.setTimeout(timeout);
86.         //发起一个同步调用
87.         GlobalBeginResponse response = (GlobalBeginResponse) syncCall(request);
88.         if (response.getResultCode() == ResultCode.Failed) {
89.             throw new TmTransactionException(TransactionExceptionCode.BeginFailed, response.getMsg());
90.         }
91.         return response.getXid();
92.     }
93.     ...
94.    
95.     private AbstractTransactionResponse syncCall(AbstractTransactionRequest request) throws TransactionException {
96.         try {
97.             //TMNettyRemotingClient会和Seata Server基于Netty建立长连接
98.             return (AbstractTransactionResponse) TmNettyRemotingClient.getInstance().sendSyncRequest(request);
99.         } catch (TimeoutException toe) {
100.             throw new TmTransactionException(TransactionExceptionCode.IO, "RPC timeout", toe);
101.         }
102.     }
103. }
105. public abstract class AbstractNettyRemotingClient extends AbstractNettyRemoting implements RemotingClient {
106.     ...
107.     @Override
108.     public Object sendSyncRequest(Object msg) throws TimeoutException {
109.         //因为Seata Server是可以多节点部署实现高可用架构的，所以这里调用loadBalance()方法进行负载均衡
110.         String serverAddress = loadBalance(getTransactionServiceGroup(), msg);
111.         //获取RPC调用的超时时间
112.         long timeoutMillis = this.getRpcRequestTimeout();
113.         //构建一个RPC消息
114.         RpcMessage rpcMessage = buildRequestMessage(msg, ProtocolConstants.MSGTYPE_RESQUEST_SYNC);
116.         //send batch message
117.         //put message into basketMap, @see MergedSendRunnable
118.         //默认是不开启批量消息发送
119.         if (this.isEnableClientBatchSendRequest()) {
120.             ...
121.         } else {
122.             //通过网络连接管理器clientChannelManager，获取与指定Seata Server建立的网络连接Channel
123.             //然后通过网络连接Channel把RpcMessage发送出去
124.             Channel channel = clientChannelManager.acquireChannel(serverAddress);
125.             return super.sendSync(channel, rpcMessage, timeoutMillis);
126.         }
127.     }
128.     ...
129. }
131. public abstract class AbstractNettyRemoting implements Disposable {
132.     ...
133.     protected Object sendSync(Channel channel, RpcMessage rpcMessage, long timeoutMillis) throws TimeoutException {
134.         if (timeoutMillis <= 0) {
135.             throw new FrameworkException("timeout should more than 0ms");
136.         }
137.         if (channel == null) {
138.             LOGGER.warn("sendSync nothing, caused by null channel.");
139.             return null;
140.         }
142.         //把发送出去的请求封装到MessageFuture中，然后存放到futures这个Map里
143.         MessageFuture messageFuture = new MessageFuture();
144.         messageFuture.setRequestMessage(rpcMessage);
145.         messageFuture.setTimeout(timeoutMillis);
146.         futures.put(rpcMessage.getId(), messageFuture);
148.         channelWritableCheck(channel, rpcMessage.getBody());
150.         //获取远程地址
151.         String remoteAddr = ChannelUtil.getAddressFromChannel(channel);
152.         doBeforeRpcHooks(remoteAddr, rpcMessage);
154.         //异步化发送数据，同时对发送结果添加监听器
155.         //如果发送失败，则会对网络连接Channel进行销毁处理
156.         channel.writeAndFlush(rpcMessage).addListener((ChannelFutureListener) future -> {
157.             if (!future.isSuccess()) {
158.                 MessageFuture messageFuture1 = futures.remove(rpcMessage.getId());
159.                 if (messageFuture1 != null) {
160.                     messageFuture1.setResultMessage(future.cause());
161.                 }
162.                 destroyChannel(future.channel());
163.             }
164.         });
166.         try {
167.             //然后通过请求响应组件MessageFuture同步等待Seata Server返回该请求的响应
168.             Object result = messageFuture.get(timeoutMillis, TimeUnit.MILLISECONDS);
169.             doAfterRpcHooks(remoteAddr, rpcMessage, result);
170.             return result;
171.         } catch (Exception exx) {
172.             LOGGER.error("wait response error:{},ip:{},request:{}", exx.getMessage(), channel.remoteAddress(), rpcMessage.getBody());
173.             if (exx instanceof TimeoutException) {
174.                 throw (TimeoutException) exx;
175.             } else {
176.                 throw new RuntimeException(exx);
177.             }
178.         }
179.     }
180.     ...
181. }

复制代码

 
15.状态操作组件获取状态和State状态类继承体系
(1)状态操作组件StateInstruction获取状态
(2)State状态类的继承体系
 
(1)状态操作组件StateInstruction获取状态
在状态机引擎ProcessCtrlStateMachineEngine的startInternal()方法中，会通过流程上下文ProcessContext来获取状态操作组件StateInstruction。
 
根据状态操作组件StateInstruction，可以获取状态机的开始State(状态)、也可以获取指定的某个State(状态)。

1. public class ProcessCtrlStateMachineEngine implements StateMachineEngine {
2.     ...
3.     //启动状态机实例StateMachineInstance
4.     //@param stateMachineName 状态机名称
5.     //@param tenantId 租户ID
6.     //@param businessKey 业务key
7.     //@param startParams 状态机实例的启动参数
8.     //@param async 是否异步化运行
9.     //@param callback 异步化运行时的回调接口
10.     private StateMachineInstance startInternal(String stateMachineName, String tenantId, String businessKey, Map<String, Object> startParams, boolean async, AsyncCallback callback) throws EngineExecutionException {
11.         try {
12.             //如果指定需要异步运行，但是状态机配置里是不允许异步运行的，则会抛异常
13.             if (async && !stateMachineConfig.isEnableAsync()) {
14.                 throw new EngineExecutionException("Asynchronous start is disabled. please set StateMachineConfig.enableAsync=true first.", FrameworkErrorCode.AsynchronousStartDisabled);
15.             }
17.             if (StringUtils.isEmpty(tenantId)) {
18.                 tenantId = stateMachineConfig.getDefaultTenantId();
19.             }
21.             //创建状态机实例StateMachineInstance
22.             StateMachineInstance instance = createMachineInstance(stateMachineName, tenantId, businessKey, startParams);
24.             //创建一个流程上下文构造器ProcessContextBuilder实例，用来构造流程运行时的上下文
25.             ProcessContextBuilder contextBuilder = ProcessContextBuilder.create()
26.                 .withProcessType(ProcessType.STATE_LANG)//设置流程类型
27.                 .withOperationName(DomainConstants.OPERATION_NAME_START)//设置操作名称
28.                 .withAsyncCallback(callback)//设置异步化时的回调接口
29.                 .withInstruction(new StateInstruction(stateMachineName, tenantId))//设置状态获取组件
30.                 .withStateMachineInstance(instance)//设置状态机实例
31.                 .withStateMachineConfig(getStateMachineConfig())//设置状态机配置
32.                 .withStateMachineEngine(this);//设置状态机引擎
34.             //上下文变量Map
35.             Map<String, Object> contextVariables;
36.             if (startParams != null) {
37.                 contextVariables = new ConcurrentHashMap<>(startParams.size());
38.                 nullSafeCopy(startParams, contextVariables);
39.             } else {
40.                 contextVariables = new ConcurrentHashMap<>();
41.             }
42.             instance.setContext(contextVariables);
44.             //设置流程上下文构造器ProcessContextBuilder实例
45.             contextBuilder.withStateMachineContextVariables(contextVariables);
46.             contextBuilder.withIsAsyncExecution(async);
47.             //通过流程上下文构造器ProcessContextBuilder构建出一个流程上下文ProcessContext
48.             ProcessContext processContext = contextBuilder.build();
50.             //如果状态机定义StateMachine是支持持久化的 且 状态日志的存储组件不为null
51.             if (instance.getStateMachine().isPersist() && stateMachineConfig.getStateLogStore() != null) {
52.                 //通过状态机实例日志和状态实例日志的存储组件StateLogStore，记录状态机实例StateMachineInstance的启动事件日志 + 开启全局事务
53.                 //比如在DB中更新状态机实例StateMachineInstance的启动状态
54.                 stateMachineConfig.getStateLogStore().recordStateMachineStarted(instance, processContext);
55.             }
56.             if (StringUtils.isEmpty(instance.getId())) {
57.                 //生成状态机实例StateMachineInstance的序号
58.                 instance.setId(stateMachineConfig.getSeqGenerator().generate(DomainConstants.SEQ_ENTITY_STATE_MACHINE_INST));
59.             }
60.          
61.             //从流程上下文中获取状态操作组件StateInstruction
62.             StateInstruction stateInstruction = processContext.getInstruction(StateInstruction.class);
63.             //获取循环策略
64.             Loop loop = LoopTaskUtils.getLoopConfig(processContext, stateInstruction.getState(processContext));
65.             if (null != loop) {
66.                 stateInstruction.setTemporaryState(new LoopStartStateImpl());
67.             }
69.             if (async) {
70.                 stateMachineConfig.getAsyncProcessCtrlEventPublisher().publish(processContext);
71.             } else {
72.                 stateMachineConfig.getProcessCtrlEventPublisher().publish(processContext);
73.             }
74.             return instance;
75.         } finally {
76.             if (stateMachineConfig.getStateLogStore() != null) {
77.                 stateMachineConfig.getStateLogStore().clearUp();
78.             }
79.         }
80.     }
81.     ...
82. }
84. //State Instruction，状态操作组件StateInstruction
85. public class StateInstruction implements Instruction {
86.     private String stateName;
87.     private String stateMachineName;
88.     private String tenantId;
89.     private boolean end;
90.     //Temporary state node for running temporary nodes in the state machine
91.     private State temporaryState;
92.    
93.     public StateInstruction() {
94.     }
96.     public StateInstruction(String stateMachineName, String tenantId) {
97.         this.stateMachineName = stateMachineName;
98.         this.tenantId = tenantId;
99.     }
101.     //根据流程上下文获取状态，可以是状态机的开始State(状态)、也可以是指定的某个State(状态)
102.     public State getState(ProcessContext context) {
103.         if (getTemporaryState() != null) {
104.             return temporaryState;
105.         }
107.         String stateName = getStateName();
108.         String stateMachineName = getStateMachineName();
109.         String tenantId = getTenantId();
111.         if (StringUtils.isEmpty(stateMachineName)) {
112.             throw new EngineExecutionException("StateMachineName is required", FrameworkErrorCode.ParameterRequired);
113.         }
115.         StateMachineConfig stateMachineConfig = (StateMachineConfig)context.getVariable(DomainConstants.VAR_NAME_STATEMACHINE_CONFIG);
116.         StateMachine stateMachine = stateMachineConfig.getStateMachineRepository().getStateMachine(stateMachineName, tenantId);
117.         if (stateMachine == null) {
118.             throw new EngineExecutionException("StateMachine[" + stateMachineName + "] is not exist", FrameworkErrorCode.ObjectNotExists);
119.         }
121.         //默认就是获取状态机定义StateMachine中的第一个状态State
122.         if (StringUtils.isEmpty(stateName)) {
123.             stateName = stateMachine.getStartState();
124.             setStateName(stateName);
125.         }
127.         //根据状态名字stateName，从状态机定义中获取对应的状态State
128.         State state = stateMachine.getStates().get(stateName);
129.         if (state == null) {
130.             throw new EngineExecutionException("State[" + stateName + "] is not exist", FrameworkErrorCode.ObjectNotExists);
131.         }
133.         return state;
134.     }
135.     ...
136. }

复制代码

(2)State状态类的继承体系

[code]//A State in StateMachinepublic interface State {    String getName();    String getComment();    String getType();    String getNext();    Map getExtensions();    StateMachine getStateMachine();}public abstract class BaseState implements State {    private transient String name;//状态名称    private String type;//状态类型    private String comment;//状态备注    private String next;//下一个状态名称    private Map extensions;//状态扩展属性    private transient StateMachine stateMachine;//状态所属的状态机    ...}//The state of the execution task (abstract class), the specific task to be executed is determined by the subclasspublic abstract class AbstractTaskState extends BaseState implements TaskState {    private String compensateState;//补偿状态    private boolean isForCompensation;//是否用于补偿    private boolean isForUpdate;//是否用于更新    private List retry;//重试列表    private List catches;//异常匹配列表    private List input;//状态输入列表    private Map output;//状态输出数据    private Map status;//Map    private List inputExpressions;//输入表达式列表    private Map outputExpressions;//输出表达式数据    private boolean isPersist = true;//是否持久化    private Boolean retryPersistModeUpdate;//是否更新Saga重试持久化的模式    private Boolean compensatePersistModeUpdate;//是否更新Saga补偿持久化的模式    private Loop loop;//循环对象    ...        public static class RetryImpl implements Retry {        private List exceptions;//重试的过程中遇到了哪些异常        private List