😊 @ 作者: 一恍过去
🎊 @ 社区: Java技术栈交流
🎉 @ 主题: CountDownLatch实战应用——批量数据多线程协调异步处理(子线程执行事务回滚)
⏱️ @ 创作时间: 2023年11月26日

CountDownLatch实战应用——批量数据多线程协调异步处理(子线程执行事务回滚)-LMLPHP

前言

通过CountDownLatch开启多个子线程,由子线程完成数据的处理,子线程完成数据处理后进行等待,直到所有的子线程完成数据处理后,再判断是否进行回滚,如果需要回滚则所有线程执行回滚操作

如果需要由子线程处理完数据,但是由主线程进行事务提交或者回滚,参考:https://lhz1219.blog.csdn.net/article/details/134630258

1、概述

CountDownLatch是一个同步器工具类,用来协调多个线程之间的同步,能够使一个线程在等待另外一些线程完成各自工作之后,再继续执行,不可重置使用。

2、实现

使用一个计数器进行实现,计数器初始值为线程的数量,当每一个线程完成自己任务后,计数器的值就会减一,当计数器的值为0时,在CountDownLatch上等待的线程就可以恢复执行接下来的任务。

3、方法说明:

  • public void countDown():递减锁存器的计数,如果计数到达零,则释放所有等待的线程。如果当前计数大于零,则将计数减少.
  • public viod await() /boolean await(long timeout,TimeUnit unit) :使当前线程在锁存器倒计数至零之前一直等待,除非线程被中断或超出了指定的等待时间。如果当前计数为零,则此方法立刻返回true值。当线程调用了CountDownLatch对象的该方法后,当前线程会被阻塞,直到下面的情况之一发生才会返回:
    • 如果计数到达零,则该方法返回true值。
    • 如果当前线程,在进入此方法时已经设置了该线程的中断状态;或者在等待时被中断,则抛出InterruptedException,并且清除当前线程的已中断状态。
    • 如果超出了指定的等待时间,则返回值为false。如果该时间小于等于零,则该方法根本不会等待。参数:timeout-要等待的最长时间、unit-timeout 参数的时间单位

4、代码实例

有用到hutool的工具包,pom如下:

        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.0.7</version>
        </dependency>

Controller:

@RestController
@RequestMapping("/test")
@Slf4j
public class TestController {

    @Resource
    private CountDownService countDownService;
   

    /**
     * CountDownLatch实现多线程(多个子线程)异步事务处理数据,全部子线程回滚数据
     *
     * @return
     */
    @ApiOperation(value = "测试CountDownLatch", notes = "测试")
    @ApiOperationSupport(order = 5)
    @GetMapping("/countDown/handleDataSonBack")
    public String handleDataSonBack() {
        countDownService.handleDataSonBack();
        return "success";
    }

Sevice:

@Service
@Slf4j
public class CountDownService {
    @Resource
    private TestMapper testMapper;

    @Resource
    private ApplicationContext applicationContext;
    
	/**
     * CountDownLatch实现多线程(多个子线程)异步事务处理数据,全部子线程回滚数据
     *
     * @return
     */
    @Transactional(rollbackFor = Exception.class)
    public void handleDataSonBack() {
        List<TestEntity> testList = getData();
        AtomicBoolean errorTag = new AtomicBoolean(false);
        long start = System.currentTimeMillis();
        // 使用多线程对list集合进行分批次处理,实际情况可以根据具体耗时来决定
        // 比如:一万条数据,每条单独处理需要50ms,按批次一个线程处理200条数据,分为50个批次,实际情况根据业务来定
        // 需要使用hutool工具类进行分组
        // 由于开启了事务回滚,异步的线程数量要小于,dataSource中配置的maximum-pool-size数量
        List<List<TestEntity>> splitList = CollUtil.split(testList, 200);
        // 设置countDown大小
        CountDownLatch countDownLatch = new CountDownLatch(splitList.size());

        // 再创建一个CountDownLatch,大小固定为一,用于子线程相互等待,最后确定是否回滚
        CountDownLatch errorCountDown = new CountDownLatch(1);

        // 异步调用其他Service,执行业务处理
        CountDownService bean = applicationContext.getBean(CountDownService.class);
        // 简单创建一个线程池,这里的线程池可以自定义,为了方便直接使用
        ExecutorService executorService = Executors.newCachedThreadPool();
        splitList.forEach(list -> {
            // 线程处理
            executorService.execute(() -> {
                bean.handleDataAsyncSonBack(list, countDownLatch, errorCountDown, errorTag);
            });
        });

        executorService.shutdown();
        try {
            // 主线程阻塞
            countDownLatch.await();
            // 可以设置最大阻塞时间,防止线程一直挂起
            /*boolean await = countDownLatch.await(1, TimeUnit.SECONDS);
            if (!await) {
                // 超过时间子线程都还没有结束,直接都回滚
                errorTag.set(true);
            }*/
            log.info("继续执行主线程");

            // 继续执行后续的操作,比如insert、update等
            TestEntity entity = new TestEntity();
            entity.setId(new Random().nextInt(999999999));
            entity.setCount(1);
            entity.setCommodityCode("handleTestMain");
            entity.setMoney(new Random().nextInt(1000000));
            entity.setUserId("user-handleTestMain");
            testMapper.insert(entity);

        } catch (Exception e) {
            log.error("主线程业务执行异常");
            errorTag.set(true);
        } finally {
            // 主线程业务执行完成后,执行errorCountDown计时器减一,使得所有阻塞的子线程,继续执行进入到异常判断中
            errorCountDown.countDown();
        }

        long end = System.currentTimeMillis();
        log.info("数据处理完成,耗时:{}", (end - start) / 1000);
        // 如果出现异常
        if (errorTag.get()) {
            throw new RuntimeException("异步业务执行出现异常");
        }
        log.info("主线程执行完成");
    }

    /**
     * 子线程异步处理,并且实现回滚
     * 由于开启了事务回滚,异步的线程数量要小于,dataSource中配置的maximum-pool-size数量
     */
    @Transactional(rollbackFor = Exception.class)
    public void handleDataAsyncSonBack(List<TestEntity> list, CountDownLatch countDownLatch, CountDownLatch errorCountDown, AtomicBoolean errorTag) {
        try {
            log.info("开始执行子线程");
            for (TestEntity entity : list) {
                if (errorTag.get()) {
                    break;
                }

                // 对实体类的业务处理,此处模拟业务处理,耗时50ms
                ThreadUtil.sleep(50);
                // 数据库查询操作
                testMapper.insert(entity);

                // 模拟数据处理中,出现了异常
                if (entity.getCount().equals(2000)) {
                    throw new RuntimeException("子线程执行异常");
                }
            }
        } catch (Exception e) {
            log.error("子线程异常:{}", e.getMessage(), e);
            errorTag.set(true);
        } finally {
            // 子线程中,业务处理完成后,利用countDown的特性,计数器减一操作
            countDownLatch.countDown();
        }

        log.info("handleDataAsyncSonBack-业务处理完成从,等待其他子线程");
        // 子阻塞,直到其他子线程完成操作
        try {
            errorCountDown.await();
        } catch (Exception e) {
            errorTag.set(true);
        }
        log.info("handleDataAsyncSonBack-子线程执行完成");
        if (errorTag.get()) {
            // 抛出异常,回滚数据
            throw new RuntimeException("handleDataAsyncSonBack-子线程业务执行异常");
        }
    }

    /**
     * 模拟解析的excel等文件的数据
     */
    private List<TestEntity> getData() {
        List<TestEntity> list = new ArrayList<>();
        // 此处模拟一万条数据
        for (int i = 1; i <= 10000; i++) {
            TestEntity entity = new TestEntity();
            entity.setId(new Random().nextInt(999999999));
            entity.setCount(i);
            entity.setCommodityCode("code-" + i);
            entity.setMoney(new Random().nextInt(1000000));
            entity.setUserId("user-" + i);

            list.add(entity);
        }
        return list;
    }
}

CountDownLatch实战应用——批量数据多线程协调异步处理(子线程执行事务回滚)-LMLPHP

CountDownLatch实战应用——批量数据多线程协调异步处理(子线程执行事务回滚)-LMLPHP

11-26 19:27