• 炸锅了，Java多线程批量操作，居然有人不做事务控制

    项目代码基于：MySql 数据，开发框架为：SpringBoot、Mybatis

    开发语言为：Java8

    项目代码：https://gitee.com/john273766764/springboot-mybatis-threads

    文章目录

    • 前言

    • 循环操作的代码

    • 使用手动事务的操作代码

    • 尝试多线程进行数据修改

    • 基于两个CountDownLatch控制多线程事务提交

    • 基于TransactionStatus集合来控制多线程事务提交

    • 使用union连接多个select实现批量update

    • 总结

    前言

    公司业务中遇到一个需求，需要同时修改最多约5万条数据，而且还不支持批量或异步修改操作。于是只能写个for循环操作，但操作耗时太长，只能一步一步寻找其他解决方案。

    具体操作如下：

    一、循环操作的代码

    先写一个最简单的for循环代码，看看耗时情况怎么样。

    1. /***
    2.  * 一条一条依次对50000条数据进行更新操作
    3.  * 耗时：2m27s,1m54s
    4.  */
    5. @Test
    6. void updateStudent() {
    7.     List<Student> allStudents = studentMapper.getAll();
    8.     allStudents.forEach(s -> {
    9.         //更新教师信息
    10.         String teacher = s.getTeacher();
    11.         String newTeacher = "TNO_" + new Random().nextInt(100);
    12.         s.setTeacher(newTeacher);
    13.         studentMapper.update(s);
    14.     });
    15. }

    循环修改整体耗时约 1分54秒，且代码中没有手动事务控制应该是自动事务提交，所以每次操作事务都会提交所以操作比较慢，我们先对代码中添加手动事务控制，看查询效率怎样。

    二、使用手动事务的操作代码

    修改后的代码如下：

    1. @Autowired
    2. private DataSourceTransactionManager dataSourceTransactionManager;
    3.  
    4. @Autowired
    5. private TransactionDefinition transactionDefinition;
    6.  
    7. /**
    8.  * 由于希望更新操作 一次性完成，需要手动控制添加事务
    9.  * 耗时：24s
    10.  * 从测试结果可以看出，添加事务后插入数据的效率有明显的提升
    11.  */
    12. @Test
    13. void updateStudentWithTrans() {
    14.     List<Student> allStudents = studentMapper.getAll();
    15.     TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
    16.     try {
    17.         allStudents.forEach(s -> {
    18.             //更新教师信息
    19.             String teacher = s.getTeacher();
    20.             String newTeacher = "TNO_" + new Random().nextInt(100);
    21.             s.setTeacher(newTeacher);
    22.             studentMapper.update(s);
    23.         });
    24.         dataSourceTransactionManager.commit(transactionStatus);
    25.     } catch (Throwable e) {
    26.         dataSourceTransactionManager.rollback(transactionStatus);
    27.         throw e;
    28.     }
    29. }

    添加手动事务操控制后，整体耗时约 24秒，这相对于自动事务提交的代码，快了约5倍，对于大量循环数据库提交操作，添加手动事务可以有效提高操作效率。

    三、尝试多线程进行数据修改

    添加数据库手动事务后操作效率有明细提高，但还是比较长，接下来尝试多线程提交看是不是能够再快一些。

    先添加一个Service将批量修改操作整合一下，具体代码如下：

    StudentServiceImpl.java

    1. @Service
    2. public class StudentServiceImpl implements StudentService {
    3.     @Autowired
    4.     private StudentMapper studentMapper;
    5.  
    6.     @Autowired
    7.     private DataSourceTransactionManager dataSourceTransactionManager;
    8.  
    9.     @Autowired
    10.     private TransactionDefinition transactionDefinition;
    11.  
    12.     @Override
    13.     public void updateStudents(List<Student> students, CountDownLatch threadLatch) {
    14.         TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
    15.         System.out.println("子线程：" + Thread.currentThread().getName());
    16.         try {
    17.             students.forEach(s -> {
    18.                 // 更新教师信息
    19.                 // String teacher = s.getTeacher();
    20.                 String newTeacher = "TNO_" + new Random().nextInt(100);
    21.                 s.setTeacher(newTeacher);
    22.                 studentMapper.update(s);
    23.             });
    24.             dataSourceTransactionManager.commit(transactionStatus);
    25.             threadLatch.countDown();
    26.         } catch (Throwable e) {
    27.             e.printStackTrace();
    28.             dataSourceTransactionManager.rollback(transactionStatus);
    29.         }
    30.     }
    31. }

    批量测试代码，我们采用了多线程进行提交，修改后测试代码如下：

    1. @Autowired
    2. private DataSourceTransactionManager dataSourceTransactionManager;
    3.  
    4. @Autowired
    5. private TransactionDefinition transactionDefinition;
    6.  
    7. @Autowired
    8. private StudentService studentService;
    9.  
    10. /**
    11.  * 对用户而言，27s 任是一个较长的时间，我们尝试用多线程的方式来经行修改操作看能否加快处理速度
    12.  * 预计创建10个线程，每个线程进行5000条数据修改操作
    13.  * 耗时统计
    14.  * 1 线程数：1      耗时：25s
    15.  * 2 线程数：2      耗时：14s
    16.  * 3 线程数：5      耗时：15s
    17.  * 4 线程数：10     耗时：15s
    18.  * 5 线程数：100    耗时：15s
    19.  * 6 线程数：200    耗时：15s
    20.  * 7 线程数：500    耗时：17s
    21.  * 8 线程数：1000    耗时：19s
    22.  * 8 线程数：2000    耗时：23s
    23.  * 8 线程数：5000    耗时：29s
    24.  */
    25. @Test
    26. void updateStudentWithThreads() {
    27.     //查询总数据
    28.     List<Student> allStudents = studentMapper.getAll();
    29.     // 线程数量
    30.     final Integer threadCount = 100;
    31.  
    32.     //每个线程处理的数据量
    33.     final Integer dataPartionLength = (allStudents.size() + threadCount - 1) / threadCount;
    34.  
    35.     // 创建多线程处理任务
    36.     ExecutorService studentThreadPool = Executors.newFixedThreadPool(threadCount);
    37.     CountDownLatch threadLatchs = new CountDownLatch(threadCount);
    38.  
    39.     for (int i = 0; i < threadCount; i++) {
    40.         // 每个线程处理的数据
    41.         List<Student> threadDatas = allStudents.stream()
    42.                 .skip(i * dataPartionLength).limit(dataPartionLength).collect(Collectors.toList());
    43.         studentThreadPool.execute(() -> {
    44.             studentService.updateStudents(threadDatas, threadLatchs);
    45.         });
    46.     }
    47.     try {
    48.         // 倒计时锁设置超时时间 30s
    49.         threadLatchs.await(30, TimeUnit.SECONDS);
    50.     } catch (Throwable e) {
    51.         e.printStackTrace();
    52.     }
    53.  
    54.     System.out.println("主线程完成");
    55. }

    多线程提交修改时，我们尝试了不同线程数对提交速度的影响，具体可以看下面表格，

    多线程修改50000条数据时 不同线程数耗时对比（秒）

    根据表格，我们线程数增大提交速度并非一直增大，在当前情况下约在2-5个线程数时，提交速度最快（实际线程数还是需要根据服务器配置实际测试）。

    四、基于两个CountDownLatch控制多线程事务提交

    由于多线程提交时，每个线程事务时单独的，无法保证一致性，我们尝试给多线程添加事务控制，来保证每个线程都是在插入数据完成后在提交事务，

    这里我们使用两个 CountDownLatch 来控制主线程与子线程事务提交，并设置了超时时间为 30 秒。我们对代码进行了一点修改：

    1. @Override
    2. public void updateStudentsThread(List<Student> students, CountDownLatch threadLatch, CountDownLatch mainLatch, StudentTaskError taskStatus) {
    3.     TransactionStatus transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
    4.     System.out.println("子线程：" + Thread.currentThread().getName());
    5.     try {
    6.         students.forEach(s -> {
    7.             // 更新教师信息
    8.             // String teacher = s.getTeacher();
    9.             String newTeacher = "TNO_" + new Random().nextInt(100);
    10.             s.setTeacher(newTeacher);
    11.             studentMapper.update(s);
    12.         });
    13.     } catch (Throwable e) {
    14.         taskStatus.setIsError();
    15.     } finally {
    16.         threadLatch.countDown(); // 切换到主线程执行
    17.     }
    18.     try {
    19.         mainLatch.await();  //等待主线程执行
    20.     } catch (Throwable e) {
    21.         taskStatus.setIsError();
    22.     }
    23.     // 判断是否有错误，如有错误 就回滚事务
    24.     if (taskStatus.getIsError()) {
    25.         dataSourceTransactionManager.rollback(transactionStatus);
    26.     } else {
    27.         dataSourceTransactionManager.commit(transactionStatus);
    28.     }
    29. }
    1. /**
    2.  * 由于每个线程都是单独的事务，需要添加对线程事务的统一控制
    3.  * 我们这边使用两个 CountDownLatch 对子线程的事务进行控制
    4.  */
    5. @Test
    6. void updateStudentWithThreadsAndTrans() {
    7.     //查询总数据
    8.     List<Student> allStudents = studentMapper.getAll();
    9.     // 线程数量
    10.     final Integer threadCount = 4;
    11.  
    12.     //每个线程处理的数据量
    13.     final Integer dataPartionLength = (allStudents.size() + threadCount - 1) / threadCount;
    14.  
    15.     // 创建多线程处理任务
    16.     ExecutorService studentThreadPool = Executors.newFixedThreadPool(threadCount);
    17.     CountDownLatch threadLatchs = new CountDownLatch(threadCount); // 用于计算子线程提交数量
    18.     CountDownLatch mainLatch = new CountDownLatch(1); // 用于判断主线程是否提交
    19.     StudentTaskError taskStatus = new StudentTaskError(); // 用于判断子线程任务是否有错误
    20.  
    21.     for (int i = 0; i < threadCount; i++) {
    22.         // 每个线程处理的数据
    23.         List<Student> threadDatas = allStudents.stream()
    24.                 .skip(i * dataPartionLength).limit(dataPartionLength)
    25.                 .collect(Collectors.toList());
    26.         studentThreadPool.execute(() -> {
    27.             studentService.updateStudentsThread(threadDatas, threadLatchs, mainLatch, taskStatus);
    28.         });
    29.     }
    30.     try {
    31.         // 倒计时锁设置超时时间 30s
    32.         boolean await = threadLatchs.await(30, TimeUnit.SECONDS);
    33.         if (!await) { // 等待超时，事务回滚
    34.             taskStatus.setIsError();
    35.         }
    36.     } catch (Throwable e) {
    37.         e.printStackTrace();
    38.         taskStatus.setIsError();
    39.     }
    40.     mainLatch.countDown(); // 切换到子线程执行
    41.     studentThreadPool.shutdown(); //关闭线程池
    42.  
    43.     System.out.println("主线程完成");
    44. }

    本想再次测试一下不同线程数对执行效率的影响时，发现当线程数超过10个时，执行时就报错。具体错误内容如下：

    1. 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.
    2.  at org.springframework.jdbc.datasource.DataSourceTransactionManager.doBegin(DataSourceTransactionManager.java:309)
    3.  at org.springframework.transaction.support.AbstractPlatformTransactionManager.startTransaction(AbstractPlatformTransactionManager.java:400)
    4.  at org.springframework.transaction.support.AbstractPlatformTransactionManager.getTransaction(AbstractPlatformTransactionManager.java:373)
    5.  at com.example.springbootmybatis.service.Impl.StudentServiceImpl.updateStudentsThread(StudentServiceImpl.java:58)
    6.  at com.example.springbootmybatis.StudentTest.lambda$updateStudentWithThreadsAndTrans$3(StudentTest.java:164)
    7.  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    8.  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    9.  at java.lang.Thread.run(Thread.java:748)
    10. Caused by: java.sql.SQLTransientConnectionException: HikariPool-1 - Connection is not available, request timed out after 30055ms.
    11.  at com.zaxxer.hikari.pool.HikariPool.createTimeoutException(HikariPool.java:696)
    12.  at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:197)
    13.  at com.zaxxer.hikari.pool.HikariPool.getConnection(HikariPool.java:162)
    14.  at com.zaxxer.hikari.HikariDataSource.getConnection(HikariDataSource.java:128)
    15.  at org.springframework.jdbc.datasource.DataSourceTransactionManager.doBegin(DataSourceTransactionManager.java:265)
    16.  ... 7 more

    错误的大致意思时，不能为数据库事务打开 jdbc Connection，连接在30s的时候超时了。由于前面启动的十个线程需要等待主线程完成后才能提交，所以一直占用连接未释放，造成后面的进程创建连接超时。

    看错误日志中错误的来源是 HikariPool ，我们来重新配置一下这个连接池的参数，将最大连接数修改为100，具体配置如下：

    1. # 连接池中允许的最小连接数。缺省值：10
    2. spring.datasource.hikari.minimum-idle=10
    3. # 连接池中允许的最大连接数。缺省值：10
    4. spring.datasource.hikari.maximum-pool-size=100
    5. # 自动提交
    6. spring.datasource.hikari.auto-commit=true
    7. # 一个连接idle状态的最大时长（毫秒），超时则被释放（retired），缺省:10分钟
    8. spring.datasource.hikari.idle-timeout=30000
    9. # 一个连接的生命时长（毫秒），超时而且没被使用则被释放（retired），缺省:30分钟，建议设置比数据库超时时长少30秒
    10. spring.datasource.hikari.max-lifetime=1800000
    11. # 等待连接池分配连接的最大时长（毫秒），超过这个时长还没可用的连接则发生SQLException， 缺省:30秒

    再次执行测试发现没有报错，修改线程数为20又执行了一下，同样执行成功了。

    五、基于TransactionStatus集合来控制多线程事务提交

    在同事推荐下我们使用事务集合来进行多线程事务控制，主要代码如下

    1. @Service
    2. public class StudentsTransactionThread {
    3.  
    4.     @Autowired
    5.     private StudentMapper studentMapper;
    6.     @Autowired
    7.     private StudentService studentService;
    8.     @Autowired
    9.     private PlatformTransactionManager transactionManager;
    10.  
    11.     List<TransactionStatus> transactionStatuses = Collections.synchronizedList(new ArrayList<TransactionStatus>());
    12.  
    13.     @Transactional(propagation = Propagation.REQUIRED, rollbackFor = {Exception.class})
    14.     public void updateStudentWithThreadsAndTrans() throws InterruptedException {
    15.  
    16.         //查询总数据
    17.         List<Student> allStudents = studentMapper.getAll();
    18.  
    19.         // 线程数量
    20.         final Integer threadCount = 2;
    21.  
    22.         //每个线程处理的数据量
    23.         final Integer dataPartionLength = (allStudents.size() + threadCount - 1) / threadCount;
    24.  
    25.         // 创建多线程处理任务
    26.         ExecutorService studentThreadPool = Executors.newFixedThreadPool(threadCount);
    27.         CountDownLatch threadLatchs = new CountDownLatch(threadCount);
    28.         AtomicBoolean isError = new AtomicBoolean(false);
    29.         try {
    30.             for (int i = 0; i < threadCount; i++) {
    31.                 // 每个线程处理的数据
    32.                 List<Student> threadDatas = allStudents.stream()
    33.                         .skip(i * dataPartionLength).limit(dataPartionLength).collect(Collectors.toList());
    34.                 studentThreadPool.execute(() -> {
    35.                     try {
    36.                         try {
    37.                             studentService.updateStudentsTransaction(transactionManager, transactionStatuses, threadDatas);
    38.                         } catch (Throwable e) {
    39.                             e.printStackTrace();
    40.                             isError.set(true);
    41.                         }finally {
    42.                             threadLatchs.countDown();
    43.                         }
    44.                     } catch (Exception e) {
    45.                         e.printStackTrace();
    46.                         isError.set(true);
    47.                     }
    48.                 });
    49.             }
    50.  
    51.             // 倒计时锁设置超时时间 30s
    52.             boolean await = threadLatchs.await(30, TimeUnit.SECONDS);
    53.             // 判断是否超时
    54.             if (!await) {
    55.                 isError.set(true);
    56.             }
    57.         } catch (Throwable e) {
    58.             e.printStackTrace();
    59.             isError.set(true);
    60.         }
    61.  
    62.         if (!transactionStatuses.isEmpty()) {
    63.             if (isError.get()) {
    64.                 transactionStatuses.forEach(s -> transactionManager.rollback(s));
    65.             } else {
    66.                 transactionStatuses.forEach(s -> transactionManager.commit(s));
    67.             }
    68.         }
    69.  
    70.         System.out.println("主线程完成");
    71.     }
    72. }
    1. @Override
    2. @Transactional(propagation = Propagation.REQUIRED, rollbackFor = {Exception.class})
    3. public void updateStudentsTransaction(PlatformTransactionManager transactionManager, List<TransactionStatus> transactionStatuses, List<Student> students) {
    4.     // 使用这种方式将事务状态都放在同一个事务里面
    5.     DefaultTransactionDefinition def = new DefaultTransactionDefinition();
    6.     def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); // 事物隔离级别，开启新事务，这样会比较安全些。
    7.     TransactionStatus status = transactionManager.getTransaction(def); // 获得事务状态
    8.     transactionStatuses.add(status);
    9.  
    10.     students.forEach(s -> {
    11.         // 更新教师信息
    12.         // String teacher = s.getTeacher();
    13.         String newTeacher = "TNO_" + new Random().nextInt(100);
    14.         s.setTeacher(newTeacher);
    15.         studentMapper.update(s);
    16.     });
    17.     System.out.println("子线程：" + Thread.currentThread().getName());
    18. }

    由于这个中方式去前面方式相同，需要等待线程执行完成后才会提交事务，所有任会占用Jdbc连接池，如果线程数量超过连接池最大数量会产生连接超时。所以在使用过程中任要控制线程数量，

    六、使用union连接多个select实现批量update

    有些情况写不支持，批量update，但支持insert 多条数据，这个时候可尝试将需要更新的数据拼接成多条select 语句，然后使用union 连接起来，再使用update 关联这个数据进行update，具体代码演示如下：

    1. update student,(
    2.  (select  1 as id,'teacher_A' as teacher) union
    3.  (select  2 as id,'teacher_A' as teacher) union
    4.  (select  3 as id,'teacher_A' as teacher) union
    5.  (select  4 as id,'teacher_A' as teacher)
    6.     /* ....more data ... */
    7.     ) as new_teacher
    8. set
    9.  student.teacher=new_teacher.teacher
    10. where
    11.  student.id=new_teacher.id

    这种方式在Mysql 数据库没有配置 allowMultiQueries=true 也可以实现批量更新。

    总结

    • 对于大批量数据库操作，使用手动事务提交可以很多程度上提高操作效率

    • 多线程对数据库进行操作时，并非线程数越多操作时间越快，按上述示例大约在2-5个线程时操作时间最快。

    • 对于多线程阻塞事务提交时，线程数量不能过多。

    • 如果能有办法实现批量更新那是最好

    PS：防止找不到本篇文章，可以收藏点赞，方便翻阅查找哦

     

  • 相关阅读:
    汉字风格迁移篇---中文字体风格迁移的多任务对抗学习
    【Electron】vue+electron应用设置菜单
    数据治理之主数据管理MDM
    （二十）mmdetection源码解读：config配置文件models Faster R-CNN
    Vue路由&nodejs环境搭建
    Java-ForkJoinPool(线程池-工作窃取算法)
    modinfo对比内核版本号
    Java 网络编程 —— 非阻塞式编程
    线段树什么的不是简简单单嘛，我教你！：基础篇
    催款神器！博途分期付款程序：分多个阶段定时锁机实现时间锁，再利用随机数生成动态密码解锁！适用于S71200/1500PLC
  • 原文地址：https://blog.csdn.net/best_luxi/article/details/124950841