在Spring事務(wù)提交后做點事兒

一顽聂、背景

最近團隊整理出的一份《Rabbit MQ消息定義規(guī)范》后，有同學(xué)提出了這樣的一個場景, 在事務(wù)還沒有執(zhí)行完消息就已經(jīng)發(fā)出去了, 導(dǎo)致后續(xù)的一些數(shù)據(jù)或邏輯上的問題產(chǎn)生, 那么既然出現(xiàn)了問題, 我們就需要解決這個問題, 正好這段時間在看Spring事務(wù)相關(guān)的知識, 所以本文就是帶著這樣的問題, 給出一些解決此問題的方案, 供大家參考.

二束凑、方案核心

本文整理了三種解決方案, 但是在給出解決方案之前, 我們需要了解一下這三種方案的技術(shù)核心點是什么, 因為這是重點中的重點, 因為三種方案背后的本質(zhì)邏輯都來源于此.
原理核心: Spring在事務(wù)正常提交完成后, 我們會看到這樣的一段代碼(更多源碼分析移步參考文章二):

// AbstractPlatformTransactionManager#commit
// 到此事務(wù)已經(jīng)正常提交結(jié)束了
try {
   // 這里就是三種方案的核心入口
   triggerAfterCommit(status);
} finally {
   // 事務(wù)回滾只會有這個
   triggerAfterCompletion(status, TransactionSynchronization.STATUS_COMMITTED);
}

這個代碼有什么用呢? 請看下面

// TransactionSynchronizationUtils#invokeAfterCommit
public static void invokeAfterCommit(List<TransactionSynchronization> synchronizations) {
   if (synchronizations != null) {
      for (TransactionSynchronization synchronization : synchronizations) {
         // 執(zhí)行注冊的TransactionSynchronization實現(xiàn)類的afterCommit方法
         synchronization.afterCommit();
      }
   }
}

到這里應(yīng)該就明白了, 我們要在事務(wù)提交結(jié)束后做點事情, 只需要想辦法搞個TransactionSynchronization的實現(xiàn)類, 并把這個實現(xiàn)類注冊到synchronizations中就可以實現(xiàn)我們想要的功能邏輯處理了, 那么接下來看看下面的解決方案是怎樣利用這一點, 解決問題的.
注意: 這里執(zhí)行afterCommit()即使拋異常了, 也不會導(dǎo)致事務(wù)回滾.

三与斤、解決方案

方案1. 利用TransactionSynchronizationManager的registerSynchronization()方法注冊TransactionSynchronization實現(xiàn)類

我們只需要在執(zhí)行的事務(wù)方法中, 添加如下代碼, 就可以完成在事務(wù)提交后的邏輯處理了

// TransactionSynchronizationAdapter是TransactionSynchronization的默認實現(xiàn)
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
    @Override
    public void afterCommit() {
        // 事務(wù)提交后需要執(zhí)行的業(yè)務(wù)邏輯: 發(fā)消息, 日志...
    }
});

當(dāng)然在每個事務(wù)方法里面寫這么一堆, 并不有優(yōu)雅, 所以可以寫了一個簡單的工具類供使用(可以根據(jù)實際使用優(yōu)化):

public class TransactionUtil {
    private static final ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(5,
            20, 30, TimeUnit.SECONDS, new ArrayBlockingQueue<>(20), new ThreadPoolExecutor.AbortPolicy());
 
    public static void afterCommit(Runnable runnable) {
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
            @Override
            public void afterCommit() {
                try {
                    THREAD_POOL_EXECUTOR.execute(runnable);
                } catch (Exception e) {
                    // 記錄日志
                }
            }
        });
    }
}
// 調(diào)用就變成了
TransactionUtil.afterCommit(() -> System.out.println("測試事務(wù)提交后的邏輯處理"));

本方案缺點:
- 事務(wù)后的邏輯需要自己完整的封裝
優(yōu)點:
- 簡單明了, 沒有額外的依賴, 上手容易
- 對Spring版本依賴小, 更通用, 3.x(或更低)以上版本均可使用.

方案2.利用Spring 4.2版本的新特性@TransactionalEventListener注解的事件機制來完成事務(wù)提交后的邏輯處理

@Autowired
private ApplicationEventPublisher publisher;
 
@Override
@Transactional(rollbackFor = Exception.class)
public void add(AdvanceChargeApplyAddInput input) {
    this.save(advanceChargeApply);
    // 發(fā)送事件
    publisher.publishEvent(advanceChargeApply);
}
// 響應(yīng)事件, 事務(wù)提交后執(zhí)行
@TransactionalEventListener
public void handle(PayloadApplicationEvent<AdvanceChargeApply> event) {
    System.out.println("TransactionalEventListener 事務(wù)提交后執(zhí)行");
}

本方案的原理是ApplicationListenerMethodTransactionalAdapter內(nèi)部封裝了@TransactionalEventListener注解, 添加了一個適配器ApplicationListenerMethodTransactionalAdapter(繼承了TransactionSynchronizationAdapter)內(nèi)部通過TransactionSynchronizationManager.registerSynchronization() 注冊一個TransactionSynchronization, 然后執(zhí)行afterCommit()時, 會調(diào)用ApplicationListenerMethodAdapter#processEvent(), 然后通過反射調(diào)用handle()方法.

本方案缺點:
- 需要Spring4.2 及以上版本才能支持
- 增加了事件的方法和接口的邏輯處理
優(yōu)點:
- 實現(xiàn)了邏輯處理的解耦

方案3. 方案的核心依然是繼承TransactionSynchronizationAdapter, 然后通過TransactionSynchronizationManager.registerSynchronization() 注冊一個TransactionSynchronization, 實現(xiàn)事務(wù)提交后的處理邏輯.

本方案缺點:
- 代碼邏輯顯得有點復(fù)雜
優(yōu)點:
- 可以實現(xiàn)更多邏輯的處理
- 整個邏輯實現(xiàn)更完善

四贺奠、總結(jié)

三種方案都利用了Spring在事務(wù)提交后的邏輯處理機制, 實現(xiàn)了不同的解決方案. 每種方案也各有優(yōu)缺點, 在此已簡單列舉, 私以為方案一最簡單, 對spring版本依賴也更小, 可以優(yōu)先考慮.
　本文主要是討論事務(wù)正常提交后的邏輯處理方式, 如果要在事務(wù)前或回滾后做邏輯處理, 同樣可以通過實現(xiàn)TransactionSynchronization不同的方法實現(xiàn), 原理如本文.
　最后, 歡迎大家對本文的方案進行討論補充和完善.

參考文章:
　http://ifeve.com/spring4-2/
　https://github.com/xmw9160/spring-framework-3.2.x
　http://azagorneanu.blogspot.com/2013/06/transaction-synchronization-callbacks.html