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.将业务处理器路由到状态机下一个要执行的状态
12.状态机引擎创建状态机实例的源码
基于流程控制的状态机引擎实例
ProcessCtrlStateMachineEngine的start()方法调用其startInternal()方法启动一个状态机实例时，会先创建一个状态机实例StateMachineInstance。
在调用
ProcessCtrlStateMachineEngine的createMachineInstance()方法创建一个状态机实例StateMachineInstance的过程中，会先通过状态机定义的仓储组件StateMachineRepository来获取一个状态机定义StateMachine，然后将状态机定义StateMachine注入到状态机实例对象中，以此来完成状态机实例对象的实例化。
//ProcessCtrl-based state machine engine
//基于流程控制的状态机引擎
public class ProcessCtrlStateMachineEngine implements StateMachineEngine {
//需要在seata-saga.xml配置文件中注入状态机配置实例
private StateMachineConfig stateMachineConfig;
...
@Override
public StateMachineInstance start(String stateMachineName, String tenantId, Map<String, Object> startParams) throws EngineExecutionException {
return startInternal(stateMachineName, tenantId, null, startParams, false, null);
}
@Override
public StateMachineInstance startAsync(String stateMachineName, String tenantId, Map<String, Object> startParams, AsyncCallback callback) throws EngineExecutionException {
return startInternal(stateMachineName, tenantId, null, startParams, true, callback);
}
@Override
public StateMachineInstance startWithBusinessKey(String stateMachineName, String tenantId, String businessKey, Map<String, Object> startParams) throws EngineExecutionException {
return startInternal(stateMachineName, tenantId, businessKey, startParams, false, null);
}
@Override
public StateMachineInstance startWithBusinessKeyAsync(String stateMachineName, String tenantId, String businessKey, Map<String, Object> startParams, AsyncCallback callback) throws EngineExecutionException {
return startInternal(stateMachineName, tenantId, businessKey, startParams, true, callback);
}
...
//启动状态机实例StateMachineInstance
//@param stateMachineName 状态机名称
//@param tenantId 租户ID
//@param businessKey 业务key
//@param startParams 状态机实例的启动参数
//@param async 是否异步化运行
//@param callback 异步化运行时的回调接口
private StateMachineInstance startInternal(String stateMachineName, String tenantId, String businessKey, Map<String, Object> startParams, boolean async, AsyncCallback callback) throws EngineExecutionException {
try {
//如果指定需要异步运行，但是状态机配置里是不允许异步运行的，则会抛异常
if (async && !stateMachineConfig.isEnableAsync()) {
throw new EngineExecutionException("Asynchronous start is disabled. please set StateMachineConfig.enableAsync=true first.", FrameworkErrorCode.AsynchronousStartDisabled);
}
if (StringUtils.isEmpty(tenantId)) {
tenantId = stateMachineConfig.getDefaultTenantId();
}
//创建状态机实例StateMachineInstance
StateMachineInstance instance = createMachineInstance(stateMachineName, tenantId, businessKey, startParams);
...
}
...
}
private StateMachineInstance createMachineInstance(String stateMachineName, String tenantId, String businessKey, Map<String, Object> startParams) {
//通过状态机定义的仓储组件StateMachineRepository，来获取一个状态机定义StateMachine
StateMachine stateMachine = stateMachineConfig.getStateMachineRepository().getStateMachine(stateMachineName, tenantId);
if (stateMachine == null) {
throw new EngineExecutionException("StateMachine[" + stateMachineName + "] is not exists", FrameworkErrorCode.ObjectNotExists);
}
//根据状态机定义StateMachine，实例化一个状态机实例对象StateMachineInstanceImpl
StateMachineInstanceImpl inst = new StateMachineInstanceImpl();
inst.setStateMachine(stateMachine);
inst.setMachineId(stateMachine.getId());
inst.setTenantId(tenantId);
inst.setBusinessKey(businessKey);
//设置状态机实例的启动参数
inst.setStartParams(startParams);
if (startParams != null) {
if (StringUtils.hasText(businessKey)) {
startParams.put(DomainConstants.VAR_NAME_BUSINESSKEY, businessKey);
}
String parentId = (String)startParams.get(DomainConstants.VAR_NAME_PARENT_ID);
if (StringUtils.hasText(parentId)) {
inst.setParentId(parentId);
startParams.remove(DomainConstants.VAR_NAME_PARENT_ID);
}
}
inst.setStatus(ExecutionStatus.RU);
inst.setRunning(true);
inst.setGmtStarted(new Date());
inst.setGmtUpdated(inst.getGmtStarted());
return inst;
}
...
}13.ProcessContextBuilder构建流程上下文
ProcessCtrlStateMachineEngine.startInternal()方法在执行过程中会创建一个流程上下文构造器ProcessContextBuilder实例，然后根据这个流程上下文构造器ProcessContextBuilder构建出一个流程上下文ProcessContext。
public class ProcessCtrlStateMachineEngine implements StateMachineEngine {
...
//启动状态机实例StateMachineInstance
//@param stateMachineName 状态机名称
//@param tenantId 租户ID
//@param businessKey 业务key
//@param startParams 状态机实例的启动参数
//@param async 是否异步化运行
//@param callback 异步化运行时的回调接口
private StateMachineInstance startInternal(String stateMachineName, String tenantId, String businessKey, Map<String, Object> startParams, boolean async, AsyncCallback callback) throws EngineExecutionException {
try {
//如果指定需要异步运行，但是状态机配置里是不允许异步运行的，则会抛异常
if (async && !stateMachineConfig.isEnableAsync()) {
throw new EngineExecutionException("Asynchronous start is disabled. please set StateMachineConfig.enableAsync=true first.", FrameworkErrorCode.AsynchronousStartDisabled);
}
if (StringUtils.isEmpty(tenantId)) {
tenantId = stateMachineConfig.getDefaultTenantId();
}
//创建状态机实例StateMachineInstance
StateMachineInstance instance = createMachineInstance(stateMachineName, tenantId, businessKey, startParams);
//创建一个流程上下文构造器ProcessContextBuilder实例，用来构造流程运行时的上下文
ProcessContextBuilder contextBuilder = ProcessContextBuilder.create()
.withProcessType(ProcessType.STATE_LANG)//设置流程类型
.withOperationName(DomainConstants.OPERATION_NAME_START)//设置操作名称
.withAsyncCallback(callback)//设置异步化时的回调接口
.withInstruction(new StateInstruction(stateMachineName, tenantId))//设置状态获取组件
.withStateMachineInstance(instance)//设置状态机实例
.withStateMachineConfig(getStateMachineConfig())//设置状态机配置
.withStateMachineEngine(this);//设置状态机引擎
//上下文变量Map
Map<String, Object> contextVariables;
if (startParams != null) {
contextVariables = new ConcurrentHashMap<>(startParams.size());
nullSafeCopy(startParams, contextVariables);
} else {
contextVariables = new ConcurrentHashMap<>();
}
instance.setContext(contextVariables);
//设置流程上下文构造器ProcessContextBuilder实例
contextBuilder.withStateMachineContextVariables(contextVariables);
contextBuilder.withIsAsyncExecution(async);
//通过流程上下文构造器ProcessContextBuilder构建出一个流程上下文ProcessContext
ProcessContext processContext = contextBuilder.build();
//如果状态机定义StateMachine是支持持久化的 且 状态日志的存储组件不为null
if (instance.getStateMachine().isPersist() && stateMachineConfig.getStateLogStore() != null) {
//通过状态机实例日志和状态实例日志的存储组件StateLogStore，记录状态机实例StateMachineInstance的启动事件日志 + 开启全局事务
//比如在DB中更新状态机实例StateMachineInstance的启动状态
stateMachineConfig.getStateLogStore().recordStateMachineStarted(instance, processContext);
}
if (StringUtils.isEmpty(instance.getId())) {
//生成状态机实例StateMachineInstance的序号
instance.setId(stateMachineConfig.getSeqGenerator().generate(DomainConstants.SEQ_ENTITY_STATE_MACHINE_INST));
}
StateInstruction stateInstruction = processContext.getInstruction(StateInstruction.class);
Loop loop = LoopTaskUtils.getLoopConfig(processContext, stateInstruction.getState(processContext));
if (null != loop) {
stateInstruction.setTemporaryState(new LoopStartStateImpl());
}
if (async) {
stateMachineConfig.getAsyncProcessCtrlEventPublisher().publish(processContext);
} else {
stateMachineConfig.getProcessCtrlEventPublisher().publish(processContext);
}
return instance;
} finally {
if (stateMachineConfig.getStateLogStore() != null) {
stateMachineConfig.getStateLogStore().clearUp();
}
}
}
...
}14.StateLogStore记录日志和开启Saga全局事务
在状态机引擎
ProcessCtrlStateMachineEngine的startInternal()方法中，会通过调用StateLogStore的recordStateMachineStarted()方法，记录状态机实例StateMachineInstance的启动事件到DB，以及开启Saga全局事务。
public class DbAndReportTcStateLogStore extends AbstractStore implements StateLogStore {
private static final Logger LOGGER = LoggerFactory.getLogger(DbAndReportTcStateLogStore.class);
//插入状态机实例到数据库的SQL语句
private static final StateMachineInstanceToStatementForInsert STATE_MACHINE_INSTANCE_TO_STATEMENT_FOR_INSERT = new StateMachineInstanceToStatementForInsert();
//更新数据库状态机实例的SQL语句
private static final StateMachineInstanceToStatementForUpdate STATE_MACHINE_INSTANCE_TO_STATEMENT_FOR_UPDATE = new StateMachineInstanceToStatementForUpdate();
//查询数据库中的状态机实例的SQL语句
private static final ResultSetToStateMachineInstance RESULT_SET_TO_STATE_MACHINE_INSTANCE = new ResultSetToStateMachineInstance();
//插入状态实例到数据库的SQL语句
private static final StateInstanceToStatementForInsert STATE_INSTANCE_TO_STATEMENT_FOR_INSERT = new StateInstanceToStatementForInsert();
//更新数据库中的状态实例的SQL语句
private static final StateInstanceToStatementForUpdate STATE_INSTANCE_TO_STATEMENT_FOR_UPDATE = new StateInstanceToStatementForUpdate();
//查询数据库中的状态实例的SQL语句
private static final ResultSetToStateInstance RESULT_SET_TO_STATE_INSTANCE = new ResultSetToStateInstance();
//Saga全局事务模版
private SagaTransactionalTemplate sagaTransactionalTemplate;
//参数序列化组件
private Serializer<Object, String> paramsSerializer = new ParamsSerializer();
//异常序列化组件
private Serializer<Exception, byte[]> exceptionSerializer = new ExceptionSerializer();
//状态日志的存储SQL语句
private StateLogStoreSqls stateLogStoreSqls;
//默认的租户ID
private String defaultTenantId;
//序号生成器
private SeqGenerator seqGenerator;
//记录状态机实例的启动事件日志到DB + 开启Saga全局事务
@Override
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.
//如果parentId是空的，那么当前事务就是Saga分布式事务的启动入口；
//如果parentId不是空的，则已经有服务使用状态机启动了Saga分布式事务；
String parentId = machineInstance.getParentId();
if (StringUtils.isEmpty(parentId)) {
//1.开启Saga全局事务
beginTransaction(machineInstance, context);
}
try {
if (StringUtils.isEmpty(machineInstance.getId()) && seqGenerator != null) {
machineInstance.setId(seqGenerator.generate(DomainConstants.SEQ_ENTITY_STATE_MACHINE_INST));
}
//bind SAGA branch type
RootContext.bindBranchType(BranchType.SAGA);
//save to db
machineInstance.setSerializedStartParams(paramsSerializer.serialize(machineInstance.getStartParams()));
//2.记录日志到DB
int effect = executeUpdate(stateLogStoreSqls.getRecordStateMachineStartedSql(dbType), STATE_MACHINE_INSTANCE_TO_STATEMENT_FOR_INSERT, machineInstance);
if (effect < 1) {
throw new StoreException("StateMachineInstance record start error, Xid: " + machineInstance.getId(), FrameworkErrorCode.OperationDenied);
}
} catch (StoreException e) {
LOGGER.error("Record statemachine start error: {}, StateMachine: {}, XID: {}, Reason: {}", e.getErrcode(), machineInstance.getStateMachine().getName(), machineInstance.getId(), e.getMessage(), e);
this.clearUp();
throw e;
}
}
}
protected void beginTransaction(StateMachineInstance machineInstance, ProcessContext context) {
if (sagaTransactionalTemplate != null) {
//获取状态机配置
StateMachineConfig stateMachineConfig = (StateMachineConfig) context.getVariable(DomainConstants.VAR_NAME_STATEMACHINE_CONFIG);
//构建事务信息
TransactionInfo transactionInfo = new TransactionInfo();
transactionInfo.setTimeOut(stateMachineConfig.getTransOperationTimeout());
transactionInfo.setName(Constants.SAGA_TRANS_NAME_PREFIX + machineInstance.getStateMachine().getName());
try {
//通过Saga事务模版开启全局事务
GlobalTransaction globalTransaction = sagaTransactionalTemplate.beginTransaction(transactionInfo);
machineInstance.setId(globalTransaction.getXid());
context.setVariable(DomainConstants.VAR_NAME_GLOBAL_TX, globalTransaction);
Map<String, Object> machineContext = machineInstance.getContext();
if (machineContext != null) {
machineContext.put(DomainConstants.VAR_NAME_GLOBAL_TX, globalTransaction);
}
} catch (ExecutionException e) {
String xid = null;
if (e.getTransaction() != null) {
xid = e.getTransaction().getXid();
}
throw new EngineExecutionException(e, e.getCode() + ", TransName:" + transactionInfo.getName() + ", XID: " + xid + ", Reason: " + e.getMessage(), FrameworkErrorCode.TransactionManagerError);
} finally {
if (Boolean.TRUE.equals(context.getVariable(DomainConstants.VAR_NAME_IS_ASYNC_EXECUTION))) {
RootContext.unbind();
RootContext.unbindBranchType();
}
}
}
}
...
}
public class StateLogStoreSqls {
private String recordStateMachineStartedSql;
private static final String RECORD_STATE_MACHINE_STARTED_SQL = "INSERT INTO ${TABLE_PREFIX}state_machine_inst\n"
+ "(id, machine_id, tenant_id, parent_id, gmt_started, business_key, start_params, is_running, status, "
+ "gmt_updated)\n"
+ "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
...
public StateLogStoreSqls(String tablePrefix) {
this.tablePrefix = tablePrefix;
init();
}
private void init() {
recordStateMachineStartedSql = RECORD_STATE_MACHINE_STARTED_SQL.replaceAll(TABLE_PREFIX_REGEX, tablePrefix);
...
}
public String getRecordStateMachineStartedSql(String dbType) {
return recordStateMachineStartedSql;
}
...
}
public class AbstractStore {
protected DataSource dataSource;
...
protected <T> int executeUpdate(String sql, ObjectToStatement<T> objectToStatement, T o) {
Connection connection = null;
PreparedStatement stmt = null;
try {
connection = dataSource.getConnection();
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Preparing SQL: {}", sql);
}
stmt = connection.prepareStatement(sql);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("setting params to PreparedStatement: {}", BeanUtils.beanToString(o));
}
objectToStatement.toStatement(o, stmt);
int count = stmt.executeUpdate();
if (!connection.getAutoCommit()) {
connection.commit();
}
return count;
} catch (SQLException e) {
throw new StoreException(e);
} finally {
closeSilent(stmt);
closeSilent(connection);
}
}
...
}
public class DefaultSagaTransactionalTemplate implements SagaTransactionalTemplate, ApplicationContextAware, DisposableBean, InitializingBean {
...
@Override
public GlobalTransaction beginTransaction(TransactionInfo txInfo) throws TransactionalExecutor.ExecutionException {
GlobalTransaction tx = GlobalTransactionContext.getCurrentOrCreate();
try {
triggerBeforeBegin();
//开启一个全局会话
tx.begin(txInfo.getTimeOut(), txInfo.getName());
triggerAfterBegin();
} catch (TransactionException txe) {
throw new TransactionalExecutor.ExecutionException(tx, txe, TransactionalExecutor.Code.BeginFailure);
}
return tx;
}
...
}执行GlobalTransaction的begin()方法开启一个全局会话：
//The type Default global transaction.
//默认的全局事务
public class DefaultGlobalTransaction implements GlobalTransaction {
private TransactionManager transactionManager;
private String xid;
private GlobalStatus status;
private GlobalTransactionRole role;
...
DefaultGlobalTransaction() {
//全局事务角色是全局事务发起者
this(null, GlobalStatus.UnKnown, GlobalTransactionRole.Launcher);
}
//Instantiates a new Default global transaction.
DefaultGlobalTransaction(String xid, GlobalStatus status, GlobalTransactionRole role) {
this.transactionManager = TransactionManagerHolder.get();//全局事务管理者
this.xid = xid;
this.status = status;
this.role = role;
}
...
@Override
public void begin(int timeout, String name) throws TransactionException {
if (role != GlobalTransactionRole.Launcher) {
assertXIDNotNull();
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Ignore Begin(): just involved in global transaction [{}]", xid);
}
return;
}
assertXIDNull();
String currentXid = RootContext.getXID();
if (currentXid != null) {
throw new IllegalStateException("Global transaction already exists," + " can't begin a new global transaction, currentXid = " + currentXid);
}
//通过全局事务管理器去真正开启全局事务，一旦开启成功，就可以获取到一个xid
xid = transactionManager.begin(null, null, name, timeout);
status = GlobalStatus.Begin;
//把xid绑定到RootContext的线程本地变量副本里去
RootContext.bind(xid);
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Begin new global transaction [{}]", xid);
}
}
...
}
public class TransactionManagerHolder {
private static final Logger LOGGER = LoggerFactory.getLogger(TransactionManagerHolder.class);
private static class SingletonHolder {
private static TransactionManager INSTANCE = null;
static {
try {
INSTANCE = EnhancedServiceLoader.load(TransactionManager.class);
LOGGER.info("TransactionManager Singleton {}", INSTANCE);
} catch (Throwable anyEx) {
LOGGER.error("Failed to load TransactionManager Singleton! ", anyEx);
}
}
}
//Get transaction manager.
public static TransactionManager get() {
if (SingletonHolder.INSTANCE == null) {
throw new ShouldNeverHappenException("TransactionManager is NOT ready!");
}
return SingletonHolder.INSTANCE;
}
private TransactionManagerHolder() {
}
}
//The type Default transaction manager.
//默认的全局事务管理器
public class DefaultTransactionManager implements TransactionManager {
@Override
public String begin(String applicationId, String transactionServiceGroup, String name, int timeout) throws TransactionException {
//构建一个全局事务开启请求GlobalBeginRequest
GlobalBeginRequest request = new GlobalBeginRequest();
request.setTransactionName(name);
request.setTimeout(timeout);
//发起一个同步调用
GlobalBeginResponse response = (GlobalBeginResponse) syncCall(request);
if (response.getResultCode() == ResultCode.Failed) {
throw new TmTransactionException(TransactionExceptionCode.BeginFailed, response.getMsg());
}
return response.getXid();
}
...
private AbstractTransactionResponse syncCall(AbstractTransactionRequest request) throws TransactionException {
try {
//TMNettyRemotingClient会和Seata Server基于Netty建立长连接
return (AbstractTransactionResponse) TmNettyRemotingClient.getInstance().sendSyncRequest(request);
} catch (TimeoutException toe) {
throw new TmTransactionException(TransactionExceptionCode.IO, "RPC timeout", toe);
}
}
}
public abstract class AbstractNettyRemotingClient extends AbstractNettyRemoting implements RemotingClient {
...
@Override
public Object sendSyncRequest(Object msg) throws TimeoutException {
//因为Seata Server是可以多节点部署实现高可用架构的，所以这里调用loadBalance()方法进行负载均衡
String serverAddress = loadBalance(getTransactionServiceGroup(), msg);
//获取RPC调用的超时时间
long timeoutMillis = this.getRpcRequestTimeout();
//构建一个RPC消息
RpcMessage rpcMessage = buildRequestMessage(msg, ProtocolConstants.MSGTYPE_RESQUEST_SYNC);
//send batch message
//put message into basketMap, @see MergedSendRunnable
//默认是不开启批量消息发送
if (this.isEnableClientBatchSendRequest()) {
...
} else {
//通过网络连接管理器clientChannelManager，获取与指定Seata Server建立的网络连接Channel
//然后通过网络连接Channel把RpcMessage发送出去
Channel channel = clientChannelManager.acquireChannel(serverAddress);
return super.sendSync(channel, rpcMessage, timeoutMillis);
}
}
...
}
public abstract class AbstractNettyRemoting implements Disposable {
...
protected Object sendSync(Channel channel, RpcMessage rpcMessage, long timeoutMillis) throws TimeoutException {
if (timeoutMillis <= 0) {
throw new FrameworkException("timeout should more than 0ms");
}
if (channel == null) {
LOGGER.warn("sendSync nothing, caused by null channel.");
return null;
}
//把发送出去的请求封装到MessageFuture中，然后存放到futures这个Map里
MessageFuture messageFuture = new MessageFuture();
messageFuture.setRequestMessage(rpcMessage);
messageFuture.setTimeout(timeoutMillis);
futures.put(rpcMessage.getId(), messageFuture);
channelWritableCheck(channel, rpcMessage.getBody());
//获取远程地址
String remoteAddr = ChannelUtil.getAddressFromChannel(channel);
doBeforeRpcHooks(remoteAddr, rpcMessage);
//异步化发送数据，同时对发送结果添加监听器
//如果发送失败，则会对网络连接Channel进行销毁处理
channel.writeAndFlush(rpcMessage).addListener((ChannelFutureListener) future -> {
if (!future.isSuccess()) {
MessageFuture messageFuture1 = futures.remove(rpcMessage.getId());
if (messageFuture1 != null) {
messageFuture1.setResultMessage(future.cause());
}
destroyChannel(future.channel());
}
});
try {
//然后通过请求响应组件MessageFuture同步等待Seata Server返回该请求的响应
Object result = messageFuture.get(timeoutMillis, TimeUnit.MILLISECONDS);
doAfterRpcHooks(remoteAddr, rpcMessage, result);
return result;
} catch (Exception exx) {
LOGGER.error("wait response error:{},ip:{},request:{}", exx.getMessage(), channel.remoteAddress(), rpcMessage.getBody());
if (exx instanceof TimeoutException) {
throw (TimeoutException) exx;
} else {
throw new RuntimeException(exx);
}
}
}
...
}15.状态操作组件获取状态和State状态类继承体系
(1)状态操作组件StateInstruction获取状态
(2)State状态类的继承体系
(1)状态操作组件StateInstruction获取状态
在状态机引擎
ProcessCtrlStateMachineEngine的startInternal()方法中，会通过流程上下文ProcessContext来获取状态操作组件StateInstruction。
根据状态操作组件StateInstruction，可以获取状态机的开始State(状态)、也可以获取指定的某个State(状态)。
public class ProcessCtrlStateMachineEngine implements StateMachineEngine {
...
//启动状态机实例StateMachineInstance
//@param stateMachineName 状态机名称
//@param tenantId 租户ID
//@param businessKey 业务key
//@param startParams 状态机实例的启动参数
//@param async 是否异步化运行
//@param callback 异步化运行时的回调接口
private StateMachineInstance startInternal(String stateMachineName, String tenantId, String businessKey, Map<String, Object> startParams, boolean async, AsyncCallback callback) throws EngineExecutionException {
try {
//如果指定需要异步运行，但是状态机配置里是不允许异步运行的，则会抛异常
if (async && !stateMachineConfig.isEnableAsync()) {
throw new EngineExecutionException("Asynchronous start is disabled. please set StateMachineConfig.enableAsync=true first.", FrameworkErrorCode.AsynchronousStartDisabled);
}
if (StringUtils.isEmpty(tenantId)) {
tenantId = stateMachineConfig.getDefaultTenantId();
}
//创建状态机实例StateMachineInstance
StateMachineInstance instance = createMachineInstance(stateMachineName, tenantId, businessKey, startParams);
//创建一个流程上下文构造器ProcessContextBuilder实例，用来构造流程运行时的上下文
ProcessContextBuilder contextBuilder = ProcessContextBuilder.create()
.withProcessType(ProcessType.STATE_LANG)//设置流程类型
.withOperationName(DomainConstants.OPERATION_NAME_START)//设置操作名称
.withAsyncCallback(callback)//设置异步化时的回调接口
.withInstruction(new StateInstruction(stateMachineName, tenantId))//设置状态获取组件
.withStateMachineInstance(instance)//设置状态机实例
.withStateMachineConfig(getStateMachineConfig())//设置状态机配置
.withStateMachineEngine(this);//设置状态机引擎
//上下文变量Map
Map<String, Object> contextVariables;
if (startParams != null) {
contextVariables = new ConcurrentHashMap<>(startParams.size());
nullSafeCopy(startParams, contextVariables);
} else {
contextVariables = new ConcurrentHashMap<>();
}
instance.setContext(contextVariables);
//设置流程上下文构造器ProcessContextBuilder实例
contextBuilder.withStateMachineContextVariables(contextVariables);
contextBuilder.withIsAsyncExecution(async);
//通过流程上下文构造器ProcessContextBuilder构建出一个流程上下文ProcessContext
ProcessContext processContext = contextBuilder.build();
//如果状态机定义StateMachine是支持持久化的 且 状态日志的存储组件不为null
if (instance.getStateMachine().isPersist() && stateMachineConfig.getStateLogStore() != null) {
//通过状态机实例日志和状态实例日志的存储组件StateLogStore，记录状态机实例StateMachineInstance的启动事件日志 + 开启全局事务
//比如在DB中更新状态机实例StateMachineInstance的启动状态
stateMachineConfig.getStateLogStore().recordStateMachineStarted(instance, processContext);
}
if (StringUtils.isEmpty(instance.getId())) {
//生成状态机实例StateMachineInstance的序号
instance.setId(stateMachineConfig.getSeqGenerator().generate(DomainConstants.SEQ_ENTITY_STATE_MACHINE_INST));
}
//从流程上下文中获取状态操作组件StateInstruction
StateInstruction stateInstruction = processContext.getInstruction(StateInstruction.class);
//获取循环策略
Loop loop = LoopTaskUtils.getLoopConfig(processContext, stateInstruction.getState(processContext));
if (null != loop) {
stateInstruction.setTemporaryState(new LoopStartStateImpl());
}
if (async) {
stateMachineConfig.getAsyncProcessCtrlEventPublisher().publish(processContext);
} else {
stateMachineConfig.getProcessCtrlEventPublisher().publish(processContext);
}
return instance;
} finally {
if (stateMachineConfig.getStateLogStore() != null) {
stateMachineConfig.getStateLogStore().clearUp();
}
}
}
...
}
//State Instruction，状态操作组件StateInstruction
public class StateInstruction implements Instruction {
private String stateName;
private String stateMachineName;
private String tenantId;
private boolean end;
//Temporary state node for running temporary nodes in the state machine
private State temporaryState;
public StateInstruction() {
}
public StateInstruction(String stateMachineName, String tenantId) {
this.stateMachineName = stateMachineName;
this.tenantId = tenantId;
}
//根据流程上下文获取状态，可以是状态机的开始State(状态)、也可以是指定的某个State(状态)
public State getState(ProcessContext context) {
if (getTemporaryState() != null) {
return temporaryState;
}
String stateName = getStateName();
String stateMachineName = getStateMachineName();
String tenantId = getTenantId();
if (StringUtils.isEmpty(stateMachineName)) {
throw new EngineExecutionException("StateMachineName is required", FrameworkErrorCode.ParameterRequired);
}
StateMachineConfig stateMachineConfig = (StateMachineConfig)context.getVariable(DomainConstants.VAR_NAME_STATEMACHINE_CONFIG);
StateMachine stateMachine = stateMachineConfig.getStateMachineRepository().getStateMachine(stateMachineName, tenantId);
if (stateMachine == null) {
throw new EngineExecutionException("StateMachine[" + stateMachineName + "] is not exist", FrameworkErrorCode.ObjectNotExists);
}
//默认就是获取状态机定义StateMachine中的第一个状态State
if (StringUtils.isEmpty(stateName)) {
stateName = stateMachine.getStartState();
setStateName(stateName);
}
//根据状态名字stateName，从状态机定义中获取对应的状态State
State state = stateMachine.getStates().get(stateName);
if (state == null) {
throw new EngineExecutionException("State[" + stateName + "] is not exist", FrameworkErrorCode.ObjectNotExists);
}
return state;
}
...
}(2)State状态类的继承体系
//A State in StateMachine
public interface State {
String getName();
String getComment();
String getType();
String getNext();
Map<String, Object> getExtensions();
StateMachine getStateMachine();
}
public abstract class BaseState implements State {
private transient String name;//状态名称
private String type;//状态类型
private String comment;//状态备注
private String next;//下一个状态名称
private Map<String, Object> extensions;//状态扩展属性
private transient StateMachine stateMachine;//状态所属的状态机
...
}
//The state of the execution task (abstract class), the specific task to be executed is determined by the subclass
public abstract class AbstractTaskState extends BaseState implements TaskState {
private String compensateState;//补偿状态
private boolean isForCompensation;//是否用于补偿
private boolean isForUpdate;//是否用于更新
private List<Retry> retry;//重试列表
private List<ExceptionMatch> catches;//异常匹配列表
private List<Object> input;//状态输入列表
private Map<String, Object> output;//状态输出数据
private Map<String, String> status;//Map<String/* expression */, String /* status */>
private List<Object> inputExpressions;//输入表达式列表
private Map<String, Object> outputExpressions;//输出表达式数据
private boolean isPersist = true;//是否持久化
private Boolean retryPersistModeUpdate;//是否更新Saga重试持久化的模式
private Boolean compensatePersistModeUpdate;//是否更新Saga补偿持久化的模式
private Loop loop;//循环对象
...
public static class RetryImpl implements Retry {
private List<String> exceptions;//重试的过程中遇到了哪些异常
private List<Class<? extends Exception>> exceptionClasses;//异常Class
private double intervalSeconds;//重试间隔，单位是秒
private int maxAttempts;//最大重试次数
private double backoffRate;//回退比例
...
}
public static class ExceptionMatchImpl implements ExceptionMatch {
List<String> exceptions;
List<Class<? extends Exception>> exceptionClasses;
String next;
...
}
public static class LoopImpl implements Loop {
private int parallel;//并行度
private String collection;
private String elementVariableName;
private String elementIndexName;
private String completionCondition;
...
}
...
}16.启动状态机实例时发布流程上下文到事件总线
(1)启动状态机实例时的主要工作总结
(2)发布流程上下文到事件总线的源码
(1)启动状态机实例时的主要工作总结
调用状态机引擎
ProcessCtrlStateMachineEngine的startInternal()方法时：
一.首先创建状态机实例StateMachineInstance
二.然后通过ProcessContextBuilder流程上下文构造器构建出一个流程上下文ProcessContext
三.接着记录状态机实例的启动日志 + 开启全局事务
四.然后从流程上下文获取状态操作组件来设置状态
五.最后发布流程上下文ProcessContext到事件总线
public class ProcessCtrlStateMachineEngine implements StateMachineEngine {
...
//启动状态机实例StateMachineInstance
//@param stateMachineName 状态机名称
//@param tenantId 租户ID
//@param businessKey 业务key
//@param startParams 状态机实例的启动参数
//@param async 是否异步化运行
//@param callback 异步化运行时的回调接口
private StateMachineInstance startInternal(String stateMachineName, String tenantId, String businessKey, Map<String, Object> startParams, boolean async, AsyncCallback callback) throws EngineExecutionException {
try {
//如果指定需要异步运行，但是状态机配置里是不允许异步运行的，则会抛异常
if (async && !stateMachineConfig.isEnableAsync()) {
throw new EngineExecutionException("Asynchronous start is disabled. please set StateMachineConfig.enableAsync=true first.", FrameworkErrorCode.AsynchronousStartDisabled);
}
if (StringUtils.isEmpty(tenantId)) {
tenantId = stateMachineConfig.getDefaultTenantId();
}
//1.创建状态机实例StateMachineInstance
StateMachineInstance instance = createMachineInstance(stateMachineName, tenantId, businessKey, startParams);
//创建一个流程上下文构造器ProcessContextBuilder实例，用来构造流程运行时的上下文
ProcessContextBuilder contextBuilder = ProcessContextBuilder.create()
.withProcessType(ProcessType.STATE_LANG)//设置流程类型
.withOperationName(DomainConstants.OPERATION_NAME_START)//设置操作名称
.withAsyncCallback(callback)//设置异步化时的回调接口
.withInstruction(new StateInstruction(stateMachineName, tenantId))//设置状态获取组件
.withStateMachineInstance(instance)//设置状态机实例
.withStateMachineConfig(getStateMachineConfig())//设置状态机配置
.withStateMachineEngine(this);//设置状态机引擎
//上下文变量Map
Map<String, Object> contextVariables;
if (startParams != null) {
contextVariables = new ConcurrentHashMap<>(startParams.size());
nullSafeCopy(startParams, contextVariables);
} else {
contextVariables = new ConcurrentHashMap<>();
}
instance.setContext(contextVariables);
//设置流程上下文构造器ProcessContextBuilder实例
contextBuilder.withStateMachineContextVariables(contextVariables);
contextBuilder.withIsAsyncExecution(async);
//2.通过流程上下文构造器ProcessContextBuilder构建出一个流程上下文ProcessContext
ProcessContext processContext = contextBuilder.build();
//如果状态机定义StateMachine是支持持久化的 且 状态日志的存储组件不为null
if (instance.getStateMachine().isPersist() && stateMachineConfig.getStateLogStore() != null) {
//3.通过状态机实例日志和状态实例日志的存储组件StateLogStore，记录状态机实例StateMachineInstance的启动事件日志 + 开启全局事务
//比如在DB中更新状态机实例StateMachineInstance的启动状态
stateMachineConfig.getStateLogStore().recordStateMachineStarted(instance, processContext);
}
if (StringUtils.isEmpty(instance.getId())) {
//生成状态机实例StateMachineInstance的序号
instance.setId(stateMachineConfig.getSeqGenerator().generate(DomainConstants.SEQ_ENTITY_STATE_MACHINE_INST));
}
//从流程上下文中获取状态操作组件StateInstruction
StateInstruction stateInstruction = processContext.getInstruction(StateInstruction.class);
//4.先获取State，再根据State获取循环策略Loop，然后判定Loop是否为空来决定是否设置"状态操作组件StateInstruction"
Loop loop = LoopTaskUtils.getLoopConfig(processContext, stateInstruction.getState(processContext));
if (null != loop) {
stateInstruction.setTemporaryState(new LoopStartStateImpl());
}
//5.发布流程上下文到事件总线
if (async) {
stateMachineConfig.getAsyncProcessCtrlEventPublisher().publish(processContext);
} else {
stateMachineConfig.getProcessCtrlEventPublisher().publish(processContext);
}
return instance;
} finally {
if (stateMachineConfig.getStateLogStore() != null) {
stateMachineConfig.getStateLogStore().clearUp();
}
}
}
...
}(2)发布流程上下文到事件总线的源码
首先通过状态机配置实例stateMachineConfig的
getProcessCtrlEventPublisher()方法获取流程控制事件发布器ProcessCtrlEventPublisher，然后通过流程控制事件发布器的publish()方法，将流程上下文ProcessContext发布到Seata的事件总线EventBus中。
一.同步发送流程上下文到事件总线的分析
此时会将流程上下文放入到发布器内部的栈结构中，然后判断如果是栈结构的第一个元素，则使用一个while循环不断从栈结构中弹出流程上下文，调用流程控制事件消费者进行处理。
二.异步发送流程上下文到事件总线的分析
此时会通过线程池去执行调用流程控制事件消费者来处理流程上下文的逻辑。
public class DefaultStateMachineConfig implements StateMachineConfig, ApplicationContextAware, InitializingBean {
//流程控制的事件发布器(同步发布)
private ProcessCtrlEventPublisher syncProcessCtrlEventPublisher;
//流程控制的事件发布器(异步发布)
private ProcessCtrlEventPublisher asyncProcessCtrlEventPublisher;
...
//因为DefaultStateMachineConfig实现了InitializingBean接口
//所以继承DefaultStateMachineConfig的Spring Bean初始化之后，就会回调afterPropertiesSet()方法
@Override
public void afterPropertiesSet() throws Exception {
init();
}
//初始化方法
protected void init() throws Exception {
...
//创建流程控制的事件发布器(同步发布)
if (syncProcessCtrlEventPublisher == null) {
//创建流程控制的事件发布器
ProcessCtrlEventPublisher syncEventPublisher = new ProcessCtrlEventPublisher();
//创建流程控制器
ProcessControllerImpl processorController = createProcessorController(syncEventPublisher);
//创建流程控制事件的消费者
ProcessCtrlEventConsumer processCtrlEventConsumer = new ProcessCtrlEventConsumer();
processCtrlEventConsumer.setProcessController(processorController);
//创建事件总线
DirectEventBus directEventBus = new DirectEventBus();
syncEventPublisher.setEventBus(directEventBus);
directEventBus.registerEventConsumer(processCtrlEventConsumer);
syncProcessCtrlEventPublisher = syncEventPublisher;
}
//如果启用了异步化执行 且 流程控制的事件发布器(异步发布)为null
if (enableAsync && asyncProcessCtrlEventPublisher == null) {
ProcessCtrlEventPublisher asyncEventPublisher = new ProcessCtrlEventPublisher();
ProcessControllerImpl processorController = createProcessorController(asyncEventPublisher);
ProcessCtrlEventConsumer processCtrlEventConsumer = new ProcessCtrlEventConsumer();
processCtrlEventConsumer.setProcessController(processorController);
AsyncEventBus asyncEventBus = new AsyncEventBus();
asyncEventBus.setThreadPoolExecutor(getThreadPoolExecutor());
asyncEventPublisher.setEventBus(asyncEventBus);
asyncEventBus.registerEventConsumer(processCtrlEventConsumer);
asyncProcessCtrlEventPublisher = asyncEventPublisher;
}
...
}
...
@Override
public ProcessCtrlEventPublisher getProcessCtrlEventPublisher() {
return syncProcessCtrlEventPublisher;
}
@Override
public ProcessCtrlEventPublisher getAsyncProcessCtrlEventPublisher() {
return asyncProcessCtrlEventPublisher;
}
...
}
//ProcessCtrl Event Pulisher
public class ProcessCtrlEventPublisher implements EventPublisher<ProcessContext> {
private EventBus<ProcessContext> eventBus;
@Override
public boolean publish(ProcessContext event) throws FrameworkException {
return eventBus.offer(event);
}
public void setEventBus(EventBus<ProcessContext> eventBus) {
this.eventBus = eventBus;
}
}
//同步发布流程上下文到事件总线
public class DirectEventBus extends AbstractEventBus<ProcessContext> {
private static final Logger LOGGER = LoggerFactory.getLogger(DirectEventBus.class);
private static final String VAR_NAME_SYNC_EXE_STACK = "_sync_execution_stack_";
@Override
public boolean offer(ProcessContext context) throws FrameworkException {
List<EventConsumer> eventHandlers = getEventConsumers(context.getClass());
if (CollectionUtils.isEmpty(eventHandlers)) {
if (LOGGER.isWarnEnabled()) {
LOGGER.warn("cannot find event handler by class: " + context.getClass());
}
return false;
}
//将流程上下文放入到其内部的栈结构中
boolean isFirstEvent = false;
Stack<ProcessContext> currentStack = (Stack<ProcessContext>)context.getVariable(VAR_NAME_SYNC_EXE_STACK);
if (currentStack == null) {
synchronized (context) {
currentStack = (Stack<ProcessContext>)context.getVariable(VAR_NAME_SYNC_EXE_STACK);
if (currentStack == null) {
currentStack = new Stack<>();
context.setVariable(VAR_NAME_SYNC_EXE_STACK, currentStack);
isFirstEvent = true;
}
}
}
currentStack.push(context);
//如果是栈结构的第一个元素，则开启一个while循环，不断从栈结构中弹出流程上下文，调用流程控制事件消费者进行处理
if (isFirstEvent) {
try {
while (currentStack.size() > 0) {
ProcessContext currentContext = currentStack.pop();
for (EventConsumer eventHandler : eventHandlers) {
eventHandler.process(currentContext);
}
}
} finally {
context.removeVariable(VAR_NAME_SYNC_EXE_STACK);
}
}
return true;
}
}
public abstract class AbstractEventBus<E> implements EventBus<E> {
private List<EventConsumer> eventConsumerList = new ArrayList<>();
@Override
public List<EventConsumer> getEventConsumers(Class clazz) {
List<EventConsumer> acceptedConsumers = new ArrayList<>();
for (EventConsumer eventConsumer : eventConsumerList) {
if (eventConsumer.accept(clazz)) {
acceptedConsumers.add(eventConsumer);
}
}
return acceptedConsumers;
}
@Override
public void registerEventConsumer(EventConsumer eventConsumer) {
eventConsumerList.add(eventConsumer);
}
}
//异步发布流程上下文到事件总线
public class AsyncEventBus extends AbstractEventBus<ProcessContext> {
private static final Logger LOGGER = LoggerFactory.getLogger(AsyncEventBus.class);
private ThreadPoolExecutor threadPoolExecutor;
@Override
public boolean offer(ProcessContext context) throws FrameworkException {
List<EventConsumer> eventConsumers = getEventConsumers(context.getClass());
if (CollectionUtils.isEmpty(eventConsumers)) {
if (LOGGER.isWarnEnabled()) {
LOGGER.warn("cannot find event handler by class: " + context.getClass());
}
return false;
}
//通过线程池去调用流程控制事件消费者来处理流程上下文
for (EventConsumer eventConsumer : eventConsumers) {
threadPoolExecutor.execute(() -> eventConsumer.process(context));
}
return true;
}
public void setThreadPoolExecutor(ThreadPoolExecutor threadPoolExecutor) {
this.threadPoolExecutor = threadPoolExecutor;
}
}17.通过业务处理器处理状态机当前需要执行的状态
Seata Saga状态机实例的处理流程，其实就是通过流程控制事件消费者处理流程上下文来触发的，也就是调用ProcessCtrlEventConsumer的process()方法时触发的。
在流程处理控制器ProcessController的process()方法中，首先会调用自定义业务处理器
CustomizeBusinessProcessor的process()方法来处理状态机当前需要执行的状态State，然后会继续调用自定义业务处理器的route()方法将自定义业务处理器路由到状态机下一个要执行的状态State。
在业务处理器
CustomizeBusinessProcessor的process()方法中，会调用状态机流程处理器
StateMachineProcessHandler的process()方法，也就是会调用状态处理器ServiceTaskStateHandler的process()方法处理状态。
在ServiceTaskStateHandler的process()方法中，最终会触发调用服务调用组件SpringBeanServiceInvoker的invoke()方法，通过反射机制执行状态机当前State指定的服务方法，最后将执行结果放入流程上下文里的状态机实例中。
如果执行异常，则触发回滚补偿处理。
//ProcessCtrl Event Consumer
public class ProcessCtrlEventConsumer implements EventConsumer<ProcessContext> {
private ProcessController processController;
@Override
public void process(ProcessContext event) throws FrameworkException {
//调用流程处理器的process()方法
processController.process(event);
}
@Override
public boolean accept(Class<ProcessContext> clazz) {
return ProcessContext.class.isAssignableFrom(clazz);
}
public void setProcessController(ProcessController processController) {
this.processController = processController;
}
}
//Default implementation of Process controller
//默认的流程处理器
public class ProcessControllerImpl implements ProcessController {
private static final Logger LOGGER = LoggerFactory.getLogger(ProcessControllerImpl.class);
private BusinessProcessor businessProcessor;
@Override
public void process(ProcessContext context) throws FrameworkException {
try {
//1.BusinessProcessor处理状态机当前需要执行的状态State，调用CustomizeBusinessProcessor.process()方法
businessProcessor.process(context);
//2.BusinessProcessor路由到状态机的下一个需要执行的状态State，调用CustomizeBusinessProcessor.route()方法
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);
}
}
public void setBusinessProcessor(BusinessProcessor businessProcessor) {
this.businessProcessor = businessProcessor;
}
}
//Customizable Business Processor
public class CustomizeBusinessProcessor implements BusinessProcessor {
private Map<String, ProcessHandler> processHandlers;
...
@Override
public void process(ProcessContext context) throws FrameworkException {
ProcessType processType = matchProcessType(context);
if (processType == null) {
if (LOGGER.isWarnEnabled()) {
LOGGER.warn("Process type not found, context= {}", context);
}
throw new FrameworkException(FrameworkErrorCode.ProcessTypeNotFound);
}
ProcessHandler processor = processHandlers.get(processType.getCode());
if (processor == null) {
LOGGER.error("Cannot find process handler by type {}, context= {}", processType.getCode(), context);
throw new FrameworkException(FrameworkErrorCode.ProcessHandlerNotFound);
}
//调用StateMachineProcessHandler.process()方法处理
processor.process(context);
}
@Override
public void route(ProcessContext context) throws FrameworkException {
ProcessType processType = matchProcessType(context);
if (processType == null) {
if (LOGGER.isWarnEnabled()) {
LOGGER.warn("Process type not found, the process is no longer advanced, context= {}", context);
}
return;
}
RouterHandler router = routerHandlers.get(processType.getCode());
if (router == null) {
LOGGER.error("Cannot find router handler by type {}, context= {}", processType.getCode(), context);
return;
}
//先调用DefaultRouterHandler的route()方法
//再调用StateMachineProcessRouter.route()方法，将StateInstruction路由到下一个State
router.route(context);
}
...
}
public class StateMachineProcessHandler implements ProcessHandler {
private final Map<String, StateHandler> stateHandlers = new ConcurrentHashMap<>();
@Override
public void process(ProcessContext context) throws FrameworkException {
//获取状态处理器StateHandler后，调用stateHandler.process()进行状态的业务逻辑处理
StateInstruction instruction = context.getInstruction(StateInstruction.class);
State state = instruction.getState(context);
String stateType = state.getType();
StateHandler stateHandler = stateHandlers.get(stateType);
List<StateHandlerInterceptor> interceptors = null;
if (stateHandler instanceof InterceptableStateHandler) {
interceptors = ((InterceptableStateHandler)stateHandler).getInterceptors();
}
//获取状态拦截器StateHandlerInterceptor后，
//调用interceptor.preProcess()进行预处理，调用interceptor.postProcess()进行后置处理
List<StateHandlerInterceptor> executedInterceptors = null;
Exception exception = null;
try {
if (CollectionUtils.isNotEmpty(interceptors)) {
executedInterceptors = new ArrayList<>(interceptors.size());
for (StateHandlerInterceptor interceptor : interceptors) {
executedInterceptors.add(interceptor);
interceptor.preProcess(context);
}
}
//调用ServiceTaskStateHandler的process()方法
stateHandler.process(context);
} catch (Exception e) {
exception = e;
throw e;
} finally {
if (CollectionUtils.isNotEmpty(executedInterceptors)) {
for (int i = executedInterceptors.size() - 1; i >= 0; i--) {
StateHandlerInterceptor interceptor = executedInterceptors.get(i);
interceptor.postProcess(context, exception);
}
}
}
}
...
}
public class ServiceTaskStateHandler implements StateHandler, InterceptableStateHandler {
private static final Logger LOGGER = LoggerFactory.getLogger(ServiceTaskStateHandler.class);
private List<StateHandlerInterceptor> interceptors = new ArrayList<>();
@Override
public void process(ProcessContext context) throws EngineExecutionException {
StateInstruction instruction = context.getInstruction(StateInstruction.class);
ServiceTaskStateImpl state = (ServiceTaskStateImpl) instruction.getState(context);
String serviceName = state.getServiceName();
String methodName = state.getServiceMethod();
StateInstance stateInstance = (StateInstance) context.getVariable(DomainConstants.VAR_NAME_STATE_INST);
Object result;
try {
List<Object> input = (List<Object>) context.getVariable(DomainConstants.VAR_NAME_INPUT_PARAMS);
//Set the current task execution status to RU (Running)
stateInstance.setStatus(ExecutionStatus.RU);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(">>>>>>>>>>>>>>>>>>>>>> Start to execute State[{}], ServiceName[{}], Method[{}], Input:{}", state.getName(), serviceName, methodName, input);
}
if (state instanceof CompensateSubStateMachineState) {
//If it is the compensation of the substate machine,
//directly call the state machine's compensate method
result = compensateSubStateMachine(context, state, input, stateInstance, (StateMachineEngine) context.getVariable(DomainConstants.VAR_NAME_STATEMACHINE_ENGINE));
} else {
StateMachineConfig stateMachineConfig = (StateMachineConfig) context.getVariable(DomainConstants.VAR_NAME_STATEMACHINE_CONFIG);
//1.获取服务调用组件ServiceInvoker
ServiceInvoker serviceInvoker = stateMachineConfig.getServiceInvokerManager().getServiceInvoker(state.getServiceType());
if (serviceInvoker == null) {
throw new EngineExecutionException("No such ServiceInvoker[" + state.getServiceType() + "]", FrameworkErrorCode.ObjectNotExists);
}
if (serviceInvoker instanceof ApplicationContextAware) {
((ApplicationContextAware) serviceInvoker).setApplicationContext(stateMachineConfig.getApplicationContext());
}
//2.调用SpringBeanServiceInvoker的invoke()方法
result = serviceInvoker.invoke(state, input.toArray());
}
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("<<<<<<<<<<<<<<<<<<<<<< State[{}], ServiceName[{}], Method[{}] Execute finish. result: {}", state.getName(), serviceName, methodName, result);
}
if (result != null) {
//3.将执行结果放入流程上下文里的状态机实例中
stateInstance.setOutputParams(result);
((HierarchicalProcessContext) context).setVariableLocally(DomainConstants.VAR_NAME_OUTPUT_PARAMS, result);
}
} catch (Throwable e) {
LOGGER.error("<<<<<<<<<<<<<<<<<<<<<< State[{}], ServiceName[{}], Method[{}] Execute failed.", state.getName(), serviceName, methodName, e);
((HierarchicalProcessContext) context).setVariableLocally(DomainConstants.VAR_NAME_CURRENT_EXCEPTION, e);
//如果执行异常，下面的调用会获取触发异常的State，然后进行回滚补偿处理
EngineUtils.handleException(context, state, e);
}
}
...
}18.通过服务调用组件来执行State指定的服务方法
执行服务调用组件SpringBeanServiceInvoker的invoke()方法时，如果State需要异步执行，那么会通过线程池来异步执行doInvoke()方法。其中，doInvoke()方法首先会获取状态State的服务对应的Spring Bean，然后通过反射执行Bean的指定Method。如果出现异常，则会进行重试处理。
public class SpringBeanServiceInvoker implements ServiceInvoker, ApplicationContextAware {
private ApplicationContext applicationContext;
private ThreadPoolExecutor threadPoolExecutor;
...
@Override
public Object invoke(ServiceTaskState serviceTaskState, Object... input) throws Throwable {
ServiceTaskStateImpl state = (ServiceTaskStateImpl) serviceTaskState;
//如果State需要异步执行，那么就通过线程池来执行
if (state.isAsync()) {
if (threadPoolExecutor == null) {
if (LOGGER.isWarnEnabled()) {
LOGGER.warn("threadPoolExecutor is null, Service[{}.{}] cannot execute asynchronously, executing " + "synchronously now. stateName: {}", state.getServiceName(), state.getServiceMethod(), state.getName());
}
return doInvoke(state, input);
}
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Submit Service[{}.{}] to asynchronously executing. stateName: {}", state.getServiceName(), state.getServiceMethod(), state.getName());
}
threadPoolExecutor.execute(new Runnable() {
@Override
public void run() {
try {
doInvoke(state, input);
} catch (Throwable e) {
LOGGER.error("Invoke Service[" + state.getServiceName() + "." + state.getServiceMethod() + "] failed.", e);
}
}
});
return null;
} else {
return doInvoke(state, input);
}
}
protected Object doInvoke(ServiceTaskStateImpl state, Object[] input) throws Throwable {
//获取状态State的服务对应的Spring Bean
Object bean = applicationContext.getBean(state.getServiceName());
//获取状态State的方法
Method method = state.getMethod();
if (method == null) {
synchronized (state) {
method = state.getMethod();
if (method == null) {
method = findMethod(bean.getClass(), state.getServiceMethod(), state.getParameterTypes());
if (method != null) {
state.setMethod(method);
}
}
}
}
if (method == null) {
throw new EngineExecutionException("No such method[" + state.getServiceMethod() + "] on BeanClass[" + bean.getClass() + "]", FrameworkErrorCode.NoSuchMethod);
}
Object[] args = new Object[method.getParameterCount()];
try {
Class[] paramTypes = method.getParameterTypes();
if (input != null && input.length > 0) {
int len = input.length < paramTypes.length ? input.length : paramTypes.length;
for (int i = 0; i < len; i++) {
args[i] = toJavaObject(input[i], paramTypes[i]);
}
}
} catch (Exception e) {
throw new EngineExecutionException(e, "Input to java object error, Method[" + state.getServiceMethod() + "] on BeanClass[" + bean.getClass() + "]", FrameworkErrorCode.InvalidParameter);
}
if (!Modifier.isPublic(method.getModifiers())) {
throw new EngineExecutionException("Method[" + method.getName() + "] must be public", FrameworkErrorCode.MethodNotPublic);
}
Map<Retry, AtomicInteger> retryCountMap = new HashMap<>();
while (true) {
try {
//通过反射去调用State中配置的服务和方法
return invokeMethod(bean, method, args);
} catch (Throwable e) {
//重试处理
Retry matchedRetryConfig = matchRetryConfig(state.getRetry(), e);
if (matchedRetryConfig == null) {
throw e;
}
AtomicInteger retryCount = CollectionUtils.computeIfAbsent(retryCountMap, matchedRetryConfig, key -> new AtomicInteger(0));
if (retryCount.intValue() >= matchedRetryConfig.getMaxAttempts()) {
throw e;
}
double intervalSeconds = matchedRetryConfig.getIntervalSeconds();
double backoffRate = matchedRetryConfig.getBackoffRate();
long currentInterval = (long) (retryCount.intValue() > 0 ? (intervalSeconds * backoffRate * retryCount.intValue() * 1000) : (intervalSeconds * 1000));
if (LOGGER.isWarnEnabled()) {
LOGGER.warn("Invoke Service[" + state.getServiceName() + "." + state.getServiceMethod() + "] failed, will retry after " + currentInterval + " millis, current retry count: " + retryCount.intValue(), e);
}
try {
Thread.sleep(currentInterval);
} catch (InterruptedException e1) {
LOGGER.warn("Retry interval sleep error", e1);
}
retryCount.incrementAndGet();
}
}
}
protected Object invokeMethod(Object serviceBean, Method method, Object... input) throws Throwable {
try {
return method.invoke(serviceBean, input);
} catch (InvocationTargetException e) {
Throwable targetExp = e.getTargetException();
if (targetExp == null) {
throw new EngineExecutionException(e, e.getMessage(), FrameworkErrorCode.MethodInvokeError);
}
throw targetExp;
}
}
...
}19.将业务处理器路由到状态机下一个要执行的状态
在流程处理控制器ProcessController的process()方法中，会调用自定义的业务处理器
CustomizeBusinessProcessor的route()方法，将自定义业务处理器路由到状态机下一个需要执行的状态State。
CustomizeBusinessProcessor的route()方法，首先会调用DefaultRouterHandler的route()方法，然后再调用StateMachineProcessRouter的route()方法，将状态操作组件StateInstruction的状态名字设置为下一个State。
在DefaultRouterHandler的route()方法中，首先会调用StateMachineProcessRouter的route()方法，也就是调用TaskStateRouter的route()方法，将状态操作组件StateInstruction的状态名字设置为下一个State。然后将状态操作组件StateInstruction设置到流程上下文中，接着再将流程上下文ProcessContext发布到事件总线，继续由流程控制事件消费者进行消费处理。
//Default implementation of Process controller
//默认的流程处理器
public class ProcessControllerImpl implements ProcessController {
private static final Logger LOGGER = LoggerFactory.getLogger(ProcessControllerImpl.class);
private BusinessProcessor businessProcessor;
@Override
public void process(ProcessContext context) throws FrameworkException {
try {
//1.BusinessProcessor处理状态机当前需要执行的状态State，调用CustomizeBusinessProcessor.process()方法
businessProcessor.process(context);
//2.BusinessProcessor路由到状态机下一个需要执行的状态State，调用CustomizeBusinessProcessor.route()方法
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);
}
}
public void setBusinessProcessor(BusinessProcessor businessProcessor) {
this.businessProcessor = businessProcessor;
}
}
//Customizable Business Processor
public class CustomizeBusinessProcessor implements BusinessProcessor {
private Map<String, ProcessHandler> processHandlers;
...
@Override
public void route(ProcessContext context) throws FrameworkException {
ProcessType processType = matchProcessType(context);
if (processType == null) {
if (LOGGER.isWarnEnabled()) {
LOGGER.warn("Process type not found, the process is no longer advanced, context= {}", context);
}
return;
}
RouterHandler router = routerHandlers.get(processType.getCode());
if (router == null) {
LOGGER.error("Cannot find router handler by type {}, context= {}", processType.getCode(), context);
return;
}
//先调用DefaultRouterHandler的route()方法
//再调用StateMachineProcessRouter.route()方法，将StateInstruction路由到下一个State
router.route(context);
}
...
}
public class DefaultRouterHandler implements RouterHandler {
private EventPublisher<ProcessContext> eventPublisher;
private Map<String, ProcessRouter> processRouters;
public static ProcessType matchProcessType(ProcessContext context) {
ProcessType processType = (ProcessType)context.getVariable(ProcessContext.VAR_NAME_PROCESS_TYPE);
if (processType == null) {
processType = ProcessType.STATE_LANG;
}
return processType;
}
@Override
public void route(ProcessContext context) throws FrameworkException {
try {
ProcessType processType = matchProcessType(context);
if (processType == null) {
if (LOGGER.isWarnEnabled()) {
LOGGER.warn("Process type not found, context= {}", context);
}
throw new FrameworkException(FrameworkErrorCode.ProcessTypeNotFound);
}
ProcessRouter processRouter = processRouters.get(processType.getCode());
if (processRouter == null) {
LOGGER.error("Cannot find process router by type {}, context = {}", processType.getCode(), context);
throw new FrameworkException(FrameworkErrorCode.ProcessRouterNotFound);
}
//1.调用StateMachineProcessRouter.route()方法，将StateInstruction路由到下一个State
Instruction instruction = processRouter.route(context);
if (instruction == null) {
LOGGER.info("route instruction is null, process end");
} else {
//2.将流程上下文ProcessContext发布到事件总线，继续由流程控制事件消费者进行消费处理
context.setInstruction(instruction);
eventPublisher.publish(context);
}
} catch (FrameworkException e) {
throw e;
} catch (Exception ex) {
throw new FrameworkException(ex, ex.getMessage(), FrameworkErrorCode.UnknownAppError);
}
}
public void setEventPublisher(EventPublisher<ProcessContext> eventPublisher) {
this.eventPublisher = eventPublisher;
}
public void setProcessRouters(Map<String, ProcessRouter> processRouters) {
this.processRouters = processRouters;
}
}
public class StateMachineProcessRouter implements ProcessRouter {
private final Map<String, StateRouter> stateRouters = new ConcurrentHashMap<>();
@Override
public Instruction route(ProcessContext context) throws FrameworkException {
StateInstruction stateInstruction = context.getInstruction(StateInstruction.class);
State state;
if (stateInstruction.getTemporaryState() != null) {
state = stateInstruction.getTemporaryState();
stateInstruction.setTemporaryState(null);
} else {
StateMachineConfig stateMachineConfig = (StateMachineConfig)context.getVariable(DomainConstants.VAR_NAME_STATEMACHINE_CONFIG);
StateMachine stateMachine = stateMachineConfig.getStateMachineRepository().getStateMachine(stateInstruction.getStateMachineName(), stateInstruction.getTenantId());
state = stateMachine.getStates().get(stateInstruction.getStateName());
}
String stateType = state.getType();
StateRouter router = stateRouters.get(stateType);
Instruction instruction = null;
List<StateRouterInterceptor> interceptors = null;
if (router instanceof InterceptableStateRouter) {
interceptors = ((InterceptableStateRouter)router).getInterceptors();
}
List<StateRouterInterceptor> executedInterceptors = null;
Exception exception = null;
try {
if (CollectionUtils.isNotEmpty(interceptors)) {
executedInterceptors = new ArrayList<>(interceptors.size());
for (StateRouterInterceptor interceptor : interceptors) {
executedInterceptors.add(interceptor);
interceptor.preRoute(context, state);
}
}
//调用TaskStateRouter的route()方法
instruction = router.route(context, state);
} catch (Exception e) {
exception = e;
throw e;
} finally {
if (CollectionUtils.isNotEmpty(executedInterceptors)) {
for (int i = executedInterceptors.size() - 1; i >= 0; i--) {
StateRouterInterceptor interceptor = executedInterceptors.get(i);
interceptor.postRoute(context, state, instruction, exception);
}
}
//if 'Succeed' or 'Fail' State did not configured, we must end the state machine
if (instruction == null && !stateInstruction.isEnd()) {
EngineUtils.endStateMachine(context);
}
}
return instruction;
}
...
}
public class TaskStateRouter implements StateRouter {
@Override
public Instruction route(ProcessContext context, State state) throws EngineExecutionException {
StateInstruction stateInstruction = context.getInstruction(StateInstruction.class);
if (stateInstruction.isEnd()) {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("StateInstruction is ended, Stop the StateMachine executing. StateMachine[{}] Current State[{}]", stateInstruction.getStateMachineName(), state.getName());
}
return null;
}
//check if in loop async condition
if (Boolean.TRUE.equals(context.getVariable(DomainConstants.VAR_NAME_IS_LOOP_STATE))) {
return null;
}
//The current CompensationTriggerState can mark the compensation process is started and perform compensation route processing.
State compensationTriggerState = (State)context.getVariable(DomainConstants.VAR_NAME_CURRENT_COMPEN_TRIGGER_STATE);
if (compensationTriggerState != null) {
return compensateRoute(context, compensationTriggerState);
}
//There is an exception route, indicating that an exception is thrown, and the exception route is prioritized.
String next = (String)context.getVariable(DomainConstants.VAR_NAME_CURRENT_EXCEPTION_ROUTE);
if (StringUtils.hasLength(next)) {
context.removeVariable(DomainConstants.VAR_NAME_CURRENT_EXCEPTION_ROUTE);
} else {
//获取状态State的下一个状态名字
next = state.getNext();
}
//If next is empty, the state selected by the Choice state was taken.
if (!StringUtils.hasLength(next) && context.hasVariable(DomainConstants.VAR_NAME_CURRENT_CHOICE)) {
next = (String)context.getVariable(DomainConstants.VAR_NAME_CURRENT_CHOICE);
context.removeVariable(DomainConstants.VAR_NAME_CURRENT_CHOICE);
}
if (!StringUtils.hasLength(next)) {
return null;
}
//根据下一个状态名字获取状态State
StateMachine stateMachine = state.getStateMachine();
State nextState = stateMachine.getState(next);
if (nextState == null) {
throw new EngineExecutionException("Next state[" + next + "] is not exits", FrameworkErrorCode.ObjectNotExists);
}
//将下一个状态名字设置到状态操作组件stateInstruction中
stateInstruction.setStateName(next);
if (null != LoopTaskUtils.getLoopConfig(context, nextState)) {
stateInstruction.setTemporaryState(new LoopStartStateImpl());
}
return stateInstruction;
}
...
}