夜雨聆风
在服务A中开启全局事务，向TC申请全局事务id
服务A发起RPC远程调用，分布式事务参与方服务B和服务C分别向TC注册分支事务
参与方提交本地事务，并记录undo_log表
参与方上报分支事务的执行状态
TC根据各分支事务执行状态，通知全局事务是要提交还是回滚
分布式事务各参与方TC指示，如果是要提交全局事务，就直接删除本地表的undo_log表里边对应的记录即可，如果是需要回滚，则根据undo_log表里边的before_image对对应的数据进行回滚
要在应用中使用Seata的AT模型也非常简单，只需要简单几步：
① 引入seata依赖
② 创建本地undo_log表
③ 在配置文件中进行Seata相关的配置
④
@GlobalTransactional注解开启分布式事务
具体操作见前面的博客内容
要想开启分布式事务，前提是RM、TM持有和TC的连接，这样才能完成通信。那么这个连接是在什么时候建立的呢？首先我们看一下@GlobalTransactional注解的包结构：
可以看到这个注解定义在org.apache.seata.spring.annotation这个包下，与此同时还定义了一个以Scanner结尾的扫描器，点开看一下这个扫描器
public classGlobalTransactionScannerextendsAbstractAutoProxyCreatorimplements CachedConfigurationChangeListener, InitializingBean, ApplicationContextAware, DisposableBean {
这个扫描器实现了InitializingBean这个接口，那么就一定实现了它的afterPropertiesSet方法
@Overridepublic void afterPropertiesSet() {if (disableGlobalTransaction) {if (LOGGER.isInfoEnabled()) {LOGGER.info("Global transaction is disabled.");}ConfigurationFactory.getInstance().addConfigListener(ConfigurationKeys.DISABLE_GLOBAL_TRANSACTION, (CachedConfigurationChangeListener) this);return;}if (initialized.compareAndSet(false, true)) {initClient(); //初始化客户端，什么客户端呢？}this.findBusinessBeanNamesNeededEnhancement();}
这里有一个初始化客户端的方法，那么初始化的是什么客户端呢？继续点进去就可以看到初始化TM和RM的代码
protectedvoidinitClient() {if (LOGGER.isInfoEnabled()) {LOGGER.info("Initializing Global Transaction Clients ... ");}if (DEFAULT_TX_GROUP_OLD.equals(txServiceGroup)) {LOGGER.warn("the default value of seata.tx-service-group: {} has already changed to {} since Seata 1.5, " +"please change your default configuration as soon as possible " +"and we don't recommend you to use default tx-service-group's value provided by seata",DEFAULT_TX_GROUP_OLD, DEFAULT_TX_GROUP);}if (StringUtils.isNullOrEmpty(applicationId) || StringUtils.isNullOrEmpty(txServiceGroup)) {throw new IllegalArgumentException(String.format("applicationId: %s, txServiceGroup: %s", applicationId, txServiceGroup));}//init TMTMClient.init(applicationId, txServiceGroup, accessKey, secretKey);if (LOGGER.isInfoEnabled()) {LOGGER.info("Transaction Manager Client is initialized. applicationId[{}] txServiceGroup[{}]", applicationId, txServiceGroup);}//init RMRMClient.init(applicationId, txServiceGroup);if (LOGGER.isInfoEnabled()) {LOGGER.info("Resource Manager is initialized. applicationId[{}] txServiceGroup[{}]", applicationId, txServiceGroup);}if (LOGGER.isInfoEnabled()) {LOGGER.info("Global Transaction Clients are initialized. ");}registerSpringShutdownHook();}
以初始化RM的代码为例，继续
public class RMClient {/*** Init.** @param applicationId the application id* @param transactionServiceGroup the transaction service group*/public static void init(String applicationId, String transactionServiceGroup) {RmNettyRemotingClient rmNettyRemotingClient = RmNettyRemotingClient.getInstance(applicationId, transactionServiceGroup);rmNettyRemotingClient.setResourceManager(DefaultResourceManager.get());rmNettyRemotingClient.setTransactionMessageHandler(DefaultRMHandler.get());rmNettyRemotingClient.init();}}
继续点进init方法来到org.apache.seata.core.rpc.netty.RmNettyRemotingClient#init
@Overridepublic void init() {// registry processorregisterProcessor();if (initialized.compareAndSet(false, true)) {super.init();// Found one or more resources that were registered before initializationif (resourceManager != null&& !resourceManager.getManagedResources().isEmpty()&& StringUtils.isNotBlank(transactionServiceGroup)) {boolean failFast = ConfigurationFactory.getInstance().getBoolean(ConfigurationKeys.ENABLE_RM_CLIENT_CHANNEL_CHECK_FAIL_FAST,DefaultValues.DEFAULT_CLIENT_CHANNEL_CHECK_FAIL_FAST);getClientChannelManager().initReconnect(transactionServiceGroup, failFast);}}}
先看一下super.init()，进入org.apache.seata.core.rpc.netty.AbstractNettyRemotingClient#init
@Overridepublic void init() {timerExecutor.scheduleAtFixedRate(() -> {try {clientChannelManager.reconnect(getTransactionServiceGroup());} catch (Exception ex) {LOGGER.warn("reconnect server failed. {}", ex.getMessage());}}, SCHEDULE_DELAY_MILLS, SCHEDULE_INTERVAL_MILLS, TimeUnit.MILLISECONDS);if (this.isEnableClientBatchSendRequest()) {mergeSendExecutorService = new ThreadPoolExecutor(MAX_MERGE_SEND_THREAD,MAX_MERGE_SEND_THREAD,KEEP_ALIVE_TIME, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<>(),new NamedThreadFactory(getThreadPrefix(), MAX_MERGE_SEND_THREAD));mergeSendExecutorService.submit(new MergedSendRunnable());}super.init();clientBootstrap.start();}
这里调了一个getTransactionServiceGroup()方法，transactionServiceGroup我们之前就在配置文件中配置过，前面提到会根据这个serviceGroup去找到对应的SeataServer，不妨点clientChannelManager.reconnect() 进去看一下，看是不是根据这个serviceGroup去连接TC，点进去之后经过几个重载的方法后来到了doReconnect方法：
void doReconnect(String transactionServiceGroup, boolean failFast) {List<String> availList;try {availList = getAvailServerList(transactionServiceGroup);} catch (Exception e) {LOGGER.error("Failed to get available servers: {}", e.getMessage(), e);throwFailFastException(failFast, "Failed to get available servers");return;}if (CollectionUtils.isEmpty(availList)) {RegistryService registryService = RegistryFactory.getInstance();String clusterName = registryService.getServiceGroup(transactionServiceGroup);if (StringUtils.isBlank(clusterName)) {LOGGER.error("can not get cluster name in registry config '{}{}', please make sure registry config correct",ConfigurationKeys.SERVICE_GROUP_MAPPING_PREFIX,transactionServiceGroup);throwFailFastException(failFast, "can not get cluster name in registry config.");return;}if (!(registryService instanceof FileRegistryServiceImpl)) {LOGGER.error("no available service found in cluster '{}', please make sure registry config correct and keep your seata server running", clusterName);}throwFailFastException(failFast, "no available service found in cluster.");return;}try {doReconnect(availList, transactionServiceGroup);} catch (Exception e) {if (failFast) {throw e;}LOGGER.error("connect server failed. {}", e.getMessage(), e);}}
确实是拿到了一个可用的地址列表availList，然后根据这个列表继续调了doReconnect方法
void doReconnect(List<String> availList, String transactionServiceGroup) {Set<String> channelAddress = new HashSet<>(availList.size());Map<String, Exception> failedMap = new HashMap<>();try {for (String serverAddress : availList) {try {acquireChannel(serverAddress);channelAddress.add(serverAddress);} catch (Exception e) {failedMap.put(serverAddress, e);}}if (failedMap.size() > 0) {if (LOGGER.isInfoEnabled()) {LOGGER.error("{} can not connect to {} cause:{}", FrameworkErrorCode.NetConnect.getErrCode(),failedMap.keySet(),failedMap.values().stream().map(Throwable::getMessage).collect(Collectors.toSet()));} else if (LOGGER.isDebugEnabled()) {failedMap.forEach((key, value) -> {LOGGER.error("{} can not connect to {} cause:{} trace information:",FrameworkErrorCode.NetConnect.getErrCode(), key, value.getMessage(), value);});}}if (availList.size() == failedMap.size()) {String invalidAddress = StringUtils.join(failedMap.keySet().iterator(), ", ");throw new FrameworkException("can not connect to [" + invalidAddress + "]");}} finally {if (CollectionUtils.isNotEmpty(channelAddress)) {List<InetSocketAddress> aliveAddress = new ArrayList<>(channelAddress.size());for (String address : channelAddress) {String[] array = NetUtil.splitIPPortStr(address);aliveAddress.add(new InetSocketAddress(array[0], Integer.parseInt(array[1])));}RegistryFactory.getInstance().refreshAliveLookup(transactionServiceGroup, aliveAddress);} else {RegistryFactory.getInstance().refreshAliveLookup(transactionServiceGroup, Collections.emptyList());}}}
可以看到会根据这个列表迭代，然后根据每个serverAddress通过acquireChannel方法获取了一个Channel（如果没有现成的连接，则会通过doConnect(serverAddress)方法创建一个channel）
Channel acquireChannel(String serverAddress) {Channel channelToServer = channels.get(serverAddress);if (channelToServer != null) {channelToServer = getExistAliveChannel(channelToServer, serverAddress);if (channelToServer != null) {return channelToServer;}}if (LOGGER.isInfoEnabled()) {LOGGER.info("will connect to {}", serverAddress);}Object lockObj = CollectionUtils.computeIfAbsent(channelLocks, serverAddress, key -> new Object());synchronized (lockObj) {return doConnect(serverAddress);}}
TM的连接建立流程类似，总的来说大致的流程如下：
前面我们知道，分布式事务的开启是靠@GlobalTransactional这个注解开启的，那么肯定就有一个地方来识别这个注解并做一些事情。如果是我们自己来做的化，最先想到的肯定是基于aop来做，而从最上面的那个包结构截图可以看到，紧挨着GlobalTransactional这个注解还定义了一个AspectTransactionalInterceptor类前
public classAspectTransactionalInterceptorimplementsMethodInterceptor{既然这个类实现了MethodInterceptor接口，那程序在执行的时候肯定会调用它的invoke方法
@Overridepublic Object invoke(MethodInvocation invocation) throws Throwable {Class<?> targetClass = invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null;Method specificMethod = ClassUtils.getMostSpecificMethod(invocation.getMethod(), targetClass);InvocationWrapper invocationWrapper = new DefaultInvocationWrapper(null, invocation.getThis(), specificMethod, invocation.getArguments());return this.globalTransactionalInterceptorHandler.invoke(invocationWrapper);}
继续点进去最后一行的globalTransactionalInterceptorHandler.invoke方法，就来到了org.apache.seata.integration.tx.api.interceptor.handler.AbstractProxyInvocationHandler#invoke
@Overridepublic Object invoke(InvocationWrapper invocation) throws Throwable {if (CollectionUtils.isNotEmpty(getMethodsToProxy()) && !getMethodsToProxy().contains(invocation.getMethod().getName())) {return invocation.proceed();}if (nextInvocationHandlerChain != null) {invocation = new NestInterceptorHandlerWrapper(nextInvocationHandlerChain, ihttp://blog.shengxiao.tech/imagesnvocation);}return doInvoke(invocation);}
最终来到org.apache.seata.integration.tx.api.interceptor.handler.GlobalTransactionalInterceptorHandler#doInvoke前
@Overrideprotected Object doInvoke(InvocationWrapper invocation) throws Throwable {Class<?> targetClass = invocation.getTarget().getClass();Method specificMethod = ClassUtils.getMostSpecificMethod(invocation.getMethod(), targetClass);if (specificMethod != null && !specificMethod.getDeclaringClass().equals(Object.class)) {boolean localDisable = disable || (ATOMIC_DEGRADE_CHECK.get() && degradeNum >= degradeCheckAllowTimes);if (!localDisable) {final AspectTransactional globalTransactionalAnnotation = getAspectTransactional(specificMethod, targetClass);final GlobalLockConfig globalLockAnnotation = getGlobalLockConfig(specificMethod, targetClass);if (globalTransactionalAnnotation != null || this.aspectTransactional != null) {AspectTransactional transactional;if (globalTransactionalAnnotation != null) {transactional = globalTransactionalAnnotation;} else {transactional = this.aspectTransactional;}return handleGlobalTransaction(invocation, transactional);} else if (globalLockAnnotation != null) {return handleGlobalLock(invocation, globalLockAnnotation);}}}return invocation.proceed();}
这里边有两个分支，一个是如果globalTransactionalAnnotation或者aspectTransactional不是空的话，就进入handleGlobalTransaction，否则进入handleGlobalLock
进入org.apache.seata.integration.tx.api.interceptor.handler.GlobalTransactionalInterceptorHandler#handleGlobalTransaction，其实只有一行transactionalTemplate.execute，但是这个方法的调用传了一个匿名内部类进来，看一下这个execute方法
public Object execute(TransactionalExecutor business) throws Throwable {// 1. Get transactionInfoTransactionInfo txInfo = business.getTransactionInfo();if (txInfo == null) {throw new ShouldNeverHappenException("transactionInfo does not exist");}// 1.1 Get current transaction, if not null, the tx role is 'GlobalTransactionRole.Participant'.GlobalTransaction tx = GlobalTransactionContext.getCurrent();// 1.2 Handle the transaction propagation.Propagation propagation = txInfo.getPropagation();SuspendedResourcesHolder suspendedResourcesHolder = null;try {switch (propagation) {case NOT_SUPPORTED:// If transaction is existing, suspend it.if (existingTransaction(tx)) {suspendedResourcesHolder = tx.suspend(false);}// Execute without transaction and return.return business.execute();case REQUIRES_NEW:// If transaction is existing, suspend it, and then begin new transaction.if (existingTransaction(tx)) {suspendedResourcesHolder = tx.suspend(false);}tx = GlobalTransactionContext.createNew();// Continue and execute with new transactionbreak;case SUPPORTS:// If transaction is not existing, execute without transaction.if (notExistingTransaction(tx)) {return business.execute();}// Continue and execute with new transactionbreak;case REQUIRED:// If current transaction is existing, execute with current transaction,else createtx = GlobalTransactionContext.getCurrentOrCreate();break;case NEVER:// If transaction is existing, throw exception.if (existingTransaction(tx)) {throw new TransactionException(String.format("Existing transaction found for transaction marked with propagation 'never', xid = %s", tx.getXid()));} else {// Execute without transaction and return.return business.execute();}case MANDATORY:// If transaction is not existing, throw exception.if (notExistingTransaction(tx)) {throw new TransactionException("No existing transaction found for transaction marked with propagation 'mandatory'");}// Continue and execute with current transaction.break;default:throw new TransactionException("Not Supported Propagation:" + propagation);}// set current tx config to holderGlobalLockConfig previousConfig = replaceGlobalLockConfig(txInfo);if (tx.getGlobalTransactionRole() == GlobalTransactionRole.Participant) {LOGGER.info("join into a existing global transaction,xid={}", tx.getXid());}try {// 2. If the tx role is 'GlobalTransactionRole.Launcher', send the request of beginTransaction to TC,// else do nothing. Of course, the hooks will still be triggered.beginTransaction(txInfo, tx);Object rs;try {// Do Your Businessrs = business.execute();} catch (Throwable ex) {// 3. The needed business exception to rollback.completeTransactionAfterThrowing(txInfo, tx, ex);throw ex;}// 4. everything is fine, commit.commitTransaction(tx, txInfo);return rs;} finally {//5. clearresumeGlobalLockConfig(previousConfig);triggerAfterCompletion(tx);cleanUp(tx);}} finally {// If the transaction is suspended, resume it.if (suspendedResourcesHolder != null) {tx.resume(suspendedResourcesHolder);}}}
这里看到两个重点内容：
事务传播属性
seata自己定义了几种事务传播属性，但基本上保持和spring的事务传播属性定义一致，比如NOT_SUPPORTED的时候，将当前事务挂起，以非事务的方式执行并返回；再比如REQUIRES_NEW的时候，将当前事务挂起，然后创建一个新的事务并继续执行后续的事务和业务逻辑。
这里一共定义了6种，相比较spring少了一种
NESTED：
NOT_SUPPORTED：不支持事务，如果有事务，则挂起，以非事务的方式运行
REQUIRES_NEW：如果当前有事务，则将当前事务挂起，然后新起一个事务执行
SUPPORTS：如果当前没有事务，则以非事务的方式执行，否则就以当前事务执行
REQUIRED：如果当前没有事务，则创建一个事务执行，否则就用当前事务执行
NEVER：如果当前存在事务，则报错，如果不存在事务，就正常按照非事务的方式执行
MANDATORY：如果当前事务不存在，则报错，如果存在，就以当前事务执行
事务的处理
上面的代码可以明显看到beginTransaction、c、commitTransaction、以及最里边的catch里边的completeTransactionAfterThrowing
1、beginTransaction方法
private void beginTransaction(TransactionInfo txInfo, GlobalTransaction tx) throws TransactionalExecutor.ExecutionException {if (tx.getGlobalTransactionRole() != GlobalTransactionRole.Launcher) {if (LOGGER.isDebugEnabled()) {LOGGER.debug("Ignore begin: just involved in global transaction [{}]", tx.getXid());}return;}try {triggerBeforeBegin();tx.begin(txInfo.getTimeOut(), txInfo.getName());triggerAfterBegin();} catch (TransactionException txe) {throw new TransactionalExecutor.ExecutionException(tx, txe,TransactionalExecutor.Code.BeginFailure);}}
这个方法里边就是触发事务开启之前、事务开启、以及触发开启事务之后三个方法，主要看下tx.begin
org.apache.seata.tm.api.DefaultGlobalTransaction#begin(int, java.lang.String)
@Overridepublic void begin(int timeout, String name) throws TransactionException {this.createTime = System.currentTimeMillis();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 = transactionManager.begin(null, null, name, timeout);status = GlobalStatus.Begin;RootContext.bind(xid);if (LOGGER.isInfoEnabled()) {LOGGER.info("Begin new global transaction [{}]", xid);}}
这个方法开启了一个事务，并且得到一个xid，然后将这个xid绑定到上下文中，具体怎么开启的事务呢？其实就是执行一个远程调用org.apache.seata.tm.DefaultTransactionManager#begin
@Overridepublic String begin(String applicationId, String transactionServiceGroup, String name, int timeout)throws TransactionException {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();}
也就是这里的syncCall，会得到一个响应
public class GlobalBeginResponse extends AbstractTransactionResponse {private String xid;private String extraData;}
2、commitTransaction
这个方法是提交事务的方法，方法里边如果超时，会回滚；如果正常提交，会在提交之前触发提交前的hook，在提交之后触发提交后的hook
private void commitTransaction(GlobalTransaction tx, TransactionInfo txInfo)throws TransactionalExecutor.ExecutionException, TransactionException {if (tx.getGlobalTransactionRole() != GlobalTransactionRole.Launcher) {if (LOGGER.isDebugEnabled()) {LOGGER.debug("Ignore commit: just involved in global transaction [{}]", tx.getXid());}return;}if (isTimeout(tx.getCreateTime(), txInfo)) {// business execution timeoutException exx = new TmTransactionException(TransactionExceptionCode.TransactionTimeout,String.format("client detected transaction timeout before commit, so change to rollback, xid = %s", tx.getXid()));rollbackTransaction(tx, exx);return;}try {triggerBeforeCommit();tx.commit();GlobalStatus afterCommitStatus = tx.getLocalStatus();TransactionalExecutor.Code code = TransactionalExecutor.Code.Unknown;switch (afterCommitStatus) {case TimeoutRollbacking:code = TransactionalExecutor.Code.Rollbacking;break;case TimeoutRollbacked:code = TransactionalExecutor.Code.RollbackDone;break;case Finished:code = TransactionalExecutor.Code.CommitFailure;break;default:}Exception statusException = null;if (GlobalStatus.isTwoPhaseHeuristic(afterCommitStatus)) {statusException = new TmTransactionException(TransactionExceptionCode.CommitHeuristic,String.format("Global transaction[%s] not found, may be rollbacked.", tx.getXid()));} else if (GlobalStatus.isOnePhaseTimeout(afterCommitStatus)) {statusException = new TmTransactionException(TransactionExceptionCode.TransactionTimeout,String.format("Global transaction[%s] is timeout and will be rollback[TC].", tx.getXid()));}if (null != statusException) {throw new TransactionalExecutor.ExecutionException(tx, statusException, code);}triggerAfterCommit();} catch (TransactionException txe) {// 4.1 Failed to committhrow new TransactionalExecutor.ExecutionException(tx, txe,TransactionalExecutor.Code.CommitFailure);}}
3、completeTransactionAfterThrowing
这个方法在catch里边，也就是说当程序执行异常的时候会执行。点进去看到里边会进行事务的回滚（事务信息不为空 且 抛出的异常是指定的事务回滚异常）
private void completeTransactionAfterThrowing(TransactionInfo txInfo, GlobalTransaction tx, Throwable originalException)throws TransactionalExecutor.ExecutionException, TransactionException {//roll backif (txInfo != null && txInfo.rollbackOn(originalException)) {rollbackTransaction(tx, originalException);} else {// not roll back on this exception, so commitcommitTransaction(tx, txInfo);}}
4、business.execute
执行的业务方法
前面看到事务的”拆解“执行过程，那么RM/TM与RC之间是如何交互的呢？总的来说有异步和同步的方式。下面以开启事务方法org.apache.seata.tm.DefaultTransactionManager#begin为例
@Overridepublic String begin(String applicationId, String transactionServiceGroup, String name, int timeout)throws TransactionException {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();}
继续看syncCall方法
private AbstractTransactionResponse syncCall(AbstractTransactionRequest request) throws TransactionException {try {return (AbstractTransactionResponse) TmNettyRemotingClient.getInstance().sendSyncRequest(request);} catch (TimeoutException toe) {throw new TmTransactionException(TransactionExceptionCode.IO, "RPC timeout", toe);}}
继续org.apache.seata.core.rpc.netty.AbstractNettyRemotingClient#sendSyncRequest(java.lang.Object)方法
@Overridepublic Object sendSyncRequest(Object msg) throws TimeoutException {String serverAddress = loadBalance(getTransactionServiceGroup(), msg);long timeoutMillis = this.getRpcRequestTimeout();RpcMessage rpcMessage = buildRequestMessage(msg, ProtocolConstants.MSGTYPE_RESQUEST_SYNC);// send batch message// put message into basketMap, @see MergedSendRunnableif (this.isEnableClientBatchSendRequest()) {// send batch message is sync request, needs to create messageFuture and put it in futures.MessageFuture messageFuture = new MessageFuture();messageFuture.setRequestMessage(rpcMessage);messageFuture.setTimeout(timeoutMillis);futures.put(rpcMessage.getId(), messageFuture);// put message into basketMapBlockingQueue<RpcMessage> basket = CollectionUtils.computeIfAbsent(basketMap, serverAddress,key -> new LinkedBlockingQueue<>());if (!basket.offer(rpcMessage)) {LOGGER.error("put message into basketMap offer failed, serverAddress:{},rpcMessage:{}",serverAddress, rpcMessage);return null;}if (LOGGER.isDebugEnabled()) {LOGGER.debug("offer message: {}", rpcMessage.getBody());}if (!isSending) {synchronized (mergeLock) {mergeLock.notifyAll();}}try {Object response = messageFuture.get(timeoutMillis, TimeUnit.MILLISECONDS);return response;} catch (Exception exx) {LOGGER.error("wait response error:{},ip:{},request:{}", exx.getMessage(), serverAddress, rpcMessage.getBody());if (exx instanceof TimeoutException) {throw (TimeoutException)exx;} else {throw new RuntimeException(exx);}}} else {Channel channel = clientChannelManager.acquireChannel(serverAddress);return super.sendSync(channel, rpcMessage, timeoutMillis);}}
在这个方法里边，外边的if分支逻辑为如果开启了批量提交请求，则走if，否则，就同步调用else里边的逻辑。而在if这个逻辑里边seata的方案是将请求封装成一个org.apache.seata.core.protocol.RpcMessage对象，然后将其丢到一个阻塞队列里边，交由定时异步任务去消费处理。这个阻塞队列又以serverAddress作为key放到了一个map里边。异步任务真正获取数据是从这个map里边获取的。
这个basketMap在哪儿消费处理呢？
前面我们在看RM、TM建立与TC连接的代码的时候就看到了org.apache.seata.core.rpc.netty.AbstractNettyRemotingClient#init
@Overridepublic void init() {timerExecutor.scheduleAtFixedRate(() -> {try {clientChannelManager.reconnect(getTransactionServiceGroup());} catch (Exception ex) {LOGGER.warn("reconnect server failed. {}", ex.getMessage());}}, SCHEDULE_DELAY_MILLS, SCHEDULE_INTERVAL_MILLS, TimeUnit.MILLISECONDS);if (this.isEnableClientBatchSendRequest()) {mergeSendExecutorService = new ThreadPoolExecutor(MAX_MERGE_SEND_THREAD,MAX_MERGE_SEND_THREAD,KEEP_ALIVE_TIME, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<>(),new NamedThreadFactory(getThreadPrefix(), MAX_MERGE_SEND_THREAD));mergeSendExecutorService.submit(new MergedSendRunnable());}super.init();clientBootstrap.start();}
这里一个将一个MergedSendRunnable的任务丢到了线程池中去交由线程池调度执行。这个类的run方法如下：
@Overridepublic void run() {while (true) {synchronized (mergeLock) {try {mergeLock.wait(MAX_MERGE_SEND_MILLS);} catch (InterruptedException e) {}}isSending = true;basketMap.forEach((address, basket) -> {if (basket.isEmpty()) {return;}MergedWarpMessage mergeMessage = new MergedWarpMessage();while (!basket.isEmpty()) {RpcMessage msg = basket.poll();mergeMessage.msgs.add((AbstractMessage) msg.getBody());mergeMessage.msgIds.add(msg.getId());}if (mergeMessage.msgIds.size() > 1) {printMergeMessageLog(mergeMessage);}Channel sendChannel = null;try {// send batch message is sync request, but there is no need to get the return value.// Since the messageFuture has been created before the message is placed in basketMap,// the return value will be obtained in ClientOnResponseProcessor.sendChannel = clientChannelManager.acquireChannel(address);AbstractNettyRemotingClient.this.sendAsyncRequest(sendChannel, mergeMessage);} catch (FrameworkException e) {if (e.getErrcode() == FrameworkErrorCode.ChannelIsNotWritable && sendChannel != null) {destroyChannel(address, sendChannel);}// fast failfor (Integer msgId : mergeMessage.msgIds) {MessageFuture messageFuture = futures.remove(msgId);Integer parentId = childToParentMap.remove(msgId);if (parentId != null) {mergeMsgMap.remove(parentId);}if (messageFuture != null) {messageFuture.setResultMessage(new RuntimeException(String.format("%s is unreachable", address), e));}}LOGGER.error("client merge call failed: {}", e.getMessage(), e);}});isSending = false;}}
这个任务类的run方法的try里边就是真正发送请求的方法AbstractNettyRemotingClient.this.sendAsyncRequest(sendChannel, mergeMessage);
@Overridepublic void sendAsyncRequest(Channel channel, Object msg) {if (channel == null) {LOGGER.warn("sendAsyncRequest nothing, caused by null channel.");return;}RpcMessage rpcMessage = buildRequestMessage(msg, msg instanceof HeartbeatMessage? ProtocolConstants.MSGTYPE_HEARTBEAT_REQUEST: ProtocolConstants.MSGTYPE_RESQUEST_ONEWAY);Object body = rpcMessage.getBody();if (body instanceof MergeMessage) {Integer parentId = rpcMessage.getId();mergeMsgMap.put(parentId, (MergeMessage)rpcMessage.getBody());if (body instanceof MergedWarpMessage) {for (Integer msgId : ((MergedWarpMessage)rpcMessage.getBody()).msgIds) {childToParentMap.put(msgId, parentId);}}}super.sendAsync(channel, rpcMessage);}
