• 【二十六】springboot实现多线程事务处理

     springboot篇章整体栏目： 

    【一】springboot整合swagger（超详细

    【二】springboot整合swagger（自定义）（超详细）

    【三】springboot整合token（超详细）

    【四】springboot整合mybatis-plus（超详细）（上）

    【五】springboot整合mybatis-plus（超详细）（下）

    【六】springboot整合自定义全局异常处理

    【七】springboot整合redis（超详细）

    【八】springboot整合AOP实现日志操作（超详细）

    【九】springboot整合定时任务（超详细）

    【十】springboot整合redis实现启动服务即将热点数据保存在全局以及redis（超详细）

    【十一】springboot整合quartz实现定时任务优化（超详细）

    【十二】springboot整合线程池解决高并发（超详细,保你理解）

    【十三】springboot整合异步调用并获取返回值（超详细）

    【十四】springboot整合WebService（超详细）

    【十五】springboot整合WebService（关于传参数）（超详细）

    【十六】springboot整合WebSocket（超详细）

    【十七】springboot整合WebSocket实现聊天室（超详细）

    【十八】springboot实现自定义全局异常处理

    【十九】springboot整合ElasticSearch实战（万字篇）

    【二十】springboot整合过滤器实战

    【二十一】springboot整合拦截器实战并对比过滤器

    【二十二】springboot整合activiti7（1） 实战演示篇

    【二十三】springboot整合spring事务详解以及实战

    【二十四】springboot使用EasyExcel和线程池实现多线程导入Excel数据

    【二十五】springboot整合jedis和redisson布隆过滤器处理缓存穿透

            在前面二十四章做了一个springboot使用EasyExcel和线程池实现多线程导入Excel数据的demo，在写时忘了做事务处理，评论区有个大佬提出来了，这章就对二十四章的代码做一个改造，完善多线程的事务处理。

            对于springboot的事务处理，前面在二十三章也做过springboot整合spring事务详解以及实战的学习，但是在多线程时，这个东西并不适用，本章就通过手写事务处理（编程式事务处理）。

            由于本章是针对二十四章的批量导入功能的扩展，所有不会再写事务处理不相关的（二十四章的内容）介绍了。

    qq交流群导航——>231378628

    目录

    一、阐述目的与实现方式

    二、手动让子线程报错

    三、改造主线程

    四、改造子线程

    五、测试

    一、阐述目的与实现方式

            前面章节实现的多线程处理excel导入功能，如果一个子线程出现错误，结果会是那个子线程的数据处理不了，而其他子线程的数据仍然正常处理保存，并不会存在事务处理的情况，本章改造代码实现事务处理，所有线程正常执行才会保存数据，否则就回滚。大致如下：

    二、手动让子线程报错

            为了后面测试事务回滚，手动让某个子线程报错，比如名为线程3的子线程，如下：

     三、改造主线程

            根据上面图的思路，首先改造主线程的代码，整体代码如下：

    1. package com.swagger.demo.service;
    2.  
    3. import com.alibaba.excel.context.AnalysisContext;
    4. import com.alibaba.excel.event.AnalysisEventListener;
    5. import com.swagger.demo.config.SpringJobBeanFactory;
    6. import com.swagger.demo.mapper.DeadManMapper;
    7. import com.swagger.demo.model.entity.DeadManExcelData;
    8. import com.swagger.demo.thread.DeadManThread;
    9. import lombok.extern.slf4j.Slf4j;
    10. import org.springframework.stereotype.Component;
    11. import org.springframework.stereotype.Service;
    12.  
    13. import java.util.ArrayList;
    14. import java.util.Collections;
    15. import java.util.List;
    16. import java.util.concurrent.*;
    17. import java.util.concurrent.atomic.AtomicBoolean;
    18.  
    19. /**
    20.  * @author zrc
    21.  * @version 1.0
    22.  * @description: TODO 最新入狱名单导入监听器
    23.  *
    24.  * @date 2022/5/30 15:56
    25.  */
    26. @Service
    27. @Slf4j
    28. @Component
    29. public class DeadManExcelListener extends AnalysisEventListener {
    30.  
    31.     /**
    32.      * 多线程保存集合，使用线程安全集合
    33.      */
    34.     private List list = Collections.synchronizedList(new ArrayList<>());
    35.  
    36.     /**
    37.      * 创建线程池必要参数
    38.      */
    39.     private static final int CORE_POOL_SIZE = 10; // 核心线程数
    40.     private static final int MAX_POOL_SIZE = 100; // 最大线程数
    41.     private static final int QUEUE_CAPACITY = 100; // 队列大小
    42.     private static final Long KEEP_ALIVE_TIME = 1L; // 存活时间
    43.  
    44.     public List getData(){
    45.         return list;
    46.     }
    47.  
    48.     public DeadManExcelListener(){
    49.  
    50.     }
    51.  
    52.     public void setData(List deadManExcelDataList){
    53.         this.list = deadManExcelDataList;
    54.     }
    55.  
    56.     @Override
    57.     public void invoke(DeadManExcelData deadManExcelData, AnalysisContext analysisContext) {
    58.         if(deadManExcelData!=null){
    59.             list.add(deadManExcelData);
    60.         }
    61.     }
    62.  
    63.     /**
    64.      * 多线程方式保存
    65.      * @param analysisContext
    66.      */
    67.     @Override
    68.     public void doAfterAllAnalysed(AnalysisContext analysisContext) {
    69.         log.info("解析结束，开始插入数据");
    70.  
    71.         // 创建线程池
    72.         ExecutorService executor = new ThreadPoolExecutor(CORE_POOL_SIZE,
    73.                 MAX_POOL_SIZE,
    74.                 KEEP_ALIVE_TIME,
    75.                 TimeUnit.SECONDS,
    76.                 new ArrayBlockingQueue<>(QUEUE_CAPACITY),
    77.                 new ThreadPoolExecutor.CallerRunsPolicy());
    78.  
    79.         // 指定每个线程需要处理的导入数量，假设每个线程处理15000条，注意配合上面线程池的大小
    80.         int singleThreadDealCount = 15000;
    81.  
    82.         // 根据假设每个线程需要处理的数量以及总数，计算需要提交到线程池的线程数量
    83.         int threadSize=(list.size()/singleThreadDealCount)+1;
    84.  
    85.         // 计算需要导入的数据总数，用于拆分时线程需要处理数据时使用
    86.         int rowSize = list.size() + 1;
    87.  
    88.         // 测试开始时间
    89.         long startTime = System.currentTimeMillis();
    90.  
    91.         // 申明该线程需要处理数据的开始位置
    92.         int startPosition = 0;
    93.  
    94.         // 申明该线程需要处理数据的结束位置
    95.         int endPosition = 0;
    96.  
    97.         // 为了让每个线程执行完后回到当前线程，使用CountDownLatch，值为线程数，每次线程执行完就会执行countDown方法减1，为0后回到主线程，也就是当前线程执行后续的代码
    98.         CountDownLatch count = new CountDownLatch(threadSize);
    99.  
    100.         // 用来控制主线程回到子线程
    101.         CountDownLatch mainCount = new CountDownLatch(1);
    102.  
    103.         // 用来控制最终回到主线程
    104.         CountDownLatch endCount = new CountDownLatch(threadSize);
    105.  
    106.         // 用来存放子线程的处理结果，若出错就保存一个false
    107.         CopyOnWriteArrayList sonResult = new CopyOnWriteArrayList();
    108.  
    109.         // 使用线程安全的对象存储，保存主线程最后总的判断结果，是提交还是回滚
    110.         AtomicBoolean ifSubmit = new AtomicBoolean(true);
    111.  
    112.         // 计算每个线程要处理的数据
    113.         for(int i=0;i
    114.             // 如果是最后一个线程，为保证程序不发生空指针异常，特殊判断结束位置
    115.             if((i+1)==threadSize){
    116.                 // 计算开始位置
    117.                 startPosition = (i * singleThreadDealCount);
    118.                 // 当前线程为划分的最后一个线程，则取总数据的最后为此线程的结束位置
    119.                 endPosition = rowSize-1;
    120.             }else{
    121.                 // 计算开始位置
    122.                 startPosition = (i * singleThreadDealCount);
    123.                 // 计算结束位置
    124.                 endPosition = (i + 1) * singleThreadDealCount;
    125.             }
    126.  
    127.             DeadManMapper deadManMapper = SpringJobBeanFactory.getBean(DeadManMapper.class);
    128.  
    129.             DeadManThread thread = new DeadManThread(count,deadManMapper,list,startPosition,endPosition
    130.             ,sonResult,mainCount,ifSubmit,endCount);
    131.             executor.execute(thread);
    132.         }
    133.         try {
    134.             count.await();
    135.             for (Boolean resp : sonResult) {
    136.                 if (!resp) {
    137.                     // 只要有一个子线程出异常，就设置最终结果为回滚
    138.                     log.info("主线程:有线程执行失败，所有线程需要回滚");
    139.                     ifSubmit.set(false);
    140.                     break;
    141.                 }
    142.             }
    143.         } catch (InterruptedException e) {
    144.             e.printStackTrace();
    145.         }finally {
    146.             // 回到子线程处理回滚或者提交事务
    147.             mainCount.countDown();
    148.         }
    149.  
    150.         try {
    151.             endCount.await();
    152.             // 逻辑处理完，关闭线程池
    153.             executor.shutdown();
    154.             long endTime = System.currentTimeMillis();
    155.             log.info("总耗时："+(endTime-startTime));
    156.         } catch (InterruptedException e) {
    157.             e.printStackTrace();
    158.         }
    159.     }
    160.  
    161. }

            新增如下4个参数（count是前面章节的已有的）。

     

    PS：CountDownLatch类前面有讲过，通过await和countDown方法能够方便的实现多个线程之间的来回切换。

    CopyOnWriteArrayList和AtomicBoolean是为了能够线程安全的保存多个线程共同使用的数据。 

            接着，重写DeadManThread线程类的构造方法，将上面新增的四个参数通过构造方法传给子线程。然后调用记录子线程第一次的cout的await方法，等待子线程第一次执行完毕，回到主线程继续执行。回到主线程后，主线程判断子线程第一次执行完后保存的返回集，判断是否存在false（子线程若报错，保存false，否则保存true）。若存在false就将idsubmit设置为false，意思是需要回滚数据，然后调用记录主线程执行的mainCount的countDown方法，让主线程执行完毕，回到子线程调用mainCount.countDown的位置继续子线程执行。

     

            当子线程根据ifSubmit判断完进行回滚还是提交事务操作后，回到主线程，主线程关闭线程池。

    四、改造子线程

            接着改造子线程，整体代码如下：

    1. package com.swagger.demo.thread;
    2.  
    3. import com.swagger.demo.config.SpringJobBeanFactory;
    4. import com.swagger.demo.mapper.DeadManMapper;
    5. import com.swagger.demo.model.entity.DeadMan;
    6. import com.swagger.demo.model.entity.DeadManExcelData;
    7. import lombok.extern.slf4j.Slf4j;
    8. import org.springframework.beans.BeanUtils;
    9. import org.springframework.jdbc.datasource.DataSourceTransactionManager;
    10. import org.springframework.stereotype.Component;
    11. import org.springframework.transaction.TransactionDefinition;
    12. import org.springframework.transaction.TransactionStatus;
    13. import org.springframework.transaction.support.DefaultTransactionDefinition;
    14.  
    15. import java.util.ArrayList;
    16. import java.util.Collections;
    17. import java.util.List;
    18. import java.util.concurrent.CopyOnWriteArrayList;
    19. import java.util.concurrent.CountDownLatch;
    20. import java.util.concurrent.atomic.AtomicBoolean;
    21.  
    22. /**
    23.  * @author zrc
    24.  * @version 1.0
    25.  * @description TODO
    26.  * @date 2022/7/22 15:40
    27.  */
    28. @Component
    29. @Slf4j
    30. public class DeadManThread implements Runnable{
    31.  
    32.     /**
    33.      * 当前线程需要处理的总数据中的开始位置
    34.      */
    35.     private int startPosition;
    36.  
    37.     /**
    38.      * 当前线程需要处理的总数据中的结束位置
    39.      */
    40.     private int endPosition;
    41.  
    42.     /**
    43.      * 需要处理的未拆分之前的全部数据
    44.      */
    45.     private List list = Collections.synchronizedList(new ArrayList<>());
    46.  
    47.     /**
    48.      * 记录子线程第一次执行是否完成
    49.      */
    50.     private CountDownLatch count;
    51.  
    52.     private DeadManMapper deadManMapper;
    53.  
    54.     /**
    55.      * 保存每个线程的执行结果
    56.      */
    57.     private CopyOnWriteArrayList sonResult;
    58.  
    59.     /**
    60.      * 记录主线程是否执行过判断每个线程的执行结果这个操作
    61.      */
    62.     private CountDownLatch mainCount;
    63.  
    64.     /**
    65.      * 记录主线程对每个线程的执行结果的判断
    66.      */
    67.     private AtomicBoolean ifSubmit;
    68.  
    69.     /**
    70.      * 声明该子线程的事务管理器
    71.      */
    72.     private DataSourceTransactionManager dataSourceTransactionManager;
    73.  
    74.     /**
    75.      * 声明该线程事务的状态
    76.      */
    77.     private TransactionStatus status;
    78.  
    79.     /**
    80.      * 记录子线程第二次执行是否完成
    81.      */
    82.     private CountDownLatch endCount;
    83.  
    84.     public DeadManThread() {
    85.  
    86.     }
    87.  
    88.     public DeadManThread(CountDownLatch count, DeadManMapper deadManMapper, List list
    89.             , int startPosition, int endPosition, CopyOnWriteArrayList sonResult,CountDownLatch mainCount
    90.     ,AtomicBoolean ifSubmit,CountDownLatch endCount) {
    91.         this.startPosition = startPosition;
    92.         this.endPosition = endPosition;
    93.         this.deadManMapper = deadManMapper;
    94.         this.list = list;
    95.         this.count = count;
    96.         this.sonResult = sonResult;
    97.         this.mainCount = mainCount;
    98.         this.ifSubmit = ifSubmit;
    99.         this.endCount = endCount;
    100.     }
    101.  
    102.     @Override
    103.     public void run() {
    104.         try{
    105.             dataSourceTransactionManager = SpringJobBeanFactory.getBean(DataSourceTransactionManager.class);
    106.             DefaultTransactionDefinition def = new DefaultTransactionDefinition();
    107.             def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
    108.             status = dataSourceTransactionManager.getTransaction(def);
    109.             if(Thread.currentThread().getName().contains("3")){
    110.                 throw new RuntimeException("线程3出问题了");
    111.             }
    112.             List deadManList = new ArrayList<>();
    113.             List newList = list.subList(startPosition, endPosition);
    114.             // 将EasyExcel对象和实体类对象进行一个转换
    115.             for (DeadManExcelData deadManExcelData : newList) {
    116.                 DeadMan deadMan = new DeadMan();
    117.                 BeanUtils.copyProperties(deadManExcelData, deadMan);
    118.                 deadManList.add(deadMan);
    119.             }
    120.             // 批量新增
    121.             deadManMapper.insertBatchSomeColumn(deadManList);
    122.             sonResult.add(true);
    123.             } catch (Exception e) {
    124.             e.printStackTrace();
    125.             sonResult.add(false);
    126.             } finally {
    127.             // 当一个线程执行完了计数要减一不然这个线程会被一直挂起
    128.             count.countDown();
    129.             try {
    130.                 log.info(Thread.currentThread().getName() + "：准备就绪,等待其他线程结果,判断是否事务提交");
    131.                 mainCount.await();
    132.             } catch (InterruptedException e) {
    133.                 e.printStackTrace();
    134.             }
    135.             if (ifSubmit.get()) {
    136.                 dataSourceTransactionManager.commit(status);
    137.                 log.info(Thread.currentThread().getName() + "：事务提交");
    138.             } else {
    139.                 dataSourceTransactionManager.rollback(status);
    140.                 log.info(Thread.currentThread().getName() + "：事务回滚");
    141.             }
    142.             // 执行完所有逻辑，等待主线程执行
    143.             endCount.countDown();
    144.         }
    145.     }
    146. }

            先改造构造方法，用于接受主线程传过来的参数，并设置到自己的内部私有。

            然后改造run方法。

            先开启自己的事务，并保存事务状态，用于后面执行提交或者回滚操作。数据处理完后，若正常结束将线程安全的返回值集合变量保存一个true，否则保存false，并执行记录第一线程执行的count的countDown方法，等待所有子线程执行完后，返回主线程执行刚才上面讲的判断sonResult的代码，等主线程执行完判断并设置ifSubmit的值后，回到子线程执行main.await之后的代码。

            如果ifSubmit是false就回滚，否则就提交，执行完后执行记录子线程第二次执行的endCount的countDown方法，等待子线程全部执行完后，回到主线程，主线程执行关闭线程池的逻辑，结束。 

    五、测试

            代码写完，测试一下。测试之前数据：

            现在是线程3会报错，调用接口测试。

            事务成功回滚，数据没有提交。若删除手动抛异常的代码，让程序正常执行，如下：

     

             数据提交成功，事务正常处理。

  • 相关阅读:
    【SA8295P 源码分析 (二)】37 - OpenWFD Server 启动流程 之 openwfd_server.c main 函数源码分析
    GDAL+Java实现获取对应栅格影像经纬度对应的像素值
    Java如何从字符串中提取数字
    相似度系列-6：单维度方法：Evaluating Coherence in Dialogue Systems using Entailment
    异步函数（async/await）
    RT-DETR手把手教程，注意力机制如何添加在网络的不同位置进行创新优化，EMA注意力为案列
    【Python大数据笔记_day07_hive中的分区表、分桶表以及一些特殊类型】
    ALTERA FPGA IPCORE核之单口RAM详细教程
    【微服务治理之监控APM】系统监控架构概述
    Roson的Qt之旅 #117 QTcpSocket和QUdpSocket详细介绍
  • 原文地址：https://blog.csdn.net/weixin_56995925/article/details/126331000