😊 @ 作者: 一恍过去
💖 @ 主页: https://blog.csdn.net/zhuocailing3390
🎊 @ 社区: Java技术栈交流
🎉 @ 主题: CountDownLatch实战应用——实现异步多线程业务处理,异常情况回滚全部子线程
⏱️ @ 创作时间: 2023年12月1日7
目录
- 1、概述
- 2、实现
- 3、方法说明:
- 4、代码实例
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、代码实例
Controller:
@RestController
@RequestMapping("/test")
@Slf4j
public class TestController {@Resourceprivate CountDownService countDownService;/*** CountDownLatch实现异步多线程不同业务处理,不同service,异常情况回滚全部子线程** @return*/ @GetMapping("/countDown/handleDataBack")public String countDownHandleDataBack() {countDownService.handleDataBack();return "success";}
Sevice:
@Service
@Slf4j
public class CountDownService {@Resourceprivate TestMapper testMapper;@Resourceprivate ApplicationContext applicationContext;/*** 主线程,同时调用多个个子线程进行业务处理,当其中一个子线程出现异常,则全部子线程进行回滚*/@Transactional(rollbackFor = Exception.class)public void handleDataBack() {AtomicBoolean errorTag = new AtomicBoolean(false);// 设置countDown大小,与异步执行的业务数量一样,比如2个CountDownLatch countDownLatch = new CountDownLatch(2);// 再创建一个CountDownLatch,大小固定为一,用于子线程相互等待,最后确定是否回滚CountDownLatch errorCountDown = new CountDownLatch(1);// 异步调用其他Service,执行业务处理CountDownService bean = applicationContext.getBean(CountDownService.class);bean.handleTestOne(countDownLatch, errorCountDown, errorTag);bean.handleTestTwo(countDownLatch, errorCountDown, errorTag);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();}// 如果出现异常if (errorTag.get()) {throw new RuntimeException("异步业务执行出现异常");}log.info("主线程执行完成");}/*** 子线程具体业务处理*/@Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRES_NEW)@Asyncpublic void handleTestOne(CountDownLatch countDownLatch, CountDownLatch errorCountDown, AtomicBoolean errorTag) {log.info("开始执行handleTestOne线程");// 模拟业务耗时ThreadUtil.sleep(2000);try {// 执行数据库操作TestEntity entity = new TestEntity();entity.setId(new Random().nextInt(999999999));entity.setCount(1);entity.setCommodityCode("handleTestOne");entity.setMoney(new Random().nextInt(1000000));entity.setUserId("user-handleTestOne");testMapper.insert(entity);// 模拟出现异常int a = 1 / 0;} catch (Exception e) {errorTag.set(true);}// 子线程中,业务处理完成后,利用countDown的特性,计数器减一操作countDownLatch.countDown();// 子阻塞,直到其他子线程完成操作try {errorCountDown.await();} catch (Exception e) {errorTag.set(true);}log.info("handleTestOne-子线程执行完成");if (errorTag.get()) {// 抛出异常,回滚数据throw new RuntimeException("handleTestOne-子线程业务执行异常");}}/*** 子线程具体业务处理*/@Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRES_NEW)@Asyncpublic void handleTestTwo(CountDownLatch countDownLatch, CountDownLatch errorCountDown, AtomicBoolean errorTag) {log.info("开始执行handleTestTwo线程");// 模拟业务耗时ThreadUtil.sleep(500);try {// 执行数据库操作TestEntity entity = new TestEntity();entity.setId(new Random().nextInt(999999999));entity.setCount(1);entity.setCommodityCode("handleTestTwo");entity.setMoney(new Random().nextInt(1000000));entity.setUserId("user-handleTestTwo");testMapper.insert(entity);} catch (Exception e) {errorTag.set(true);}// 子线程中,业务处理完成后,利用countDown的特性,计数器减一操作countDownLatch.countDown();// 子阻塞,直到其他子线程完成操作try {errorCountDown.await();} catch (Exception e) {errorTag.set(true);}log.info("handleTestTwo-子线程执行完成");if (errorTag.get()) {// 抛出异常,回滚数据throw new RuntimeException("handleTestTwo-子线程业务执行异常");}}
}