😊 @ 作者： 一恍过去
💖 @ 主页： https://blog.csdn.net/zhuocailing3390
🎊 @ 社区： Java技术栈交流
🎉 @ 主题： CountDownLatch实战应用——批量数据多线程协调异步处理(子线程执行事务回滚)
⏱️ @ 创作时间： 2023年11月26日
目录
- 前言
- 1、概述
- 2、实现
- 3、方法说明：
- 4、代码实例
前言
通过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;
}
}
