fnU28資訊網——每日最新資訊28at.com

一、背景

許多時候，我們期望在事務提交之后異步執行某些邏輯，調用外部系統，發送MQ，推送ES等等；當事務回滾時，異步操作也不執行，這些異步操作需要等待事務完成后才執行；比如出入庫的事務執行完畢后，異步發送MQ給報表系統、ES等等。fnU28資訊網——每日最新資訊28at.com

二、猜想

我們在項目中大多都是使用聲明式事務(@Transactional注解) ，spring會基于動態代理機制對我們的業務方法進行增強，控制connection，從而達到事務的目的。那么我們能否在此找尋一些蛛絲馬跡。我們來看下spring事務的相關核心類(裝配流程不詳細敘述)。fnU28資訊網——每日最新資訊28at.com

TransactionInterceptor：fnU28資訊網——每日最新資訊28at.com

public class TransactionInterceptor extends TransactionAspectSupport implements MethodInterceptor, Serializable {  @Override  @Nullable  public Object invoke(MethodInvocation invocation) throws Throwable {     // Work out the target class: may be {@code null}.     // The TransactionAttributeSource should be passed the target class     // as well as the method, which may be from an interface.     Class<?> targetClass = (invocation.getThis() != null ? AopUtils.getTargetClass(invocation.getThis()) : null);     // Adapt to TransactionAspectSupport's invokeWithinTransaction...     return invokeWithinTransaction(invocation.getMethod(), targetClass, invocation::proceed);  }}

TransactionAspectSupport(重點關注事務提交之后做了哪些事情，有哪些擴展點)。fnU28資訊網——每日最新資訊28at.com

public abstract class TransactionAspectSupport implements BeanFactoryAware, InitializingBean { protected Object invokeWithinTransaction(Method method, @Nullable Class<?> targetClass, final InvocationCallback invocation) throws Throwable {   // If the transaction attribute is null, the method is non-transactional.   TransactionAttributeSource tas = getTransactionAttributeSource();   final TransactionAttribute txAttr = (tas != null ? tas.getTransactionAttribute(method, targetClass) : null);   final TransactionManager tm = determineTransactionManager(txAttr);   if (this.reactiveAdapterRegistry != null && tm instanceof ReactiveTransactionManager) {      ReactiveTransactionSupport txSupport = this.transactionSupportCache.computeIfAbsent(method, key -> {         if (KotlinDetector.isKotlinType(method.getDeclaringClass()) && KotlinDelegate.isSuspend(method)) {            throw new TransactionUsageException(                  "Unsupported annotated transaction on suspending function detected: " + method +                  ". Use TransactionalOperator.transactional extensions instead.");         }         ReactiveAdapter adapter = this.reactiveAdapterRegistry.getAdapter(method.getReturnType());         if (adapter == null) {            throw new IllegalStateException("Cannot apply reactive transaction to non-reactive return type: " +                  method.getReturnType());         }         return new ReactiveTransactionSupport(adapter);      });      return txSupport.invokeWithinTransaction(            method, targetClass, invocation, txAttr, (ReactiveTransactionManager) tm);   }   PlatformTransactionManager ptm = asPlatformTransactionManager(tm);   final String joinpointIdentification = methodIdentification(method, targetClass, txAttr);   if (txAttr == null || !(ptm instanceof CallbackPreferringPlatformTransactionManager)) {      // 創建事務，此處也會創建connection      TransactionInfo txInfo = createTransactionIfNecessary(ptm, txAttr, joinpointIdentification);      Object retVal;      try {         // 執行目標方法         retVal = invocation.proceedWithInvocation();      }      catch (Throwable ex) {         // 目標方法異常時處理         completeTransactionAfterThrowing(txInfo, ex);         throw ex;      }      finally {		 // 重置TransactionInfo ThreadLocal         cleanupTransactionInfo(txInfo);      }      if (vavrPresent && VavrDelegate.isVavrTry(retVal)) {         // Set rollback-only in case of Vavr failure matching our rollback rules...         TransactionStatus status = txInfo.getTransactionStatus();         if (status != null && txAttr != null) {            retVal = VavrDelegate.evaluateTryFailure(retVal, txAttr, status);         }      }	  // 業務方法成功執行，提交事務(重點關注此處)，最終會調用AbstractPlatformTransactionManager#commit方法      commitTransactionAfterReturning(txInfo);      return retVal;   }   else {      final ThrowableHolder throwableHolder = new ThrowableHolder();      // It's a CallbackPreferringPlatformTransactionManager: pass a TransactionCallback in.      try {         Object result = ((CallbackPreferringPlatformTransactionManager) ptm).execute(txAttr, status -> {            TransactionInfo txInfo = prepareTransactionInfo(ptm, txAttr, joinpointIdentification, status);            try {               Object retVal = invocation.proceedWithInvocation();               if (vavrPresent && VavrDelegate.isVavrTry(retVal)) {                  // Set rollback-only in case of Vavr failure matching our rollback rules...                  retVal = VavrDelegate.evaluateTryFailure(retVal, txAttr, status);               }               return retVal;            }            catch (Throwable ex) {               if (txAttr.rollbackOn(ex)) {                  // A RuntimeException: will lead to a rollback.                  if (ex instanceof RuntimeException) {                     throw (RuntimeException) ex;                  }                  else {                     throw new ThrowableHolderException(ex);                  }               }               else {                  // A normal return value: will lead to a commit.                  throwableHolder.throwable = ex;                  return null;               }            }            finally {               cleanupTransactionInfo(txInfo);            }         });         // Check result state: It might indicate a Throwable to rethrow.         if (throwableHolder.throwable != null) {            throw throwableHolder.throwable;         }         return result;      }      catch (ThrowableHolderException ex) {         throw ex.getCause();      }      catch (TransactionSystemException ex2) {         if (throwableHolder.throwable != null) {            logger.error("Application exception overridden by commit exception", throwableHolder.throwable);            ex2.initApplicationException(throwableHolder.throwable);         }         throw ex2;      }      catch (Throwable ex2) {         if (throwableHolder.throwable != null) {            logger.error("Application exception overridden by commit exception", throwableHolder.throwable);         }         throw ex2;      }   }}}

AbstractPlatformTransactionManager：fnU28資訊網——每日最新資訊28at.com

public final void commit(TransactionStatus status) throws TransactionException {   if (status.isCompleted()) {      throw new IllegalTransactionStateException(            "Transaction is already completed - do not call commit or rollback more than once per transaction");   }   DefaultTransactionStatus defStatus = (DefaultTransactionStatus) status;   if (defStatus.isLocalRollbackOnly()) {      if (defStatus.isDebug()) {         logger.debug("Transactional code has requested rollback");      }      processRollback(defStatus, false);      return;   }   if (!shouldCommitOnGlobalRollbackOnly() && defStatus.isGlobalRollbackOnly()) {      if (defStatus.isDebug()) {         logger.debug("Global transaction is marked as rollback-only but transactional code requested commit");      }      processRollback(defStatus, true);      return;   }   // 事務提交處理   processCommit(defStatus);}private void processCommit(DefaultTransactionStatus status) throws TransactionException {   try {      boolean beforeCompletionInvoked = false;      try {         boolean unexpectedRollback = false;         prepareForCommit(status);         triggerBeforeCommit(status);         triggerBeforeCompletion(status);         beforeCompletionInvoked = true;         if (status.hasSavepoint()) {            if (status.isDebug()) {               logger.debug("Releasing transaction savepoint");            }            unexpectedRollback = status.isGlobalRollbackOnly();            status.releaseHeldSavepoint();         }         else if (status.isNewTransaction()) {            if (status.isDebug()) {               logger.debug("Initiating transaction commit");            }            unexpectedRollback = status.isGlobalRollbackOnly();            doCommit(status);         }         else if (isFailEarlyOnGlobalRollbackOnly()) {            unexpectedRollback = status.isGlobalRollbackOnly();         }         // Throw UnexpectedRollbackException if we have a global rollback-only         // marker but still didn't get a corresponding exception from commit.         if (unexpectedRollback) {            throw new UnexpectedRollbackException(                  "Transaction silently rolled back because it has been marked as rollback-only");         }      }      catch (UnexpectedRollbackException ex) {         // can only be caused by doCommit         triggerAfterCompletion(status, TransactionSynchronization.STATUS_ROLLED_BACK);         throw ex;      }      catch (TransactionException ex) {         // can only be caused by doCommit         if (isRollbackOnCommitFailure()) {            doRollbackOnCommitException(status, ex);         }         else {            triggerAfterCompletion(status, TransactionSynchronization.STATUS_UNKNOWN);         }         throw ex;      }      catch (RuntimeException | Error ex) {         if (!beforeCompletionInvoked) {            triggerBeforeCompletion(status);         }         doRollbackOnCommitException(status, ex);         throw ex;      }      // Trigger afterCommit callbacks, with an exception thrown there      // propagated to callers but the transaction still considered as committed.      try {		 // 在事務提交后觸發(追蹤到這里就離真相不遠了)         triggerAfterCommit(status);      }      finally {         triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);      }   }   finally {      cleanupAfterCompletion(status);   }}

TransactionSynchronizationUtils：fnU28資訊網——每日最新資訊28at.com

public abstract class TransactionSynchronizationUtils {  public static void triggerAfterCommit() {     // TransactionSynchronizationManager: 事務同步器管理     invokeAfterCommit(TransactionSynchronizationManager.getSynchronizations());  }  public static void invokeAfterCommit(@Nullable List<TransactionSynchronization> synchronizations) {     if (synchronizations != null) {        for (TransactionSynchronization synchronization : synchronizations) {		   // 調用TransactionSynchronization#afterCommit方法，默認實現為空，留給子類擴展		   // 那么我們想在事務提交之后做一些異步操作，實現此方法即可           synchronization.afterCommit();        }     }  }}

TransactionSynchronization：fnU28資訊網——每日最新資訊28at.com

public interface TransactionSynchronization extends Flushable {   default void afterCommit() {}}

過程中我們發現TransactionSynchronizationManager、TransactionSynchronization、TransactionSynchronizationAdapter 等相關類涉及aop的整個流程，篇幅有限，在此不詳細展開，當然我們的一些擴展也是離不開這些基礎類的。fnU28資訊網——每日最新資訊28at.com

三、實現

事務提交之后異步執行，我們需自定義synchronization.afterCommit，結合線程池一起使用，定義線程池TaskExecutor。fnU28資訊網——每日最新資訊28at.com

@Beanpublic TaskExecutor taskExecutor() {    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();    taskExecutor.setCorePoolSize(******);    taskExecutor.setMaxPoolSize(******);    taskExecutor.setKeepAliveSeconds(******);    taskExecutor.setQueueCapacity(******);    taskExecutor.setThreadNamePrefix(******);    taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());    taskExecutor.initialize();    return taskExecutor;}

定義AfterCommitExecutor接口。fnU28資訊網——每日最新資訊28at.com

public interface AfterCommitExecutor extends Executor { }

定義AfterCommitExecutorImpl實現類，注意需繼承TransactionSynchronizationAdapter類。fnU28資訊網——每日最新資訊28at.com

import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.stereotype.Component;import org.springframework.core.NamedThreadLocal;import org.springframework.core.task.TaskExecutor;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.transaction.support.TransactionSynchronizationAdapter;import org.springframework.transaction.support.TransactionSynchronizationManager;import java.util.List;import java.util.ArrayList;@Componentpublic class AfterCommitExecutorImpl extends TransactionSynchronizationAdapter implements AfterCommitExecutor {    private static final Logger LOGGER = LoggerFactory.getLogger(AfterCommitExecutorImpl.class);    // 保存要運行的任務線程    private static final ThreadLocal<List<Runnable>> RUNNABLE_THREAD_LOCAL = new NamedThreadLocal<>("AfterCommitRunnable");    // 設置線程池    @Autowired    private TaskExecutor taskExecutor;    /**     * 異步執行     *     * @param runnable 異步線程     */    @Override    public void execute(Runnable runnable) {        LOGGER.info("Submitting new runnable {} to run after commit", runnable);        // 如果事務已經提交，馬上進行異步處理        if (!TransactionSynchronizationManager.isSynchronizationActive()) {            LOGGER.info("Transaction synchronization is NOT ACTIVE. Executing right now runnable {}", runnable);            runnable.run();            return;        }        // 同一個事務的合并到一起處理(注意：沒有初始化則初始化，并注冊)        List<Runnable> threadRunnableList = RUNNABLE_THREAD_LOCAL.get();        if (null == threadRunnableList) {            threadRunnableList = new ArrayList<>();            RUNNABLE_THREAD_LOCAL.set(threadRunnableList);            TransactionSynchronizationManager.registerSynchronization(this);        }        threadRunnableList.add(runnable);    }    /**     * 監聽到事務提交之后執行方法     */    @Override    public void afterCommit() {        List<Runnable> threadRunnableList = RUNNABLE_THREAD_LOCAL.get();        LOGGER.info("Transaction successfully committed, executing {} threadRunnable", threadRunnableList.size());        for (Runnable runnable : threadRunnableList) {            try {                taskExecutor.execute(runnable);            } catch (RuntimeException e) {                LOGGER.error("Failed to execute runnable " + runnable, e);            }        }    }    /**     * 事務提交/回滾執行     *     * @param status (STATUS_COMMITTED-0、STATUS_ROLLED_BACK-1、STATUS_UNKNOWN-2)     */    @Override    public void afterCompletion(int status) {        LOGGER.info("Transaction completed with status {}", status == STATUS_COMMITTED ? "COMMITTED" : "ROLLED_BACK");        RUNNABLE_THREAD_LOCAL.remove();    }}

使用。fnU28資訊網——每日最新資訊28at.com

工具類封裝好了，使用上那就很簡便了：注入AfterCommitExecutor，調用AfterCommitExecutor.execute(runnable)方法即可

四、總結

spring如此龐大，找準切入點，許多問題都是可以找到解決思路、或者方案；fnU28資訊網——每日最新資訊28at.com

你對spring了解多少......fnU28資訊網——每日最新資訊28at.com

本文鏈接：http://www.www897cc.com/showinfo-26-10420-0.html事務提交之后異步執行工具類封裝

聲明：本網頁內容旨在傳播知識，若有侵權等問題請及時與本網聯系，我們將在第一時間刪除處理。郵件：2376512515@qq.com

上一篇： 為什么說MyBatis默認的DefaultSqlSession是線程不安全？

下一篇： AIoTel下視頻編碼（一）--移動看家視頻水印溯源技術