面試官：Java 多執行緒怎么做事務控制？一半人答不上來。。-有解無憂

專案代碼基于：MySql 資料，開發框架為：SpringBoot、Mybatis

開發語言為：Java8

前言

公司業務中遇到一個需求，需要同時修改最多約5萬條資料，而且還不支持批量或異步修改操作，于是只能寫個for回圈操作，但操作耗時太長，只能一步一步尋找其他解決方案，

具體操作如下：

一、回圈操作的代碼

先寫一個最簡單的for回圈代碼，看看耗時情況怎么樣，

/***
 * 一條一條依次對50000條資料進行更新操作
 * 耗時：2m27s,1m54s
 */
@Test
void updateStudent() {
    List<Student> allStudents = studentMapper.getAll();
    allStudents.forEach(s -> {
        //更新教師資訊
        String teacher = s.getTeacher();
        String newTeacher = "TNO_" + new Random().nextInt(100);
        s.setTeacher(newTeacher);
        studentMapper.update(s);
    });
}

回圈修改整體耗時約 1分54秒，且代碼中沒有手動事務控制應該是自動事務提交，所以每次操作事務都會提交所以操作比較慢，我們先對代碼中添加手動事務控制，看查詢效率怎樣，

最新面試題整理：https://www.javastack.cn/mst/

二、使用手動事務的操作代碼

修改后的代碼如下：

@Autowired
private DataSourceTransactionManager dataSourceTransactionManager;

@Autowired
private TransactionDefinition transactionDefinition;

/**
 * 由于希望更新操作 一次性完成，需要手動控制添加事務
 * 耗時：24s
 * 從測驗結果可以看出，添加事務后插入資料的效率有明顯的提升
 */
@Test
void updateStudentWithTrans() {
    List<Student> allStudents = studentMapper.getAll();
    TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
    try {
        allStudents.forEach(s -> {
            //更新教師資訊
            String teacher = s.getTeacher();
            String newTeacher = "TNO_" + new Random().nextInt(100);
            s.setTeacher(newTeacher);
            studentMapper.update(s);
        });
        dataSourceTransactionManager.commit(transactionStatus);
    } catch (Throwable e) {
        dataSourceTransactionManager.rollback(transactionStatus);
        throw e;
    }
}

添加手動事務操控制后，整體耗時約 24秒，這相對于自動事務提交的代碼，快了約5倍，對于大量回圈資料庫提交操作，添加手動事務可以有效提高操作效率，

三、嘗試多執行緒進行資料修改

添加資料庫手動事務后操作效率有明細提高，但還是比較長，接下來嘗試多執行緒提交看是不是能夠再快一些，

先添加一個Service將批量修改操作整合一下，具體代碼如下：

StudentServiceImpl.java

@Service
public class StudentServiceImpl implements StudentService {
    @Autowired
    private StudentMapper studentMapper;

    @Autowired
    private DataSourceTransactionManager dataSourceTransactionManager;

    @Autowired
    private TransactionDefinition transactionDefinition;

    @Override
    public void updateStudents(List<Student> students, CountDownLatch threadLatch) {
        TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
        System.out.println("子執行緒：" + Thread.currentThread().getName());
        try {
            students.forEach(s -> {
                // 更新教師資訊
                // String teacher = s.getTeacher();
                String newTeacher = "TNO_" + new Random().nextInt(100);
                s.setTeacher(newTeacher);
                studentMapper.update(s);
            });
            dataSourceTransactionManager.commit(transactionStatus);
            threadLatch.countDown();
        } catch (Throwable e) {
            e.printStackTrace();
            dataSourceTransactionManager.rollback(transactionStatus);
        }
    }
}

批量測驗代碼，我們采用了多執行緒進行提交，修改后測驗代碼如下：

@Autowired
private DataSourceTransactionManager dataSourceTransactionManager;

@Autowired
private TransactionDefinition transactionDefinition;

@Autowired
private StudentService studentService;

/**
 * 對用戶而言，27s 任是一個較長的時間，我們嘗試用多執行緒的方式來經行修改操作看能否加快處理速度
 * 預計創建10個執行緒，每個執行緒進行5000條資料修改操作
 * 耗時統計
 * 1 執行緒數：1      耗時：25s
 * 2 執行緒數：2      耗時：14s
 * 3 執行緒數：5      耗時：15s
 * 4 執行緒數：10     耗時：15s
 * 5 執行緒數：100    耗時：15s
 * 6 執行緒數：200    耗時：15s
 * 7 執行緒數：500    耗時：17s
 * 8 執行緒數：1000    耗時：19s
 * 8 執行緒數：2000    耗時：23s
 * 8 執行緒數：5000    耗時：29s
 */
@Test
void updateStudentWithThreads() {
    //查詢總資料
    List<Student> allStudents = studentMapper.getAll();
    // 執行緒數量
    final Integer threadCount = 100;

    //每個執行緒處理的資料量
    final Integer dataPartionLength = (allStudents.size() + threadCount - 1) / threadCount;

    // 創建多執行緒處理任務
    ExecutorService studentThreadPool = Executors.newFixedThreadPool(threadCount);
    CountDownLatch threadLatchs = new CountDownLatch(threadCount);

    for (int i = 0; i < threadCount; i++) {
        // 每個執行緒處理的資料
        List<Student> threadDatas = allStudents.stream()
                .skip(i * dataPartionLength).limit(dataPartionLength).collect(Collectors.toList());
        studentThreadPool.execute(() -> {
            studentService.updateStudents(threadDatas, threadLatchs);
        });
    }
    try {
        // 倒計時鎖設定超時時間 30s
        threadLatchs.await(30, TimeUnit.SECONDS);
    } catch (Throwable e) {
        e.printStackTrace();
    }

    System.out.println("主執行緒完成");
}

多執行緒提交修改時，我們嘗試了不同執行緒數對提交速度的影響，具體可以看下面表格，

多執行緒修改50000條資料時不同執行緒數耗時對比（秒）

根據表格，我們執行緒數增大提交速度并非一直增大，在當前情況下約在2-5個執行緒數時，提交速度最快（實際執行緒數還是需要根據服務器配置實際測驗），

另外，MySQL 系列面試題和答案全部整理好了，微信搜索Java技術堆疊，在后臺發送：面試，可以在線閱讀，

四、基于兩個CountDownLatch控制多執行緒事務提交

由于多執行緒提交時，每個執行緒事務時單獨的，無法保證一致性，我們嘗試給多執行緒添加事務控制，來保證每個執行緒都是在插入資料完成后在提交事務，

這里我們使用兩個 CountDownLatch 來控制主執行緒與子執行緒事務提交，并設定了超時時間為 30 秒，我們對代碼進行了一點修改：

@Override
public void updateStudentsThread(List<Student> students, CountDownLatch threadLatch, CountDownLatch mainLatch, StudentTaskError taskStatus) {
    TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
    System.out.println("子執行緒：" + Thread.currentThread().getName());
    try {
        students.forEach(s -> {
            // 更新教師資訊
            // String teacher = s.getTeacher();
            String newTeacher = "TNO_" + new Random().nextInt(100);
            s.setTeacher(newTeacher);
            studentMapper.update(s);
        });
    } catch (Throwable e) {
        taskStatus.setIsError();
    } finally {
        threadLatch.countDown(); // 切換到主執行緒執行
    }
    try {
        mainLatch.await();  //等待主執行緒執行
    } catch (Throwable e) {
        taskStatus.setIsError();
    }
    // 判斷是否有錯誤，如有錯誤 就回滾事務
    if (taskStatus.getIsError()) {
        dataSourceTransactionManager.rollback(transactionStatus);
    } else {
        dataSourceTransactionManager.commit(transactionStatus);
    }
}
/**
 * 由于每個執行緒都是單獨的事務，需要添加對執行緒事務的統一控制
 * 我們這邊使用兩個 CountDownLatch 對子執行緒的事務進行控制
 */
@Test
void updateStudentWithThreadsAndTrans() {
    //查詢總資料
    List<Student> allStudents = studentMapper.getAll();
    // 執行緒數量
    final Integer threadCount = 4;

    //每個執行緒處理的資料量
    final Integer dataPartionLength = (allStudents.size() + threadCount - 1) / threadCount;

    // 創建多執行緒處理任務
    ExecutorService studentThreadPool = Executors.newFixedThreadPool(threadCount);
    CountDownLatch threadLatchs = new CountDownLatch(threadCount); // 用于計算子執行緒提交數量
    CountDownLatch mainLatch = new CountDownLatch(1); // 用于判斷主執行緒是否提交
    StudentTaskError taskStatus = new StudentTaskError(); // 用于判斷子執行緒任務是否有錯誤

    for (int i = 0; i < threadCount; i++) {
        // 每個執行緒處理的資料
        List<Student> threadDatas = allStudents.stream()
                .skip(i * dataPartionLength).limit(dataPartionLength)
                .collect(Collectors.toList());
        studentThreadPool.execute(() -> {
            studentService.updateStudentsThread(threadDatas, threadLatchs, mainLatch, taskStatus);
        });
    }
    try {
        // 倒計時鎖設定超時時間 30s
        boolean await = threadLatchs.await(30, TimeUnit.SECONDS);
        if (!await) { // 等待超時，事務回滾
            taskStatus.setIsError();
        }
    } catch (Throwable e) {
        e.printStackTrace();
        taskStatus.setIsError();
    }
    mainLatch.countDown(); // 切換到子執行緒執行
    studentThreadPool.shutdown(); //關閉執行緒池

    System.out.println("主執行緒完成");
}

本想再次測驗一下不同執行緒數對執行效率的影響時，發現當執行緒數超過10個時，執行時就報錯，具體錯誤內容如下：

Exception in thread "pool-1-thread-2" org.springframework.transaction.CannotCreateTransactionException: Could not open JDBC Connection for transaction; nested exception is java.sql.SQLTransientConnectionException: HikariPool-1 - Connection is not available, request timed out after 30055ms.
 at org.springframework.jdbc.datasource.DataSourceTransactionManager.doBegin(DataSourceTransactionManager.java:309)
 at org.springframework.transaction.support.AbstractPlatformTransactionManager.startTransaction(AbstractPlatformTransactionManager.java:400)
 at org.springframework.transaction.support.AbstractPlatformTransactionManager.getTransaction(AbstractPlatformTransactionManager.java:373)
 at com.example.springbootmybatis.service.Impl.StudentServiceImpl.updateStudentsThread(StudentServiceImpl.java:58)
 at com.example.springbootmybatis.StudentTest.lambda$updateStudentWithThreadsAndTrans$3(StudentTest.java:164)
 at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
 at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
 at java.lang.Thread.run(Thread.java:748)
Caused by: java.sql.SQLTransientConnectionException: HikariPool-1 - Connection is not available, request timed out after 30055ms.
 at com.zaxxer.hikari.pool.HikariPool.createTimeoutException(HikariPool.java:696)
 at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:197)
 at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:162)
 at com.zaxxer.hikari.HikariDataSource.getConnection(HikariDataSource.java:128)
 at org.springframework.jdbc.datasource.DataSourceTransactionManager.doBegin(DataSourceTransactionManager.java:265)
 ... 7 more

錯誤的大致意思時，不能為資料庫事務打開 jdbc Connection，連接在30s的時候超時了，由于前面啟動的十個執行緒需要等待主執行緒完成后才能提交，所以一直占用連接未釋放，造成后面的行程創建連接超時，

看錯誤日志中錯誤的來源是 HikariPool ，我們來重新配置一下這個連接池的引數，將最大連接數修改為100，具體配置如下：

# 連接池中允許的最小連接數，預設值：10
spring.datasource.hikari.minimum-idle=10
# 連接池中允許的最大連接數，預設值：10
spring.datasource.hikari.maximum-pool-size=100
# 自動提交
spring.datasource.hikari.auto-commit=true
# 一個連接idle狀態的最大時長（毫秒），超時則被釋放（retired），預設:10分鐘
spring.datasource.hikari.idle-timeout=30000
# 一個連接的生命時長（毫秒），超時而且沒被使用則被釋放（retired），預設:30分鐘，建議設定比資料庫超時時長少30秒
spring.datasource.hikari.max-lifetime=1800000
# 等待連接池分配連接的最大時長（毫秒），超過這個時長還沒可用的連接則發生SQLException， 預設:30秒

再次執行測驗發現沒有報錯，修改執行緒數為20又執行了一下，同樣執行成功了，另外，關注公眾號Java技術堆疊，在后臺回復：面試，可以獲取我整理的 Java 系列面試題和答案，非常齊全，

五、基于TransactionStatus集合來控制多執行緒事務提交

在同事推薦下我們使用事務集合來進行多執行緒事務控制，主要代碼如下

@Service
public class StudentsTransactionThread {

    @Autowired
    private StudentMapper studentMapper;
    @Autowired
    private StudentService studentService;
    @Autowired
    private PlatformTransactionManager transactionManager;

    List<TransactionStatus> transactionStatuses = Collections.synchronizedList(new ArrayList<TransactionStatus>());

    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = {Exception.class})
    public void updateStudentWithThreadsAndTrans() throws InterruptedException {

        //查詢總資料
        List<Student> allStudents = studentMapper.getAll();

        // 執行緒數量
        final Integer threadCount = 2;

        //每個執行緒處理的資料量
        final Integer dataPartionLength = (allStudents.size() + threadCount - 1) / threadCount;

        // 創建多執行緒處理任務
        ExecutorService studentThreadPool = Executors.newFixedThreadPool(threadCount);
        CountDownLatch threadLatchs = new CountDownLatch(threadCount);
        AtomicBoolean isError = new AtomicBoolean(false);
        try {
            for (int i = 0; i < threadCount; i++) {
                // 每個執行緒處理的資料
                List<Student> threadDatas = allStudents.stream()
                        .skip(i * dataPartionLength).limit(dataPartionLength).collect(Collectors.toList());
                studentThreadPool.execute(() -> {
                    try {
                        try {
                            studentService.updateStudentsTransaction(transactionManager, transactionStatuses, threadDatas);
                        } catch (Throwable e) {
                            e.printStackTrace();
                            isError.set(true);
                        }finally {
                            threadLatchs.countDown();
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                        isError.set(true);
                    }
                });
            }

            // 倒計時鎖設定超時時間 30s
            boolean await = threadLatchs.await(30, TimeUnit.SECONDS);
            // 判斷是否超時
            if (!await) {
                isError.set(true);
            }
        } catch (Throwable e) {
            e.printStackTrace();
            isError.set(true);
        }

        if (!transactionStatuses.isEmpty()) {
            if (isError.get()) {
                transactionStatuses.forEach(s -> transactionManager.rollback(s));
            } else {
                transactionStatuses.forEach(s -> transactionManager.commit(s));
            }
        }

        System.out.println("主執行緒完成");
    }
}
@Override
@Transactional(propagation = Propagation.REQUIRED, rollbackFor = {Exception.class})
public void updateStudentsTransaction(PlatformTransactionManager transactionManager, List<TransactionStatus> transactionStatuses, List<Student> students) {
    // 使用這種方式將事務狀態都放在同一個事務里面
    DefaultTransactionDefinition def = new DefaultTransactionDefinition();
    def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); // 事物隔離級別，開啟新事務，這樣會比較安全些，
    TransactionStatus status = transactionManager.getTransaction(def); // 獲得事務狀態
    transactionStatuses.add(status);

    students.forEach(s -> {
        // 更新教師資訊
        // String teacher = s.getTeacher();
        String newTeacher = "TNO_" + new Random().nextInt(100);
        s.setTeacher(newTeacher);
        studentMapper.update(s);
    });
    System.out.println("子執行緒：" + Thread.currentThread().getName());
}

由于這個中方式去前面方式相同，需要等待執行緒執行完成后才會提交事務，所有任會占用Jdbc連接池，如果執行緒數量超過連接池最大數量會產生連接超時，所以在使用程序中任要控制執行緒數量，

六、使用union連接多個select實作批量update

有些情況寫不支持，批量update，但支持insert 多條資料，這個時候可嘗試將需要更新的資料拼接成多條select 陳述句，然后使用union 連接起來，再使用update 關聯這個資料進行update，具體代碼演示如下：

update student,(
 (select  1 as id,'teacher_A' as teacher) union
 (select  2 as id,'teacher_A' as teacher) union
 (select  3 as id,'teacher_A' as teacher) union
 (select  4 as id,'teacher_A' as teacher)
    /* ....more data ... */
    ) as new_teacher
set
 student.teacher=new_teacher.teacher
where
 student.id=new_teacher.id

這種方式在Mysql 資料庫沒有配置 allowMultiQueries=true 也可以實作批量更新，

總結

對于大批量資料庫操作，使用手動事務提交可以很多程度上提高操作效率
多執行緒對資料庫進行操作時，并非執行緒數越多操作時間越快，按上述示例大約在2-5個執行緒時操作時間最快，
對于多執行緒阻塞事務提交時，執行緒數量不能過多，
如果能有辦法實作批量更新那是最好

著作權宣告：本文為CSDN博主「圣心」的原創文章，遵循CC 4.0 BY-SA著作權協議，轉載請附上原文出處鏈接及本宣告，原文鏈接：https://blog.csdn.net/qq273766764/article/details/119972911

近期熱文推薦：

1.1,000+ 道 Java面試題及答案整理(2022最新版)

2.勁爆！Java 協程要來了，，，

3.Spring Boot 2.x 教程，太全了！

4.別再寫滿屏的爆爆爆炸類了，試試裝飾器模式，這才是優雅的方式！！

5.《Java開發手冊（嵩山版）》最新發布，速速下載！

覺得不錯，別忘了隨手點贊+轉發哦！

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/514137.html

標籤：Java

上一篇：SynchronousQueue詳解

下一篇：10分鐘教你寫一個資料庫