1. saga配置與使用
首先看下saga模式的配置,代碼地址：
<bean id="dataSource" class="org.h2.jdbcx.JdbcConnectionPool" destroy-method="dispose">
<constructor-arg>
<bean class="org.h2.jdbcx.JdbcDataSource">
<property name="URL" value="jdbc:h2:mem:seata_saga" />
<property name="user" value="sa" />
<property name="password" value="sa" />
</bean>
</constructor-arg>
</bean>
<jdbc:initialize-database data-source="dataSource">
<jdbc:script location="classpath:sql/h2_init.sql" />
</jdbc:initialize-database>
<bean id="stateMachineEngine" class="io.seata.saga.engine.impl.ProcessCtrlStateMachineEngine">
<property name="stateMachineConfig" ref="dbStateMachineConfig"></property>
</bean>
<bean id="dbStateMachineConfig" class="io.seata.saga.engine.config.DbStateMachineConfig">
<property name="dataSource" ref="dataSource"></property>
<property name="resources" value="statelang/*.json"></property>
<property name="enableAsync" value="true"></property>
<property name="threadPoolExecutor" ref="threadExecutor"></property>
<property name="applicationId" value="saga_sample"></property>
<property name="txServiceGroup" value="my_test_tx_group"></property>
</bean>
<bean id="threadExecutor"
class="org.springframework.scheduling.concurrent.ThreadPoolExecutorFactoryBean">
<property name="threadNamePrefix" value="SAGA_ASYNC_EXE_" />
<property name="corePoolSize" value="1" />
<property name="maxPoolSize" value="20" />
</bean>
<bean class="io.seata.saga.rm.StateMachineEngineHolder">
<property name="stateMachineEngine" ref="stateMachineEngine"/>
</bean>
在以上配置中的DbStateMachineConfig bean中的DefaultStateMachineConfig#afterPropertiesSet方法中會調用init方法來進行初始化狀態(tài)機實例枚驻，主要邏輯如下：
- 初始化ExpressionFactoryManager恬口，用于表達式解析寺晌，如spring Expression Language
- 初始化EvaluatorFactoryManager瑟曲，用于執(zhí)行表達式的驗證
- 初始化StateMachineRepositoryImpl，用于狀態(tài)機信息的維護闲孤，如狀態(tài)機的初始化谆级，存儲等
- 初始化本地狀態(tài)機數據庫操作服務類，StateLangStore
- 加載所有狀態(tài)機的配置StateMachineRepositoryImpl#registryByResources讼积，首先從狀態(tài)機組裝json獲取所有狀態(tài)機配置肥照，同時對比數據庫中的狀態(tài)機存儲的配置，并更新相關數據勤众，如更新時間等
- 初始化ProcessCtrlEventPublisher舆绎，用于狀態(tài)機事件的同步提交，使用的是DirectEventBus來同步處理請求
- 初始化ProcessCtrlEventPublisher决摧，用于狀態(tài)機事件的異步提交亿蒸，使用的AsyncEventBus來處理請求凑兰，AsyncEventBus主要邏輯通過線程池來異步處理
在上面流程registryByResources方法中掌桩，會從json配置中讀取所有狀態(tài)機流程，同時寫入或更新到表seata_state_machine_def中姑食，seata_state_machine_def中content字段表示json配置所有數據
然后我們看下事件處理流程波岛，首先觸發(fā)一次調用
//獲取xml配置中添加的bean，ProcessCtrlStateMachineEngine
StateMachineEngine stateMachineEngine = (StateMachineEngine) applicationContext.getBean("stateMachineEngine");
Map<String, Object> startParams = new HashMap<>(3);
String businessKey = String.valueOf(System.currentTimeMillis());
startParams.put("businessKey", businessKey);
startParams.put("count", 10);
startParams.put("amount", new BigDecimal("100"));
//sync test
StateMachineInstance inst = stateMachineEngine.startWithBusinessKey("reduceInventoryAndBalance", null, businessKey, startParams);
2. saga流程分析
stateMachineEngine.startWithBusinessKey方法會調用到ProcessCtrlStateMachineEngine#startInternal,方法主要有三個邏輯：
- 首先創(chuàng)建一個狀態(tài)機實例StateMachineInstance音半，下文中會講解狀態(tài)機類StateMachine则拷，這個大家可以理解為模板類
- 假如這個狀態(tài)機實例需要持久化，則將狀態(tài)機實例保存到state_machine_inst實例表中曹鸠，假如當前狀態(tài)機實例不是子狀態(tài)機煌茬，還會調用beginTransaction，向TC開啟全局事務
- 將當前請求組裝到context中彻桃，推送給狀態(tài)機進行處理
StateMachineInstance instance = createMachineInstance(stateMachineName, tenantId, businessKey, startParams);
if (instance.getStateMachine().isPersist() && stateMachineConfig.getStateLogStore() != null) {
stateMachineConfig.getStateLogStore().recordStateMachineStarted(instance, processContext);
}
if (async) {
stateMachineConfig.getAsyncProcessCtrlEventPublisher().publish(processContext);
} else {
stateMachineConfig.getProcessCtrlEventPublisher().publish(processContext);
}
return instance;
在createMachineInstance方法中的主要邏輯如下：
//tenantId加入為空時坛善，會采用默認值
private StateMachineInstance createMachineInstance(String stateMachineName, String tenantId, String businessKey,
Map<String, Object> startParams) {
//從數據庫中獲取對應狀態(tài)機配置信息
StateMachine stateMachine = stateMachineConfig.getStateMachineRepository().getStateMachine(stateMachineName,
StateMachineInstanceImpl inst = new StateMachineInstanceImpl();
......
inst.setStateMachine(stateMachine);
......
return inst;
}
在DbAndReportTcStateLogStore#recordStateMachineStarted方法中
public void recordStateMachineStarted(StateMachineInstance machineInstance, ProcessContext context) {
if (machineInstance != null) {
//if parentId is not null, machineInstance is a SubStateMachine, do not start a new global transaction,
//use parent transaction instead.
String parentId = machineInstance.getParentId();
if (StringUtils.hasLength(parentId)) {
if (StringUtils.isEmpty(machineInstance.getId())) {
machineInstance.setId(parentId);
}
} else {
//向TC注冊全局事務
beginTransaction(machineInstance, context);
}
if (StringUtils.isEmpty(machineInstance.getId()) && seqGenerator != null) {
machineInstance.setId(seqGenerator.generate(DomainConstants.SEQ_ENTITY_STATE_MACHINE_INST));
}
// 將當前狀態(tài)機實例保存到state_machine_inst表中
machineInstance.setSerializedStartParams(paramsSerializer.serialize(machineInstance.getStartParams()));
executeUpdate(stateLogStoreSqls.getRecordStateMachineStartedSql(dbType),
STATE_MACHINE_INSTANCE_TO_STATEMENT_FOR_INSERT, machineInstance);
}
}
ProcessCtrlEventPublisher接收請求后，具體處理的類是DirectEventBus或AsyncEventBus邻眷，而這兩個類都是通過ProcessCtrlEventConsumer來進行處理請求的眠屎，
image
從上圖可以看出，請求最終會在ProcessControllerImpl類中進行處理肆饶，具體邏輯如下：
public void process(ProcessContext context) throws FrameworkException {
try {
//businessProcessor為CustomizeBusinessProcessor類
businessProcessor.process(context);
businessProcessor.route(context);
} catch (FrameworkException fex) {
throw fex;
} catch (Exception ex) {
LOGGER.error("Unknown exception occurred, context = {}", context, ex);
throw new FrameworkException(ex, "Unknown exception occurred", FrameworkErrorCode.UnknownAppError);
}
}
上述代碼其實seata saga模式的核心方法改衩，businessProcessor.process方法是使用狀態(tài)機中的某個state來進行處理，businessProcessor.route方法用于推動狀態(tài)的變更驯镊，即使用當前state的next state來繼續(xù)處理請求
接下來看下state的實際處理類StateMachineProcessHandler葫督，該類initDefaultHandlers方法會初始化stateHandlers竭鞍，增加不同狀態(tài)的處理器
- ChoiceStateHandler 分支state處理器
- SucceedEndStateHandler，成功結束的處理器
- CompensationTriggerStateHandler候衍，具體補償觸發(fā)處理器
- ServiceTaskStateHandler笼蛛，執(zhí)行服務調用處理的處理器
- FailEndStateHandler，失敗結束的處理器
在StateMachineProcessHandler#process處理方法中蛉鹿，主要有一下三個邏輯：
- 從入參ProcessContext中取出當前狀態(tài)的類型滨砍，并獲取實際的狀態(tài)處理器
StateHandler stateHandler = stateHandlers.get(stateType);
- 假如對應的StateHandler含有攔截器，則調用對應的攔截器
for (StateHandlerInterceptor interceptor : interceptors) {
executedInterceptors.add(interceptor);
interceptor.preProcess(context);
}
- 調用實際的狀態(tài)處理器進行處理
stateHandler.process(context);
狀態(tài)處理器處理完成后妖异，然后會執(zhí)行BusinessProcessor#route方法惋戏，該方法主要是根據StateMachineProcessHandler#process結果來切換當前請求的狀態(tài)，而route方法會進到DefaultRouterHandler#route中進行處理：
該方法有三個邏輯：
- 根絕當前狀態(tài)獲取路由處理器
ProcessRouter processRouter = processRouters.get(processType.getCode());
- 實際執(zhí)行路由處理器,返回當前請求下一個要執(zhí)行的state他膳，若沒有下一個返回空
Instruction instruction = processRouter.route(context);
- 假如存在下一個state响逢，則將請求再發(fā)送給狀態(tài)機接收類進行處理，這樣推動請求一直被處理棕孙，直到被遇到終止state
//ProcessCtrlEventPublisher
eventPublisher.publish(context);
上面是saga狀態(tài)機的處理與引擎的推動舔亭，下面主要介紹下saga模式下是如何與TC通信的？
上面流程都會提到攔截器蟀俊，在初始化StateHandler或 StateRouter 時會增加攔截器钦铺，重點介紹以下兩個
- ServiceTaskHandlerInterceptor
- 在進行state 處理前執(zhí)行DbAndReportTcStateLogStore#recordStateStarted方法，保存當前state狀態(tài)到表state_inst中肢预，同時向TC注冊分支事務DbAndReportTcStateLogStore#branchRegister
- 在state處理完成后矛洞，會執(zhí)行DbAndReportTcStateLogStore#recordStateFinished方法，更新表state_inst中state的狀態(tài)烫映，同時向TC發(fā)送report消息沼本，具體方法在SagaTransactionalTemplate#branchReport
- EndStateRouterInterceptor，在執(zhí)行EndStateRouter后锭沟，會執(zhí)行攔截器中postRoute方法抽兆，該方法會執(zhí)行EngineUtils.endStateMachine，在endStateMachine方法中會調用DbAndReportTcStateLogStore#recordStateMachineFinished來更新state_machine_inst表中狀態(tài)機實例狀態(tài)族淮，同時調用SagaTransactionalTemplate#reportTransaction來通知TC該事務已完成
3. 異常處理：
- 當狀態(tài)機模版不是一個完備的辫红，即有可能分支路由后不能到終止state，則會拋異常來終止
- 在執(zhí)行過程中遇到不能處理的異常瞧筛，則會拋異常來終止
4.TC消息處理：
- beginTransaction厉熟，發(fā)起全局事務
- branchRegister，進行分支注冊
- BranchReportRequest消息较幌，主要執(zhí)行coordinator.AbstractCore#branchReport方法揍瑟，修改表branch_table中對應分支事務的狀態(tài)
- SagaCore#doGlobalReport，在全局事務report時乍炉，如果是狀態(tài)是終止狀態(tài)后绢片，則會更新全局事務的狀態(tài)滤馍，如果不是終止狀態(tài)，則會更改事務狀態(tài)底循，來進行重試
