問題(轉https://segmentfault.com/a/1190000004235193)
業務場景
業務需求上常常會有一些邊緣操做，好比主流程操做A：用戶報名課程操做入庫，邊緣操做B：發送郵件或短信通知。html
業務要求
-
操做A操做數據庫失敗後，事務回滾，那麼操做B不能執行。java
-
操做A執行成功後，操做B也必須執行成功web
如何實現
-
普通的執行A，以後執行B，是能夠知足要求1，對於要求2一般須要設計補償的操做spring
-
通常邊緣的操做，一般會設置成爲異步的，以提高性能，好比發送MQ，業務系統負責事務成功後消息發送成功，而後接收系統負責保證通知成功完成數據庫
本文內容
如何在spring事務提交以後進行異步操做，這些異步操做必須得在該事務成功提交後才執行，回滾則不執行。segmentfault
要點
-
如何在spring事務提交以後操做mvc
-
如何把操做異步化app
實現方案
使用TransactionSynchronizationManager在事務提交以後操做
public void insert(TechBook techBook){ bookMapper.insert(techBook); // send after tx commit but is async TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() { @Override public void afterCommit() { System.out.println("send email after transaction commit..."); } } ); ThreadLocalRandom random = ThreadLocalRandom.current(); if(random.nextInt() % 2 ==0){ throw new RuntimeException("test email transaction"); } System.out.println("service end"); }
該方法就能夠實如今事務提交以後進行操做dom
操做異步化
使用mq或線程池來進行異步，好比使用線程池：異步
private final ExecutorService executorService = Executors.newFixedThreadPool(5); public void insert(TechBook techBook){ bookMapper.insert(techBook); // send after tx commit but is async TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() { @Override public void afterCommit() { executorService.submit(new Runnable() { @Override public void run() { System.out.println("send email after transaction commit..."); try { Thread.sleep(10*1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("complete send email after transaction commit..."); } }); } } ); // async work but tx not work, execute even when tx is rollback // asyncService.executeAfterTxComplete(); ThreadLocalRandom random = ThreadLocalRandom.current(); if(random.nextInt() % 2 ==0){ throw new RuntimeException("test email transaction"); } System.out.println("service end"); }
封裝以上兩步
對於第二步來講，若是這類方法比較多的話，則寫起來重複性太多，於是，抽象出來以下：
這裏改造了azagorneanu的代碼:
public interface AfterCommitExecutor extends Executor { } import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; import org.springframework.transaction.support.TransactionSynchronizationAdapter; import org.springframework.transaction.support.TransactionSynchronizationManager; import java.util.ArrayList; import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @Component public class AfterCommitExecutorImpl extends TransactionSynchronizationAdapter implements AfterCommitExecutor { private static final Logger LOGGER = LoggerFactory.getLogger(AfterCommitExecutorImpl.class); private static final ThreadLocal<List<Runnable>> RUNNABLES = new ThreadLocal<List<Runnable>>(); private ExecutorService threadPool = Executors.newFixedThreadPool(5); @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> threadRunnables = RUNNABLES.get(); if (threadRunnables == null) { threadRunnables = new ArrayList<Runnable>(); RUNNABLES.set(threadRunnables); TransactionSynchronizationManager.registerSynchronization(this); } threadRunnables.add(runnable); } @Override public void afterCommit() { List<Runnable> threadRunnables = RUNNABLES.get(); LOGGER.info("Transaction successfully committed, executing {} runnables", threadRunnables.size()); for (int i = 0; i < threadRunnables.size(); i++) { Runnable runnable = threadRunnables.get(i); LOGGER.info("Executing runnable {}", runnable); try { threadPool.execute(runnable); } catch (RuntimeException e) { LOGGER.error("Failed to execute runnable " + runnable, e); } } } @Override public void afterCompletion(int status) { LOGGER.info("Transaction completed with status {}", status == STATUS_COMMITTED ? "COMMITTED" : "ROLLED_BACK"); RUNNABLES.remove(); } } public void insert(TechBook techBook){ bookMapper.insert(techBook); // send after tx commit but is async // TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() { // @Override // public void afterCommit() { // executorService.submit(new Runnable() { // @Override // public void run() { // System.out.println("send email after transaction commit..."); // try { // Thread.sleep(10*1000); // } catch (InterruptedException e) { // e.printStackTrace(); // } // System.out.println("complete send email after transaction commit..."); // } // }); // } // } // ); //send after tx commit and is async afterCommitExecutor.execute(new Runnable() { @Override public void run() { try { Thread.sleep(5*1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("send email after transactioin commit"); } }); // async work but tx not work, execute even when tx is rollback // asyncService.executeAfterTxComplete(); ThreadLocalRandom random = ThreadLocalRandom.current(); if(random.nextInt() % 2 ==0){ throw new RuntimeException("test email transaction"); } System.out.println("service end"); }
關於Spring的Async
spring爲了方便應用使用線程池進行異步化，默認提供了@Async註解，能夠整個app使用該線程池，並且只要一個@Async註解在方法上面便可，省去重複的submit操做。關於async要注意的幾點：
一、async的配置
<context:component-scan base-package="com.yami" /> <!--配置@Async註解使用的線程池，這裏的id隨便命名，最後在task:annotation-driven executor= 指定上就能夠--> <task:executor id="myExecutor" pool-size="5"/> <task:annotation-driven executor="myExecutor" />
這個必須配置在root context裏頭，並且web context不能掃描controller層外的註解，不然會覆蓋掉。
<context:component-scan base-package="com.yami.web.controller"/> <mvc:annotation-driven/>
二、async的調用問題
async方法的調用，不能由同類方法內部調用，不然攔截不生效，這是spring默認的攔截問題，必須在其餘類裏頭調用另外一個類中帶有async的註解方法，才能起到異步效果。
三、事務問題
async方法若是也開始事務的話，要注意事務傳播以及事務開銷的問題。並且在async方法裏頭使用如上的TransactionSynchronizationManager.registerSynchronization不起做用，值得注意。