一加Ace2 Pro官宣：普及16G內存 引領24G
一加官方今天繼續為本月發布的新機一加Ace2 Pro帶來預熱，公布了內存方面的信息。“淘汰 8GB ，12GB 起步，16GB 普及，24GB 引領，還有呢？#一加Ace2Pro#，2023 年 8 月，敬請期待。”同時
今天，我繼續安利一個獨門絕技：Spring 事務的鉤子函數。單純的講技術可能比較枯燥乏味。接下來，我將以一個實際的案例來描述Spring事務鉤子函數的正確使用姿勢。
拿支付系統相關的業務來舉例。在支付系統中，我們需要記錄每個賬戶的資金流水（記錄用戶A因為哪個操作扣了錢，因為哪個操作加了錢），這樣我們才能對每個賬戶的賬做到心中有數，對于支付系統而言，資金流水的數據可謂是最重要的。因此，為了防止支付系統的老大徇私舞弊，CTO提了一個流水存檔的需求：要求支付系統對每個賬戶的資金流水做一份存檔，要求支付系統在寫流水的時候，把流水相關的信息以消息的形式推送到kafka，由存檔系統消費這個消息并落地到庫里（這個庫只有存檔系統擁有寫權限）。整個需求的流程如下所示：
圖片
整個需求的流程還是比較簡單的，考慮到后續會有其他事業部也要進行數據存檔操作，CTO建議支付系統團隊內部開發一個二方庫，這個二方庫的主要功能就是發送消息到kafka中去。
既然要求開發一個二方庫，因此，我們需要考慮如下幾件事情：
1、技術棧使用的springboot，因此，這里最好以starter的方式提供
2、二方庫需要發送消息給kafka，最好是二方庫內部基于kafka生產者的api創建生產者，不要使用Spring自帶的kafkaTemplate，因為集成方有可能已經使用了kafkaTemplate。不能與集成方造成沖突。
3、減少對接方的集成難度、學習成本，最好是提供一個簡單實用的api，業務側能簡單上手。
4、發送消息這個操作需要支持事務，盡量不影響主業務
在上述的幾件事情中，最需要注意的應該就是第4點：發送消息這個操作需要支持事務，盡量不影響主業務。這是什么意思呢？首先，盡量不影響主業務，這個最簡單的方式就是使用異步機制。其次，需要支持事務是指：假設我們的api是在事務方法內部調用的，那么我們需要保證事務提交后再執行這個api。那么，我們的流水落地api應該要有這樣的功能：
圖片
內部可以判斷當前是否存在事務，如果存在事務，則需要等事務提交后再異步發送消息給kafka。如果不存在事務則直接異步發送消息給kafka。而且這樣的判斷邏輯得放在二方庫內部才行。那現在擺在我們面前的問題就是：我要如何判斷當前是否存在事務，以及如何在事務提交后再觸發我們自定義的邏輯呢？
這個類內部所有的變量、方法都是static修飾的，也就是說它其實是一個工具類。是一個事務同步器。下述是流水落地API的偽代碼，這段代碼就解決了我們上述提到的疑問：
private final ExecutorService executor = Executors.newSingleThreadExecutor();public void sendLog() { // 判斷當前是否存在事務 if (!TransactionSynchronizationManager.isSynchronizationActive()) { // 無事務，異步發送消息給kafka executor.submit(() -> { // 發送消息給kafka try { // 發送消息給kafka } catch (Exception e) { // 記錄異常信息，發郵件或者進入待處理列表，讓開發人員感知異常 } }); return; } // 有事務，則添加一個事務同步器，并重寫afterCompletion方法（此方法在事務提交后會做回調） TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() { @Override public void afterCompletion(int status) { if (status == TransactionSynchronization.STATUS_COMMITTED) { // 事務提交后，再異步發送消息給kafka executor.submit(() -> { try { // 發送消息給kafka } catch (Exception e) { // 記錄異常信息，發郵件或者進入待處理列表，讓開發人員感知異常 } }); } } });}代碼比較簡單，其主要是TransactionSynchronizationManager的使用。
我們先看下這個方法的源碼：
// TransactionSynchronizationManager.java類內部的部分代碼private static final ThreadLocal<Set<TransactionSynchronization>> synchronizations = new NamedThreadLocal<>("Transaction synchronizations");public static boolean isSynchronizationActive() { return (synchronizations.get() != null);}很明顯，synchronizations是一個線程變量（ThreadLocal）。那它是在什么時候set進去的呢？這里的話，可以參考下這個方法：org.springframework.transaction.support.TransactionSynchronizationManager#initSynchronization，其源碼如下所示：
/** * Activate transaction synchronization for the current thread. * Called by a transaction manager on transaction begin. * @throws IllegalStateException if synchronization is already active */public static void initSynchronization() throws IllegalStateException { if (isSynchronizationActive()) { throw new IllegalStateException("Cannot activate transaction synchronization - already active"); } logger.trace("Initializing transaction synchronization"); synchronizations.set(new LinkedHashSet<>());}由源碼中的注釋也可以知道，它是在事務管理器開啟事務時調用的。換句話說，只要我們的程序執行到帶有事務特性的方法時，就會在線程變量中放入一個LinkedHashSet，用來標識當前存在事務。只要isSynchronizationActive返回true，則代表當前有事務。因此，結合這兩個方法我們是指能解決我們最開始提出的疑問：**要如何判斷當前是否存在事務**
我們來看下這個方法的源代碼：
/** * Register a new transaction synchronization for the current thread. * Typically called by resource management code. * <p>Note that synchronizations can implement the * {@link org.springframework.core.Ordered} interface. * They will be executed in an order according to their order value (if any). * @param synchronization the synchronization object to register * @throws IllegalStateException if transaction synchronization is not active * @see org.springframework.core.Ordered */public static void registerSynchronization(TransactionSynchronization synchronization) throws IllegalStateException { Assert.notNull(synchronization, "TransactionSynchronization must not be null"); if (!isSynchronizationActive()) { throw new IllegalStateException("Transaction synchronization is not active"); } synchronizations.get().add(synchronization);}這里又使用到了synchronizations線程變量，我們在判斷是否存在事務時，就是判斷這個線程變量內部是否有值。那我們現在想在事務提交后觸發自定義邏輯和這個有什么關系呢？我們在上面構建流水落地api的偽代碼中有向synchronizations內部添加了一個TransactionSynchronizationAdapter，內部并重寫了afterCompletion方法，其代碼如下所示：
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() { @Override public void afterCompletion(int status) { if (status == TransactionSynchronization.STATUS_COMMITTED) { // 事務提交后，再異步發送消息給kafka executor.submit(() -> { try { // 發送消息給kafka } catch (Exception e) { // 記錄異常信息，發郵件或者進入待處理列表，讓開發人員感知異常 } }); } }});我們結合registerSynchronization的源碼來看，其實這段代碼主要就是向線程變量內部的LinkedHashSet添加了一個對象而已，但就是這么一個操作，讓Spring在事務執行的過程中變得“有事情可做”。這是什么意思呢？是因為Spring在執行事務方法時，對于操作事務的每一個階段都有一個回調操作，比如：trigger系列的回調
圖片
invoke系列的回調
圖片
而我們現在的需求就是在事務提交后觸發自定義的函數，那就是在invokeAfterCommit和invokeAfterCompletion這兩個方法來選了。首先，這兩個方法都會拿到所有TransactionSynchronization的集合（其中會包括我們上述添加的TransactionSynchronizationAdapter）。但是要注意一點：invokeAfterCommit只能拿到集合，invokeAfterCompletion除了集合還有一個int類型的參數，而這個int類型的參數其實是當前事務的一種狀態。也就是說，如果我們重寫了invokeAfterCompletion方法，我們除了能拿到集合外，還能拿到當前事務的狀態。因此，此時我們可以根據這個狀態來做不同的事情，比如：可以在事務提交時做自定義處理，也可以在事務回滾時做自定義處理等等。
上面有說到，我們判斷當前是否存在事務、添加鉤子函數都是依賴線程變量的。因此，我們在使用過程中，一定要避免切換線程。否則會出現不生效的情況。
本文鏈接：http://m.www897cc.com/showinfo-26-87482-0.html事務鉤子函數，打造高效支付系統
聲明：本網頁內容旨在傳播知識，若有侵權等問題請及時與本網聯系，我們將在第一時間刪除處理。郵件：2376512515@qq.com
上一篇： WPF UI更新技巧：掌握EventHandler的基礎與Dispatcher的高級應用
下一篇： 哈希表哪家強？幾大編程語言吵起來了！
Copyright ? 2016-2023 天津谷騏科技有限公司 版權所有 sitemap.xml
違法及侵權請聯系：2376512515@qq.com 津ICP備18001702號
津公網安備 12010102000574號