1. 程式人生 > >如何在資料庫事務提交成功後進行非同步操作

如何在資料庫事務提交成功後進行非同步操作

問題

業務場景

業務需求上經常會有一些邊緣操作,比如主流程操作A:使用者報名課程操作入庫,邊緣操作B:傳送郵件或簡訊通知。

業務要求

  • 操作A操作資料庫失敗後,事務回滾,那麼操作B不能執行。

  • 操作A執行成功後,操作B也必須執行成功

如何實現

  • 普通的執行A,之後執行B,是可以滿足要求1,對於要求2通常需要設計補償的操作

  • 一般邊緣的操作,通常會設定成為非同步的,以提升效能,比如傳送MQ,業務系統負責事務成功後訊息傳送成功,然後接收系統負責保證通知成功完成

本文內容

如何在spring事務提交之後進行非同步操作,這些非同步操作必須得在該事務成功提交後才執行,回滾則不執行。

要點

  • 如何在spring事務提交之後操作

  • 如何把操作非同步化

實現方案

使用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");
    }

該方法就可以實現在事務提交之後進行操作

操作非同步化

使用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要注意的幾點:

1、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/>

2、async的呼叫問題

async方法的呼叫,不能由同類方法內部呼叫,否則攔截不生效,這是spring預設的攔截問題,必須在其他類裡頭呼叫另一個類中帶有async的註解方法,才能起到非同步效果。

3、事務問題

async方法如果也開始事務的話,要注意事務傳播以及事務開銷的問題。而且在async方法裡頭使用如上的TransactionSynchronizationManager.registerSynchronization不起作用,值得注意。