Spring編程式事務 + Java 多線程實現(xiàn)批量操作的回滾！

前兩天，我發(fā)了一篇推文《Java多線程+List分段完美解決導入等批量更新場景問題！》。

有熱心好學的朋友留言到：

看到他的留言，我才知道，原來我們早已成了微信好友??。

關于他提到的問題，今天我再整個簡單的 demo，實現(xiàn)單事務多線程操作數(shù)據(jù)的案例。

核心問題

多線程更新或插入操作是很快，但是運行之后有網(wǎng)友發(fā)現(xiàn)，這些線程并不存在于一個事務中。他們想要一個報錯，全體回滾。但是暫時又不明白該怎么實現(xiàn)，因此我寫了本文（水文??）。

偽代碼

多線程插入偽代碼：

@Transactional
public void test() throws InterruptedException {
    CountDownLatch countDownLatch = new CountDownLatch(5);
    for (int i = 0; i <5 ; i++) {
        new Thread(new Runnable() {
            @Override
            public void run() {
                AreaController.this.insert();//插入一條數(shù)據(jù)
                countDownLatch.countDown();
            }
        }).start();
    }
    countDownLatch.await();
    int i =1/0; //主線程報錯
}

無法回滾的原因

了解 Spring 事務的網(wǎng)友，應該知道。Spring 相關數(shù)據(jù)庫連接信息都放在了 threadLocal 中。所以不同的線程享用不同的連接信息。所以前面的多線程操作不存在于一個事務中。

那么我們則么做到同步呢？

多線程單事務

要解決這個問題，就要讓多個線程共用同一個事務。

這樣以來，我們就不能使用 Spring 提供的注解了。我們需要自己操控事務管理。

簡單的案例代碼如下所示：

@Autowired
private PlatformTransactionManager transactionManager;

public void test() throws InterruptedException {
    CountDownLatch rollBackLatch = new CountDownLatch(1);
    CountDownLatch mainThreadLatch = new CountDownLatch(5);
    AtomicBoolean rollbackFlag = new AtomicBoolean(false);
    for (int i = 0; i < 5; i++) {
        int t = i;
        new Thread(() -> {
            if (rollbackFlag.get()) return; //如果其他線程已經(jīng)報錯 就停止線程
            //設置一個事務
            DefaultTransactionDefinition def = new DefaultTransactionDefinition();
            def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); // 事物隔離級別，開啟新事務，這樣會比較安全些。
            TransactionStatus status = transactionManager.getTransaction(def); // 獲得事務狀態(tài)
            try {
                AreaController.this.insert();//插入一條數(shù)據(jù)
                // if (t==4)throw new RuntimeException("報錯");
                mainThreadLatch.countDown();
                rollBackLatch.await();//線程等待
                if (rollbackFlag.get()) {
                    transactionManager.rollback(status);
                } else {
                    transactionManager.commit(status);
                }
            } catch (Exception e) {
                e.printStackTrace();
                //如果出錯了 就放開鎖 讓別的線程進入提交/回滾  本線程進行回滾
                rollbackFlag.set(true);
                rollBackLatch.countDown();
                mainThreadLatch.countDown();
                transactionManager.rollback(status);
            }
        }).start();
    }
    try {
//            int i = 1 / 0; //主線程執(zhí)行相關業(yè)務報錯
    } catch (Exception e) {
        rollbackFlag.set(true);
        rollBackLatch.countDown();
    }
    //主線程業(yè)務執(zhí)行完畢 如果其他線程也執(zhí)行完畢 且沒有報異常 正在阻塞狀態(tài)中 喚醒其他線程 提交所有的事務
    //如果其他線程或者主線程報錯 則不會進入if 會觸發(fā)回滾
    if (!rollbackFlag.get()){
        mainThreadLatch.await();
        rollBackLatch.countDown();
    }
}

這里面，最主要的是你要了解 Spring 聲明式事務 @Transactional 的實現(xiàn)原理《Spring源碼閱讀，@Transactional實現(xiàn)原理》。

首先對于 Spring 的聲明式事務 @Transactional 在多線程環(huán)境下明顯是失效了的，原因是這些方法無法被加入到同一個事務中。

但是 Spring 作為業(yè)界的標桿，它提供了另一種控制粒度更加細的編程式事務，可以通過事務管理對象 PlatformTransactionManager 來操作事務的提交與回滾，我們可以把事務相關提交回滾的代碼寫在每個線程之中。

上面我們使用了一個回滾標志 rollbackFlag 來控制事務是提交還是回滾，并且使用了 countDownLatch 來進行線程阻塞的控制。只有所有的任務都正常的執(zhí)行完畢了，才執(zhí)行事務提交，如果其中一個線程報錯了，則進行全體執(zhí)行回滾。

水文，不喜輕噴！