Java并发编程实践：CountDownLatch模拟并发请求及注意事项

﹏ヽ暗。殇╰゛Y 2023-07-10 11:13 10阅读 0赞

**目录**

原子锁使用CountDownLatch

使用CountDownLatch的注意事项

--------------------

## 原子锁使用CountDownLatch  ##

*  **原子锁：**Atomic\*类的封装类型，如：AtomicInteger、AtomicLong。
 *  **CountDownLatch：** 是一个同步工具类，它允许一个或多个线程一直等待，直到其他线程执行完后再执行，CountDownLatch比较通俗的叫法是闭锁(栅栏)，当值为0时就绪否则等待阻塞，最形象的就像赛马，发令枪不响所有马和运动员只能等。

import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.atomic.AtomicInteger;
    
    /**
     * @ClassName: ConcurrencyTest
     * @Description: TODO(功能说明:模拟并发请求)
     * @author: pengjunlin
     * @motto: 学习需要毅力，那就秀毅力
     * @date 2020/3/1 22:45
     */
    public class ConcurrencyTest {
    
        // 请求总数
        public static int clientTotal = 5000;
    
        // 原子锁
        public static AtomicInteger count = new AtomicInteger(0);
    
        public static void main(String[] args) throws Exception {
            ExecutorService executorService = Executors.newCachedThreadPool();
    
            final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
            for (int i = 0; i < clientTotal ; i++) {
                executorService.execute(() -> {
                    countDownLatch.countDown();
                    try {
                        countDownLatch.await();
                        add();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                });
            }
    
            try{
                //直到方法执行完成
                Thread.sleep(10000);
                executorService.shutdown();
                System.out.println(count);
            }catch(InterruptedException e){
                e.printStackTrace();
            }
        }
    
        private static void add() {
            count.incrementAndGet();
        }
    }

## 使用CountDownLatch的注意事项 ##

使用CountDownLatch需要注意避免阻塞，即调用await()一定要确保可以countDown()使最终结果得以执行下去而不是永远阻塞。在实际应用中尤其要注意计数递减一定要保证最终回归到0的状态，确保解除阻塞等待。如示例：

/**
         * 消费数据
         * @param count
         */
        public void consumeOnlineData(int count){
            // 控制每次处理的量
            if (count > queueConsumeNumber) {
                count = queueConsumeNumber;
            }
            // 计数器控制任务是否获取的条件
            countDownLatch = new CountDownLatch(count);
            for (int i = 0; i < count; i++) {
                UserPosition xhyPosition = MobileDataQueue.getOnlineQueue().poll();
                if (ObjectUtils.isNotEmpty(xhyPosition)) {
                    if (ObjectUtils.isNotEmpty(MobileDataQueue.executorService)) {
                        MobileDataQueue.executorService.execute(() -> {
                            try{
                                MobileDataQueue.positionRedisPipelineService.online(xhyPosition);
                            }catch (Exception e){
                                e.printStackTrace();
                            }finally {
                                countDownLatch.countDown();
                            }
                        });
                    } else {
                        MobileDataQueue.threadPoolExecutor.execute(() -> {
                            try{
                                MobileDataQueue.positionRedisPipelineService.online(xhyPosition);
                            }catch (Exception e){
                                e.printStackTrace();
                            }finally {
                                countDownLatch.countDown();
                            }
                        });
                    }
                }else{
                    countDownLatch.countDown();
                }
            }
    
            try {
                // 线程阻塞直到任务执行完成
                log.info("===========[MSG]ONLINE队列中批量消费=====await ....");
                countDownLatch.await();
                log.info("===========[MSG]ONLINE队列中批量消费=====finish");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

可以人为干预await()阻塞的时长，也可以起到解除阻塞的作用：

try {
                // 线程阻塞直到任务执行完成
                log.info("===========[MSG]ONLINE队列中批量消费=====await ....");
                countDownLatch.await(10, TimeUnit.SECONDS);
                log.info("===========[MSG]ONLINE队列中批量消费=====finish");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

注意：try catch finally 语句很重要，finally永远是异常处理完成之后释放资源的最佳方式。