高并发之CountDownLatch

2019-05-10

字数统计: 849字 | 阅读时长: 3分

概念

CountDownLatch是一个通用的同步工具，CountDownLatch允许一个或多个线程在其他线程执行操作完成前等待。

用法

CountDownLatch需要初始化一个计数器，调用await()方法使线程阻塞，调用countDown()方法让计数器减一，当计数器减为零时，所有等待的线程释放，并且后续await()的线程立即执行。

场景

使用CountDownLatch主要有两个场景

主线程阻塞直到子线程全部执行完毕。比如，我和女朋友一起去吃麻辣烫，厨师需要等我和女朋友全部点完配菜之后才可以开始制作，而我和女朋友选菜的过程是可以同时进行的。

public class CDLForSpicyHotPot {

    //设置计数器
    private static CountDownLatch latch = new CountDownLatch(2);

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(2);

        //我和女朋友开始选配菜
        ArrayList<String> users = Lists.newArrayList("Me", "GF");
        for (String user : users) {
            executor.submit(() -> {
                chooseSomething(user);
                //选好了
                latch.countDown();
            });
        }

        //厨师等待中
        latch.await();
        System.out.println(System.currentTimeMillis() + " : The Cooker begin to cook");

        if(!executor.awaitTermination(1, TimeUnit.SECONDS)){
            executor.shutdownNow();
        }
    }

    private static void chooseSomething(String user) {
        Random random = new Random();
        try {
            Thread.sleep(random.nextInt(5) * 1000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(System.currentTimeMillis() + " : " + user + " choose food meat vegetables...");
    }
}

子线程先阻塞直到主线程执行了具体步骤后再一起执行。比如我们常说的发令枪场景

public class CDLForStartingGun {

    //设置计数器
    private static CountDownLatch latch = new CountDownLatch(1);

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(5);

        for (int i = 0; i < 5; i++) {
            int finalI = i;
            executor.submit(() -> {
                try {
                    //运动员准备
                    latch.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(System.currentTimeMillis() + " : player-" + finalI + " start running");
            });
        }

        //裁判开枪
        System.out.println(System.currentTimeMillis() + " : The referee fired");
        latch.countDown();

        if(!executor.awaitTermination(1, TimeUnit.SECONDS)){
            executor.shutdownNow();
        }
    }
}

注意：此处需要注意的是线程池核心线程数和运动员数量的设置，如果CPU无法处理太多的线程，运动员就可能不会同时runing

原理

使用构造函数传入计数，CountDownLatch使用AQS的state代表计数count

//CountDownLatch的构造函数 
public CountDownLatch(int count) {
        if (count < 0) throw new IllegalArgumentException("count < 0");
        this.sync = new Sync(count);
    }

//CountDownLatch的内部类，继承了AQS
private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;

        Sync(int count) {
          	//设置AQS的state为count
            setState(count);
        }

        int getCount() {
            return getState();
        }

  			//获取计数，不为0则返回-1
        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }

        protected boolean tryReleaseShared(int releases) {
            //递减，当计数为0时唤醒阻塞线程
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }
    }

使用await()使线程阻塞，await()会调用AQS的acquireSharedInterruptibly(1)方法

public final void acquireSharedInterruptibly(int arg)
           throws InterruptedException {
       if (Thread.interrupted())
           throw new InterruptedException();
  			//Sync的实现，计数不为0则是true
       if (tryAcquireShared(arg) < 0)
         	//AQS的方法，使用park阻塞线程
           doAcquireSharedInterruptibly(arg);
   }

使用countDown()使计数减一，直到减为零

public final boolean releaseShared(int arg) {
  			//Sync的实现，递减计数值，计数为0则返回true
        if (tryReleaseShared(arg)) {
          	//AQS的实现，使用unpark唤醒阻塞线程
            doReleaseShared();
            return true;
        }
        return false;
    }

CountDownLatch包括很多其他的锁的都是基于AQS(AbstractQueuedSynchronizer)来实现的，后续我会持续更新相关的JAVA知识，希望能够帮助到大家。

欢迎关注我的公众号“shy先生”，很高兴能与大家一起成长。