Springboot之分布式事务框架Seata实现原理源码分析

环境：springboot2.2.11 + seata1.3.0

1 准备环境

   com.alibaba.cloud 
  spring-cloud-starter-alibaba-seata    
           io.seata 
      seata-all      
    
   io.seata 
  seata-all   1.3.0

开启全局事务

seata:   service: 
    disable-global-transaction: true

2 代理数据源及注册代理Bean

@Bean @DependsOn({BEAN_NAME_SPRING_APPLICATION_CONTEXT_PROVIDER, BEAN_NAME_FAILURE_HANDLER}) 
@ConditionalOnMissingBean(GlobalTransactionScanner.class) public GlobalTransactionScanner globalTransactionScanner(SeataProperties seataProperties, FailureHandler failureHandler) { 
  if (LOGGER.isInfoEnabled()) {     LOGGER.info("Automatically configure Seata"); 
  }   return new GlobalTransactionScanner(seataProperties.getApplicationId(), seataProperties.getTxServiceGroup(),failureHandler); 
}  
@Bean(BEAN_NAME_SEATA_AUTO_DATA_SOURCE_PROXY_CREATOR) @ConditionalOnProperty(prefix = StarterConstants.SEATA_PREFIX, name = {"enableAutoDataSourceProxy", "enable-auto-data-source-proxy"}, havingValue = "true", matchIfMissing = true) 
@ConditionalOnMissingBean(SeataAutoDataSourceProxyCreator.class) public SeataAutoDataSourceProxyCreator seataAutoDataSourceProxyCreator(SeataProperties seataProperties) { 
  return new SeataAutoDataSourceProxyCreator(seataProperties.isUseJdkProxy(),seataProperties.getExcludesForAutoProxying()); }

2.1 创建代理Bean

Seata通过GlobalTransactionScanner来注册我们项目中所有带有@GlobalTransactional注解的方法类。

public class GlobalTransactionScanner extends AbstractAutoProxyCreator implements InitializingBean, ApplicationContextAware, DisposableBean

AbstractAutoProxyCreator继承层次

从这里也知道GlobalTransactionScanner类其实是一个BeanPostProcessor处理器。

InstantiationAwareBeanPostProcessor类有如下3个方法是很有用的

postProcessBeforeInstantiation 实例化前执行

postProcessAfterInstantiation 实例化之后执行

postProcessProperties 属性填充时执行

当然在这里GlobalTransactionScanner类并没有覆盖这3个方法。

BeanPostProcessor相关的2个方法在父类中有实现

@Override public Object postProcessBeforeInitialization(Object bean, String beanName) { 
  return bean; } 
@Override public Object postProcessAfterInitialization(@Nullable Object bean, String beanName) { 
  if (bean != null) {     Object cacheKey = getCacheKey(bean.getClass(), beanName); 
    if (this.earlyProxyReferences.remove(cacheKey) != bean) {       return wrapIfNecessary(bean, beanName, cacheKey); 
    }   } 
  return bean; }

在实例化Bean的时候会执行父类中

postProcessAfterInitialization方法。关键是该方法中的wrapIfNecessary方法，该方法在GlobalTransactionScanner类中被重写了。

existsAnnotation 方法判断当前的类方法上是否有@GlobalTransactional注解。如果不存在会直接返回当前Bean。

interceptor 判断当前拦截器是否为空，为空创建

GlobalTransactionalInterceptor该拦截器处理全局事务的地方。

if (!AopUtils.isAopProxy(bean)) {   bean = super.wrapIfNecessary(bean, beanName, cacheKey); 
} else {   AdvisedSupport advised = SpringProxyUtils.getAdvisedSupport(bean); 
  Advisor[] advisor = buildAdvisors(beanName, getAdvicesAndAdvisorsForBean(null, null, null));   for (Advisor avr : advisor) { 
    advised.addAdvisor(0, avr);   } 
}

该片段代码，判断当前的Bean是否是代理类(JDK或CGLIB)，如果不是那么会先的执行下父类的wrapIfNecessary方法。

如果当前Bean是代理类对象，那么会获取当前代理类的AdvisedSupport(内部维护了切面类的集合)对象。这里可以看看JDK和CGLIB两种方式创建的代理类对象是否都具有AdvisedSupport对象。

public static AdvisedSupport getAdvisedSupport(Object proxy) throws Exception {   Field h; 
  if (AopUtils.isJdkDynamicProxy(proxy)) {     h = proxy.getClass().getSuperclass().getDeclaredField("h"); 
  } else {     h = proxy.getClass().getDeclaredField("CGLIB$CALLBACK_0"); 
  }   h.setAccessible(true); 
  Object dynamicAdvisedInterceptor = h.get(proxy);   Field advised = dynamicAdvisedInterceptor.getClass().getDeclaredField("advised"); 
  advised.setAccessible(true);   return (AdvisedSupport)advised.get(dynamicAdvisedInterceptor); 
}

jdk创建代理对象时使用的InvocationHandler

final class JdkDynamicAopProxy implements AopProxy, InvocationHandler, Serializable {  
     private static final Log logger = LogFactory.getLog(JdkDynamicAopProxy.class); 
    
  private final AdvisedSupport advised; }

cglib 获取CGLIB$CALLBACK_0字段，该字段是MethodInterceptor对象

public class PersonDAOImpl$$EnhancerBySpringCGLIB$$d4658dad extends PersonDAOImpl implements SpringProxy, Advised, Factory{   private boolean CGLIB$BOUND; 
  public static Object CGLIB$FACTORY_DATA;   private static final ThreadLocal CGLIB$THREAD_CALLBACKS; 
  private static final Callback[] CGLIB$STATIC_CALLBACKS;   private MethodInterceptor CGLIB$CALLBACK_0; 
}

接下来就是将Seata的拦截器添加到AdvisedSupport中。

Advisor[] advisor = buildAdvisors(beanName, getAdvicesAndAdvisorsForBean(null, null, null)); for (Advisor avr : advisor) { 
  advised.addAdvisor(0, avr); } 
protected Object[] getAdvicesAndAdvisorsForBean(Class beanClass, String beanName, TargetSource customTargetSource) throws BeansException {   return new Object[]{interceptor}; 
}

到此就将Seata的方法拦截器包装成Advisor切面添加到了当前的AdvisedSupport管理的切面集合中。

2.2 创建代理数据源

对数据源上的方法调用进行代理处理通过DataSourceProxy

public class SeataAutoDataSourceProxyCreator extends AbstractAutoProxyCreator {   private static final Logger LOGGER = LoggerFactory.getLogger(SeataAutoDataSourceProxyCreator.class); 
  private final String[] excludes;   private final Advisor advisor = new DefaultIntroductionAdvisor(new SeataAutoDataSourceProxyAdvice()); 
   public SeataAutoDataSourceProxyCreator(boolean useJdkProxy, String[] excludes) { 
    this.excludes = excludes;     setProxyTargetClass(!useJdkProxy); 
  }  
  @Override   protected Object[] getAdvicesAndAdvisorsForBean(Class beanClass, String beanName, TargetSource customTargetSource) throws BeansException { 
    if (LOGGER.isInfoEnabled()) {       LOGGER.info("Auto proxy of [{}]", beanName); 
    }     return new Object[]{advisor}; 
  }  
  @Override   protected boolean shouldSkip(Class beanClass, String beanName) { 
    return SeataProxy.class.isAssignableFrom(beanClass) || !DataSource.class.isAssignableFrom(beanClass) ||                   Arrays.asList(excludes).contains(beanClass.getName());     } 
}

shouldSkip 该方法确定了如果当前beanClass是SeataProxy的子类并且beanClass不是DataSource的子类或者当前的bean名称不再excludes集合中就会进行代理。简单说就是代理当前系统的默认数据源对象。

getAdvicesAndAdvisorsForBean 方法直接返回DefaultIntroductionAdvisor切面，通知类是SeataAutoDataSourceProxyAdvice

public class SeataAutoDataSourceProxyAdvice implements MethodInterceptor, IntroductionInfo {  
  @Override   public Object invoke(MethodInvocation invocation) throws Throwable { 
    DataSourceProxy dataSourceProxy = DataSourceProxyHolder.get().putDataSource((DataSource) invocation.getThis());     Method method = invocation.getMethod(); 
    Object[] args = invocation.getArguments();     Method m = BeanUtils.findDeclaredMethod(DataSourceProxy.class, method.getName(), method.getParameterTypes()); 
    if (m != null) {       return m.invoke(dataSourceProxy, args); 
    } else {       return invocation.proceed(); 
    }   } 
   @Override 
  public Class[] getInterfaces() {     return new Class[]{SeataProxy.class}; 
  }  
}

3 全局事务拦截器

在需要进行发起全局事务的方法是被代理的具体执行的拦截器是

GlobalTransactionalInterceptor

handleGlobalTransaction方法

通过事务模版执行，TransactionalExecutor类进行收集当前@GlobalTransactional注解上配置的相关信息封装到TransactionInfo中。

TransactionalTemplate.execute方法

// 该方法中根据不同的事务传播特性进行不同的处理。 public Object execute(TransactionalExecutor business) throws Throwable { 
    // 1 get transactionInfo     TransactionInfo txInfo = business.getTransactionInfo(); 
    if (txInfo == null) {         throw new ShouldNeverHappenException("transactionInfo does not exist"); 
    }     // 1.1 get or create a transaction 
    GlobalTransaction tx = GlobalTransactionContext.getCurrentOrCreate();  
    // 1.2 Handle the Transaction propatation and the branchType     Propagation propagation = txInfo.getPropagation(); 
    SuspendedResourcesHolder suspendedResourcesHolder = null;     try { 
        switch (propagation) {             case NOT_SUPPORTED: 
                suspendedResourcesHolder = tx.suspend(true);                 return business.execute(); 
            case REQUIRES_NEW:                 suspendedResourcesHolder = tx.suspend(true); 
                break;             case SUPPORTS: 
                if (!existingTransaction()) {                     return business.execute(); 
                }                 break; 
            case REQUIRED:                 break; 
            case NEVER:                 // 存在事务抛出异常 
                if (existingTransaction()) {                     throw new TransactionException( 
                        String.format("Existing transaction found for transaction marked with propagation 'never',xid = %s"                                       ,RootContext.getXID())); 
                } else {                     // 直接执行业务代码 
                    return business.execute();                 } 
            case MANDATORY:                 if (!existingTransaction()) { 
                    throw new TransactionException("No existing transaction found for transaction marked with propagation 'mandatory'");                 } 
                break;             default: 
                throw new TransactionException("Not Supported Propagation:" + propagation);         } 
         try { 
             // 2. 开始事务 
            beginTransaction(txInfo, tx);  
            Object rs = null;             try { 
                 // 执行我们的业务代码 
                rs = business.execute();  
            } catch (Throwable ex) {  
                // 3.the needed business exception to rollback.                 completeTransactionAfterThrowing(txInfo, tx, ex); 
                throw ex;             } 
             // 4. 一切正常提交事务。 
            commitTransaction(tx);  
            return rs;         } finally { 
            //5. clear             triggerAfterCompletion(); 
            cleanUp();         } 
    } finally {         tx.resume(suspendedResourcesHolder); 
    }  
}

3.1 获取全局事务

GlobalTransaction tx = GlobalTransactionContext.getCurrentOrCreate(); public static GlobalTransaction getCurrentOrCreate() { 
  // 首次这里会返回null，执行createNew方法     GlobalTransaction tx = getCurrent(); 
  if (tx == null) {     return createNew(); 
  }   return tx; 
} // 获取全局事务对象 
private static GlobalTransaction getCurrent() {   String xid = RootContext.getXID(); 
  if (xid == null) {     return null; 
  }   return new DefaultGlobalTransaction(xid, GlobalStatus.Begin, GlobalTransactionRole.Participant); 
} private static GlobalTransaction createNew() { 
  return new DefaultGlobalTransaction(); }

3.2 开始全局事务

beginTransaction(txInfo, tx); private void beginTransaction(TransactionInfo txInfo, GlobalTransaction tx) throws TransactionalExecutor.ExecutionException { 
  try {     triggerBeforeBegin(); 
    tx.begin(txInfo.getTimeOut(), txInfo.getName());     triggerAfterBegin(); 
  } catch (TransactionException txe) {     throw new TransactionalExecutor.ExecutionException(tx, txe, TransactionalExecutor.Code.BeginFailure); 
  } } 
// 开始全局事务，并且通过TC获取全局事务唯一ID  xid public void begin(int timeout, String name) throws TransactionException { 
  // 全局事务的开启必须是Launcher     if (role != GlobalTransactionRole.Launcher) { 
    assertXIDNotNull();     if (LOGGER.isDebugEnabled()) { 
      LOGGER.debug("Ignore Begin(): just involved in global transaction [{}]", xid);     } 
    return;   } 
  assertXIDNull();   if (RootContext.getXID() != null) { 
    throw new IllegalStateException();   } 
  xid = transactionManager.begin(null, null, name, timeout);   status = GlobalStatus.Begin; 
  // 将当前获取到的xid绑定到当前thread上（ThreadLocal）     RootContext.bind(xid); 
  if (LOGGER.isInfoEnabled()) {     LOGGER.info("Begin new global transaction [{}]", xid); 
  } }

将xid绑定到当前执行thread(ThreadLocal)在这里seata是通过SPI技术来实现的

private static ContextCore CONTEXT_HOLDER = ContextCoreLoader.load(); public static void bind(String xid) { 
  if (LOGGER.isDebugEnabled()) {     LOGGER.debug("bind {}", xid); 
  }   CONTEXT_HOLDER.put(KEY_XID, xid); 
} //通过SPI加载具体的ContextCore实现 
public class ContextCoreLoader {  
  private ContextCoreLoader() {   } 
   private static class ContextCoreHolder { 
    private static final ContextCore INSTANCE = Optional.ofNullable(EnhancedServiceLoader.load(ContextCore.class)).orElse(new ThreadLocalContextCore());   } 
  public static ContextCore load() {     return ContextCoreHolder.INSTANCE; 
  }  
} // meta-INF/services/io.seata.core.context.ContextCore 文件内容 
io.seata.core.context.ThreadLocalContextCore io.seata.core.context.FastThreadLocalContextCore

3.3 执行本地业务

执行本地事务时会向TC注册分支然后提交本地事务，接下来看看本地分支事务的注册及处理。

rs = business.execute();

3.3.1 提交本地事务

执行完业务代码后提交事务ConnectionProxy.commit()

@Override public void commit() throws SQLException { 
  try {     LOCK_RETRY_POLICY.execute(() -> { 
      // 事务提交         doCommit(); 
      return null;     }); 
  } catch (SQLException e) {     throw e; 
  } catch (Exception e) {     throw new SQLException(e); 
  } } 
private void doCommit() throws SQLException {   // 判断当前是否在全局事务中(xid != null)   
  if (context.inGlobalTransaction()) {     // 处理全局事务   
    processGlobalTransactionCommit();   } else if (context.isGlobalLockRequire()) { 
    processLocalCommitWithGlobalLocks();   } else { 
    targetConnection.commit();   } 
}

进入

processGlobalTransactionCommit方法

3.3.2 注册本次事务分支

private void processGlobalTransactionCommit() throws SQLException {   try { 
    register();   } catch (TransactionException e) { 
    recognizeLockKeyConflictException(e, context.buildLockKeys());   } 
  try {     UndoLogManagerFactory.getUndoLogManager(this.getDbType()).flushUndoLogs(this); 
    targetConnection.commit();   } catch (Throwable ex) { 
    report(false);     throw new SQLException(ex); 
  }   if (IS_REPORT_SUCCESS_ENABLE) { 
    report(true);   } 
  context.reset(); } 
 // 注册本次事务执行RM，将返回的branchId保存到当前的上下文中 
private void register() throws TransactionException {   if (!context.hasUndoLog() || context.getLockKeysBuffer().isEmpty()) { 
    return;   } 
  Long branchId = DefaultResourceManager.get().branchRegister(BranchType.AT, getDataSourceProxy().getResourceId(), null, context.getXid(), null, context.buildLockKeys());   // 将注册RM返回的branchId绑定到当前的上下文中ConnectionContext   
  context.setBranchId(branchId); }

进入branchRegister方法

DefaultResourceManager.java

@Override public Long branchRegister(BranchType branchType, String resourceId, String clientId, String xid, String applicationData, String lockKeys) 
  throws TransactionException {   return getResourceManager(branchType).branchRegister(branchType, resourceId, clientId, xid, applicationData, lockKeys); 
}

@Override public Long branchRegister(BranchType branchType, String resourceId, String clientId, String xid, String applicationData, String lockKeys) throws TransactionException { 
  try {     BranchRegisterRequest request = new BranchRegisterRequest(); 
    request.setXid(xid);     request.setLockKey(lockKeys); 
    request.setResourceId(resourceId);     request.setBranchType(branchType); 
    request.setApplicationData(applicationData);  
    BranchRegisterResponse response = (BranchRegisterResponse) RmNettyRemotingClient.getInstance().sendSyncRequest(request);     if (response.getResultCode() == ResultCode.Failed) { 
      throw new RmTransactionException(response.getTransactionExceptionCode(), String.format("Response[ %s ]", response.getMsg()));     } 
    return response.getBranchId();   } catch (TimeoutException toe) { 
    throw new RmTransactionException(TransactionExceptionCode.IO, "RPC Timeout", toe);   } catch (RuntimeException rex) { 
    throw new RmTransactionException(TransactionExceptionCode.BranchRegisterFailed, "Runtime", rex);   } 
}

3.3.3 记录undo log日志

undo log主要记录了数据的逻辑变化，比如一条 INSERT 语句，对应一条DELETE 的 undo log ，对于每个 UPDATE 语句，对应一条相反的 UPDATE 的 undo log ，这样在发生错误时，就能回滚到事务之前的数据状态。

UndoLogManagerFactory.getUndoLogManager(this.getDbType()).flushUndoLogs(this); // 记录undo log日志 
public void flushUndoLogs(ConnectionProxy cp) throws SQLException {   ConnectionContext connectionContext = cp.getContext(); 
  if (!connectionContext.hasUndoLog()) {     return; 
  }  
  String xid = connectionContext.getXid();   long branchId = connectionContext.getBranchId(); 
   BranchUndoLog branchUndoLog = new BranchUndoLog(); 
  branchUndoLog.setXid(xid);   branchUndoLog.setBranchId(branchId); 
  branchUndoLog.setSqlUndoLogs(connectionContext.getUndoItems());  
  UndoLogParser parser = UndoLogParserFactory.getInstance();   byte[] undoLogContent = parser.encode(branchUndoLog); 
   if (LOGGER.isDebugEnabled()) { 
    LOGGER.debug("Flushing UNDO LOG: {}", new String(undoLogContent, Constants.DEFAULT_CHARSET));   } 
   // 将undo log日志出入到undo_log表中   
  insertUndoLogWithNormal(xid, branchId, buildContext(parser.getName()), undoLogContent, cp.getTargetConnection()); } 
// 这里会根据是Oracle或MySQL自动执行；我当前环境使用的MySQL，所以使用的是MySQLUndoLogManager @Override 
protected void insertUndoLogWithNormal(String xid, long branchId, String rollbackCtx,                                        byte[] undoLogContent, Connection conn) throws SQLException { 
    insertUndoLog(xid, branchId, rollbackCtx, undoLogContent, State.Normal, conn); } 
private void insertUndoLog(String xid, long branchId, String rollbackCtx, byte[] undoLogContent, State state, Connection conn) throws SQLException {     try (PreparedStatement pst = conn.prepareStatement(INSERT_UNDO_LOG_SQL)) { 
        pst.setLong(1, branchId);         pst.setString(2, xid); 
        pst.setString(3, rollbackCtx);         pst.setBlob(4, BlobUtils.bytes2Blob(undoLogContent)); 
        pst.setInt(5, state.getValue());         pst.executeUpdate(); 
    } catch (Exception e) {         if (!(e instanceof SQLException)) { 
            e = new SQLException(e);         } 
        throw (SQLException) e;     } 
}

3.3.4 本地事务提交

业务数据的更新和前面步骤中生成的 UNDO LOG 一并提交。

targetConnection.commit();

3.3.5 重置当前上下文

将当前ConnectionProxy中的ConnectionContext重置

context.reset(); // 重置 
void reset(String xid) {   this.xid = xid; 
  branchId = null;   this.isGlobalLockRequire = false; 
  lockKeysBuffer.clear();   sqlUndoItemsBuffer.clear(); 
}

到此整个全局事务的第一阶段完成了(通过feign的调用也成功返回);接下来是第二阶段的提交。

3.4 全局事物提交

// TransactionalTemplate.java commitTransaction(tx); 
private void commitTransaction(GlobalTransaction tx) throws TransactionalExecutor.ExecutionException {   try { 
    triggerBeforeCommit();     tx.commit(); 
    triggerAfterCommit();   } catch (TransactionException txe) { 
    // 4.1 Failed to commit     throw new TransactionalExecutor.ExecutionException(tx, txe, TransactionalExecutor.Code.CommitFailure); 
  } } 
// DefaultGlobalTransaction.java public void commit() throws TransactionException { 
  int retry = COMMIT_RETRY_COUNT <= 0 ? DEFAULT_TM_COMMIT_RETRY_COUNT : COMMIT_RETRY_COUNT;   try { 
    while (retry > 0) {       try { 
        // 根据当前的全局唯一事务id xid提交事务。         status = transactionManager.commit(xid); 
        break;       } catch (Throwable ex) { 
        retry--;           if (retry == 0) { 
            throw new TransactionException("Failed to report global commit", ex);           } 
      }     } 
  } finally {     if (RootContext.getXID() != null && xid.equals(RootContext.getXID())) { 
      suspend(true);     } 
  } } 
// DefaultTransactionManager.java @Override 
public GlobalStatus commit(String xid) throws TransactionException {   GlobalCommitRequest globalCommit = new GlobalCommitRequest(); 
  globalCommit.setXid(xid);   GlobalCommitResponse response = (GlobalCommitResponse) syncCall(globalCommit); 
  return response.getGlobalStatus(); } 
// 通过Netty同步调用 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); 
  } }

到此第二阶段的事务就提交完成了。

3.5 全局事务回滚

参与全局事务的任何一个分支发生异常将对整个事务进行回滚。

3.5.1 全局事务发起端异常

代码片段

try {   // Do Your Business 
  rs = business.execute(); } catch (Throwable ex) { 
  // 3.the needed business exception to rollback.   completeTransactionAfterThrowing(txInfo, tx, ex); 
  throw ex; }

当本地业务执行时发生异常后执行

completeTransactionAfterThrowing方法。

private void completeTransactionAfterThrowing(TransactionInfo txInfo, GlobalTransaction tx, Throwable originalException) throws TransactionalExecutor.ExecutionException {   //roll back 
  if (txInfo != null && txInfo.rollbackOn(originalException)) {     try { 
      // 全局事务回滚         rollbackTransaction(tx, originalException); 
    } catch (TransactionException txe) {       // Failed to rollback 
      throw new TransactionalExecutor.ExecutionException(tx, txe, TransactionalExecutor.Code.RollbackFailure, originalException);     } 
  } else {     // not roll back on this exception, so commit 
    commitTransaction(tx);   } 
}

进入rollbackTransaction方法。

private void rollbackTransaction(GlobalTransaction tx, Throwable originalException) throws TransactionException, TransactionalExecutor.ExecutionException {   triggerBeforeRollback(); 
  tx.rollback();   triggerAfterRollback(); 
  // 3.1 Successfully rolled back   throw new TransactionalExecutor.ExecutionException(tx, GlobalStatus.RollbackRetrying.equals(tx.getLocalStatus()) ? TransactionalExecutor.Code.RollbackRetrying : TransactionalExecutor.Code.RollbackDone, originalException); 
}

进入rollback方法。

public void rollback() throws TransactionException {   // Participant(参与者)，如果当前是参与者那么直接返回，全局事务的回滚必须是Launcher 
  if (role == GlobalTransactionRole.Participant) {     return; 
  }   assertXIDNotNull(); 
  int retry = ROLLBACK_RETRY_COUNT <= 0 ? DEFAULT_TM_ROLLBACK_RETRY_COUNT : ROLLBACK_RETRY_COUNT;   try { 
    while (retry > 0) {       try { 
        status = transactionManager.rollback(xid);         break; 
      } catch (Throwable ex) {         retry--; 
        if (retry == 0) {           throw new TransactionException("Failed to report global rollback", ex); 
        }       } 
    }   } finally { 
    if (RootContext.getXID() != null && xid.equals(RootContext.getXID())) {       suspend(true); 
    }   } 
}

进入

transactionManager.rollback方法

public GlobalStatus rollback(String xid) throws TransactionException {   GlobalRollbackRequest globalRollback = new GlobalRollbackRequest(); 
  globalRollback.setXid(xid);   GlobalRollbackResponse response = (GlobalRollbackResponse) syncCall(globalRollback); 
  return response.getGlobalStatus(); }

到此全局事务就进行了回滚

3.5.2 全局事务参与者异常 4 XID的传递 4.1 RestTemplate

@Configuration(proxyBeanMethods = false) public class SeataRestTemplateAutoConfiguration { 
   // 该拦截器的作用就是为请求的Header中传递XID 
  @Bean   public SeataRestTemplateInterceptor seataRestTemplateInterceptor() { 
    return new SeataRestTemplateInterceptor();   } 
  // 获取当前IOC容器中所有的的RestTemplate对象   @Autowired(required = false) 
  private Collection restTemplates;  
  @Autowired   private SeataRestTemplateInterceptor seataRestTemplateInterceptor; 
   // 当前这个Bean（SeataRestTemplateAutoConfiguration）被创建且相关属性被注入后执行 
  @PostConstruct   public void init() { 
    if (this.restTemplates != null) {       // 为所有的RestTemplate设置一个拦截器。SeataRestTemplateInterceptor 
      for (RestTemplate restTemplate : restTemplates) {         List interceptors = new ArrayList(restTemplate.getInterceptors()); 
        interceptors.add(this.seataRestTemplateInterceptor);         restTemplate.setInterceptors(interceptors); 
      }     } 
  } }

public class SeataRestTemplateInterceptor implements ClientHttpRequestInterceptor {  
  @Override   public ClientHttpResponse intercept(HttpRequest httpRequest, byte[] bytes, ClientHttpRequestExecution clientHttpRequestExecution) throws IOException { 
    HttpRequestWrapper requestWrapper = new HttpRequestWrapper(httpRequest);  
    String xid = RootContext.getXID();  
    if (!StringUtils.isEmpty(xid)) {       requestWrapper.getHeaders().add(RootContext.KEY_XID, xid); 
    }     return clientHttpRequestExecution.execute(requestWrapper, bytes); 
  }  
}

该拦截器的作用就是为当前的请求header中放置TX_XID头信息。(是不是有点过分了，所有的RestTemplate都添加这个Header;应该像@LoadBalanced一样只有添加有该注解的才具有负载均衡的作用)。

到此RestTemplate调用方式传递XID值信息就这么简单。

4.2 Feign

@Configuration(proxyBeanMethods = false) @ConditionalOnClass(Client.class) 
@AutoConfigureBefore(FeignAutoConfiguration.class) public class SeataFeignClientAutoConfiguration { 
  @Bean   @Scope("prototype") 
  @ConditionalOnClass(name = "com.netflix.hystrix.HystrixCommand")   @ConditionalOnProperty(name = "feign.hystrix.enabled", havingValue = "true") 
  Feign.Builder feignHystrixBuilder(BeanFactory beanFactory) {     return SeataHystrixFeignBuilder.builder(beanFactory); 
  } }

每一个Feign客户端是一个FeignClientFactoryBean 工厂Bean。当在调用接口的时候会执行getObject方法

@Override public Object getObject() throws Exception { 
  return getTarget(); } 
  T getTarget() { 
  FeignContext context = this.applicationContext.getBean(FeignContext.class);   // 这个方法会从当前的IOC容器中获取Feign.Builder对象   
  Feign.Builder builder = feign(context);  
  if (!StringUtils.hasText(this.url)) {     if (!this.name.startsWith("http")) { 
      this.url = "http://" + this.name;     } else { 
      this.url = this.name;     } 
    this.url += cleanPath();     // 负载均衡调用目标服务（通过服务发现调用）   
    return (T) loadBalance(builder, context, new HardCodedTarget<>(this.type, this.name, this.url));   } 
  // 下面是通过配置的url直接调用目标服务。     if (StringUtils.hasText(this.url) && !this.url.startsWith("http")) { 
    this.url = "http://" + this.url;   } 
  String url = this.url + cleanPath();   Client client = getOptional(context, Client.class); 
  if (client != null) {     if (client instanceof LoadBalancerFeignClient) { 
      client = ((LoadBalancerFeignClient) client).getDelegate();     } 
    if (client instanceof FeignBlockingLoadBalancerClient) {       client = ((FeignBlockingLoadBalancerClient) client).getDelegate(); 
    }     builder.client(client); 
  }   Targeter targeter = get(context, Targeter.class); 
  return (T) targeter.target(this, builder, context, new HardCodedTarget<>(this.type, this.name, url)); }

feign(content)方法在调用链中最终执行如下方法从IOC容器中获取Feign.Builder

public  T getInstance(String name, Class type) {   AnnotationConfigApplicationContext context = getContext(name); 
  if (BeanFactoryUtils.beanNamesForTypeIncludingAncestors(context,type).length > 0) {     return context.getBean(type); 
  }   return null; 
}

接下来进入

SeataHystrixFeignBuilder.builder(beanFactory);方法

static Feign.Builder builder(BeanFactory beanFactory) {   return HystrixFeign.builder().retryer(Retryer.NEVER_RETRY).client(new SeataFeignClient(beanFactory)); 
}

SeataFeignClient类

// Feign的执行就通过Client接口调用。 public class SeataFeignClient implements Client { 
   private final Client delegate; 
   private final BeanFactory beanFactory; 
   private static final int MAP_SIZE = 16; 
   SeataFeignClient(BeanFactory beanFactory) { 
    this.beanFactory = beanFactory;     this.delegate = new Client.Default(null, null); 
  }  
  SeataFeignClient(BeanFactory beanFactory, Client delegate) {     this.delegate = delegate; 
    this.beanFactory = beanFactory;   } 
   @Override 
  public Response execute(Request request, Request.Options options) throws IOException {     Request modifiedRequest = getModifyRequest(request); 
    return this.delegate.execute(modifiedRequest, options);   } 
   // 该方法给请求headers中添加TX_XID请求header信息。  
  private Request getModifyRequest(Request request) {     String xid = RootContext.getXID(); 
    if (StringUtils.isEmpty(xid)) {       return request; 
    }  
    Map> headers = new HashMap<>(MAP_SIZE);     headers.putAll(request.headers()); 
     List seataXid = new ArrayList<>(); 
    seataXid.add(xid);     headers.put(RootContext.KEY_XID, seataXid); 
    return Request.create(request.method(), request.url(), headers, request.body(), request.charset());   } 
 }

5 参与者如何加入全局事务

在被调用端(通过Feign调用服务)接口服务上没有加入任何注解或是特殊的代码那它又是如何加入到整个全局事务中的呢?

在2.2中介绍了seata自动配置会为我们自动的创建数据源代理。就是通过这个代理数据源来完成的DataSourceProxy。

事务方法在执行时都会先拿到Connection对象，这里系统默认的DataSource已经被代理成DataSourceProxy。

5.1 参与者获取XID

SeataHandlerInterceptorConfiguration注册一个拦截器SeataHandlerInterceptor;SeataHandlerInterceptor拦截器对我们的所有请求进行拦截

public class SeataHandlerInterceptorConfiguration implements WebMvcConfigurer {  
  @Override   public void addInterceptors(InterceptorRegistry registry) { 
    registry.addInterceptor(new SeataHandlerInterceptor()).addPathPatterns("/**");   } 
 }

拦截器从Header中获取TX_XID

public class SeataHandlerInterceptor implements HandlerInterceptor {   @Override 
  public boolean preHandle(HttpServletRequest request, HttpServletResponse response,Object handler) {     String xid = RootContext.getXID(); 
    // 从Header中获取TX_XID信息。如果存在就绑定到RootContext上下文中。       String rpcXid = request.getHeader(RootContext.KEY_XID); 
    if (StringUtils.isBlank(xid) && rpcXid != null) {       RootContext.bind(rpcXid); 
    }     return true; 
  }   @Override 
  public void afterCompletion(HttpServletRequest request, HttpServletResponse response,Object handler, Exception e) {     if (StringUtils.isNotBlank(RootContext.getXID())) { 
      String rpcXid = request.getHeader(RootContext.KEY_XID);       if (StringUtils.isEmpty(rpcXid)) { 
        return;       } 
      // 执行完后解除绑定         String unbindXid = RootContext.unbind(); 
      if (!rpcXid.equalsIgnoreCase(unbindXid)) {         if (unbindXid != null) { 
          RootContext.bind(unbindXid);         } 
      }     } 
  } }

参与者通过拦截器的方式将xid拿到并且绑定到上下文中。

5.2 获取代理连接对象

一个数据库操作执行

DataSourceProxy.getConnection方法获取ConnectionProxy对象。

@Override public ConnectionProxy getConnection() throws SQLException { 
  Connection targetConnection = targetDataSource.getConnection();   return new ConnectionProxy(this, targetConnection); 
}

用ConnectionProxy代理默认数据源的的Connection对象。

在ConnectionProxy对象中有个非常重要的属性

public class ConnectionProxy extends AbstractConnectionProxy {  
  private ConnectionContext context = new ConnectionContext();      
}

在一个数据库操作做最后事务提交的时候会通过ConnectionContext对象来判断是否是全局事务xid是否为空。

5.3 绑定XID到ConnectionContext

由于事务在提交的时候需要从ConnectionContext中获取判断是否全局事务(xid是否为空);xid是在Statement执行时进行绑定的。

执行相关SQL语句是通过StatementProxy, PreparedStatementProxy，这两个对象都是通过ConnectionProxy获取。

public class PreparedStatementProxy extends AbstractPreparedStatementProxy implements PreparedStatement, ParametersHolder {  
  @Override   public Map> getParameters() { 
    return parameters;   } 
   public PreparedStatementProxy(AbstractConnectionProxy connectionProxy, PreparedStatement targetStatement, String targetSQL) throws SQLException { 
    super(connectionProxy, targetStatement, targetSQL);   } 
   @Override 
  public boolean execute() throws SQLException {     return ExecuteTemplate.execute(this, (statement, args) -> statement.execute()); 
  }  
  @Override   public ResultSet executeQuery() throws SQLException { 
    return ExecuteTemplate.execute(this, (statement, args) -> statement.executeQuery());   } 
   @Override 
  public int executeUpdate() throws SQLException {     return ExecuteTemplate.execute(this, (statement, args) -> statement.executeUpdate()); 
  } }

查看executeUpdate的执行

public static  T execute(List sqlRecognizers,                                                      StatementProxy statementProxy, 
                                                     StatementCallback statementCallback,                                                      Object... args) throws SQLException { 
    if (!RootContext.requireGlobalLock() && !StringUtils.equals(BranchType.AT.name(), RootContext.getBranchType())) {         // Just work as original statement 
        return statementCallback.execute(statementProxy.getTargetStatement(), args);     } 
     String dbType = statementProxy.getConnectionProxy().getDbType(); 
    if (CollectionUtils.isEmpty(sqlRecognizers)) {         sqlRecognizers = SQLVisitorFactory.get( 
            statementProxy.getTargetSQL(),             dbType); 
    }     Executor executor; 
    if (CollectionUtils.isEmpty(sqlRecognizers)) {         executor = new PlainExecutor<>(statementProxy, statementCallback); 
    } else {         if (sqlRecognizers.size() == 1) { 
            SQLRecognizer sqlRecognizer = sqlRecognizers.get(0);             switch (sqlRecognizer.getSQLType()) { 
                case INSERT:                     executor = EnhancedServiceLoader.load(InsertExecutor.class, dbType, 
                                                          new Class[]{StatementProxy.class, StatementCallback.class, SQLRecognizer.class},                                                           new Object[]{statementProxy, statementCallback, sqlRecognizer}); 
                    break;                 case UPDATE: 
                    executor = new UpdateExecutor<>(statementProxy, statementCallback, sqlRecognizer);                     break; 
                case DELETE:                     executor = new DeleteExecutor<>(statementProxy, statementCallback, sqlRecognizer); 
                    break;                 case SELECT_FOR_UPDATE: 
                    executor = new SelectForUpdateExecutor<>(statementProxy, statementCallback, sqlRecognizer);                     break; 
                default:                     executor = new PlainExecutor<>(statementProxy, statementCallback); 
                    break;             } 
        } else {             executor = new MultiExecutor<>(statementProxy, statementCallback, sqlRecognizers); 
        }     } 
    T rs;     try { 
        rs = executor.execute(args);     } catch (Throwable ex) { 
        if (!(ex instanceof SQLException)) {             // Turn other exception into SQLException 
            ex = new SQLException(ex);         } 
        throw (SQLException) ex;     } 
    return rs; }

这里的操作UpdateExecutor，DeleteExecutor，InsertExecutor(通过SPI获取实现MySQL或Oracle)他们都继承自baseTransactionalExecutor。

rs = executor.execute(args); // 上面的execute执行baseTransactionalExecutor.execute方法。
public class baseTransactionalExecutor ... {     @Override
    public T execute(Object... args) throws Throwable {         if (RootContext.inGlobalTransaction()) {
            String xid = RootContext.getXID();             statementProxy.getConnectionProxy().bind(xid);
        }
        statementProxy.getConnectionProxy().setGlobalLockRequire(RootContext.requireGlobalLock());         return doExecute(args);
    } }

statementProxy.getConnectionProxy().bind(xid) ; // 该行代码将xid绑定到ConnectionProxy对象中的ConnectionContext上。
public void bind(String xid) {   context.bind(xid);
}

到这xid已经绑定到了ConnectionProxy中的ConnectionContext中。

Springboot之分布式事务框架Seata实现原理源码分析

Docker/k8s相关栏目本月热门文章