带你看看Java的锁(三)-CountDownLatch和CyclicBa…

2020-04-11 16:01:29来源：博客园阅读 ()

带你看看Java的锁(三)-CountDownLatch和CyclicBarrier

带你看看Java中的锁CountDownLatch和CyclicBarrier

前言

基本介绍

使用和区别

核心源码分析

总结

前言

Java JUC包中的文章已经写了好几篇了，首先我花了5篇文章从源码分析的角度和大家讲了AQS，为什么花5篇时间讲这个，是因为AQS真的很重要，JUC中提供的很多实现都是基于AQS的，所以说看懂AQS中的代码很重要，看不懂的多看几遍，上网多找找文章，这个时间学习的途径和资料真的很多，只要你愿意花时间，相信自己一定会比别人懂的更多！

好的，今天的废话有点儿多，进入正题吧，今天本来也继续写的是java中的锁第三篇CountDownLatch的，但是发现CyclicBarrier在功能上与其有相似之处，但是其实2者是不同的实现，而且网上很多人都疑惑2者的使用场景和区别，今天就顺便一起写了吧

基本介绍

CountDownLatch

CountDownLatch 中文翻译过来就是闭锁，倒计时锁的意思，CountDownLatch其实和Semaphore其所都是AQS中线程共享模式的实现，就是同时允许多个线程拥有一个资源，Semaphore是为了控制线程的并发数量，说白了就是控制占用资源的线程数量，CountDownLatch是创建闭锁的线程等待多个占用资源的线程执行完成后才能去执行自己的方法，可能解释的有点隐晦，下面我会通过demo去描述下

CyclicBarrier

CyclicBarrier 中文编译过来是屏障锁，而且可以循环利用，一会儿源码中我会做说明，它的意思是让多个线程在一个条件下做等待相当于屏障一样，只有所有的线程都到达了，才能继续做自己别的事情，暂时看不明白的，继续往下看，后面相信你能明白的！

使用和区别

CountDownLatch的使用

说了那么多，我们先通过一个Demo,看下具体的使用：

/**
 * @ClassName CountDownLatchDemo
 * @Auther burgxun
 * @Description: 倒计时锁CountDownLatch Demo
 * @Date 2020/4/11 12:51
 **/
public class CountDownLatchDemo {
    public static void main(String[] args) {
        long timeNow = System.currentTimeMillis();
        CountDownLatch countDownLatch = new CountDownLatch(3);

        ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(3, 3, 10, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(5));
        poolExecutor.prestartCoreThread();
        try {
            PrintLog("执行秒杀系统的健康检查");
            poolExecutor.execute(new CheckMQ(countDownLatch));
            poolExecutor.execute(new CheckRPCInterface(countDownLatch));
            poolExecutor.execute(new PreRedisData(countDownLatch));
            countDownLatch.await();
            PrintLog("健康检查执行完毕，共计花费：" + (System.currentTimeMillis() - timeNow));

        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            poolExecutor.shutdown();
        }
    }

    public static void PrintLog(String logContent) {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd hh:mm.ss.SSS");
        System.out.println(String.format("%s : %s", simpleDateFormat.format(new Date()), logContent));
    }

    static class CheckRPCInterface implements Runnable {

        private CountDownLatch countDownLatch;

        public CheckRPCInterface(CountDownLatch countDownLatch) {
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            try {
                PrintLog("RPC接口检测开始执行");
                Thread.sleep(1000);
                PrintLog("RPC接口检测完成");

            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                countDownLatch.countDown();
            }
        }
    }

    static class PreRedisData implements Runnable {

        private CountDownLatch countDownLatch;

        public PreRedisData(CountDownLatch countDownLatch) {
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            try {
                PrintLog("Redis数据开始预热开始执行");
                Thread.sleep(3000);
                PrintLog("Redis数据开始预热完成");

            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                countDownLatch.countDown();
            }
        }
    }

    static class CheckMQ implements Runnable {

        private CountDownLatch countDownLatch;

        public CheckMQ(CountDownLatch countDownLatch) {
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            try {
                PrintLog("MQ检测开始执行");
                Thread.sleep(2000);
                PrintLog("MQ检测完成");

            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                countDownLatch.countDown();
            }
        }
    }

}

执行结果:

org.example.CountDownLatchDemo
2020-04-11 01:23.33.859 : 执行秒杀系统的健康检查
2020-04-11 01:23.33.868 : MQ检测开始执行
2020-04-11 01:23.33.870 : RPC接口检测开始执行
2020-04-11 01:23.33.870 : Redis数据开始预热开始执行
2020-04-11 01:23.34.870 : RPC接口检测完成
2020-04-11 01:23.35.870 : MQ检测完成
2020-04-11 01:23.36.871 : Redis数据开始预热完成
2020-04-11 01:23.36.871 : 健康检查执行完毕，共计花费：3065

上面的场景是什么呢，我来描述下，秒杀系统我相信很多小伙伴多多少少都了解些，比如我们一个秒杀系统要12点上线，上线之前一定会做一下工作的准备，比如一些核心接口的检查，Redis核心数据的预热，MQ消息队列的检测等等一些，如果这些工作都执行完成后，我们可能才会开启我们的秒杀入口。上面的Demo 就是描述这样一件事，当然生产环境中还要做一些监控等等，从上面的Demo 中我们可以看到，当前的创建CountDownLatch的线程，在执行到 countDownLatch.await的时候，线程是要阻塞的，只有当其余的检测线程执行完成以后，当前线程才会被唤醒执行后面的逻辑，

这边大家注意下，这边的阻塞线程是创建countDownLatch的线程，其余的线程都是执行完成后才会进行countDownLatch.countDown操作！理解这个对理解和CyclicBarrier区别很重要

CyclicBarrier的使用

下面我在看下CyclicBarrier的Demo

/**
 * @ClassName CyclicBarrierDemo
 * @Auther burgxun
 * @Description: 屏障锁  全家出去玩的Demo
 * @Date 2020/4/11 14:08
 **/
public class CyclicBarrierDemo {
    private static int NUMBER = 3;

    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(NUMBER, () -> {
            PrintLog("好的，都收拾好了 全家可以出门旅游了");
        });
        PrintLog("全家准备出门春游");
        createThread(cyclicBarrier, "爸爸", 2000).start();
        createThread(cyclicBarrier, "妈妈", 5000).start();
        createThread(cyclicBarrier, "儿子", 3000).start();
        System.out.println("gogogogo!");
    }

    public static Thread createThread(CyclicBarrier cyclicBarrier, String name, int runTime) {
        Thread thread = new Thread(() -> {
            try {
                PrintLog(String.format("%s 开始收拾准备出门", name));
                Thread.sleep(runTime);
                PrintLog(String.format("%s 收拾花了%s 毫秒", name, runTime));
                if (cyclicBarrier.getNumberWaiting() < (NUMBER - 1)) {
                    PrintLog(String.format("%s 开始等待。。。。", name));
                }
                long time = System.currentTimeMillis();
                cyclicBarrier.await();
                PrintLog(String.format("%s 开始穿鞋出发，我等待了%s 秒了", name,
                        (System.currentTimeMillis() - time)));

            } catch (InterruptedException | BrokenBarrierException e) {
                e.printStackTrace();
            }
        });
        return thread;
    }


    public static void PrintLog(String logContent) {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd hh:mm.ss.SSS");
        System.out.println(String.format("%s : %s", simpleDateFormat.format(new Date()), logContent));
    }
}

执行结果：

2020-04-11 02:33.35.306 : 全家准备出门春游
gogogogo!
2020-04-11 02:33.35.313 : 妈妈 开始收拾准备出门
2020-04-11 02:33.35.313 : 儿子 开始收拾准备出门
2020-04-11 02:33.35.313 : 爸爸 开始收拾准备出门
2020-04-11 02:33.37.314 : 爸爸 收拾花了2000 毫秒
2020-04-11 02:33.37.314 : 爸爸 开始等待。。。。
2020-04-11 02:33.38.314 : 儿子 收拾花了3000 毫秒
2020-04-11 02:33.38.314 : 儿子 开始等待。。。。
2020-04-11 02:33.40.313 : 妈妈 收拾花了5000 毫秒
2020-04-11 02:33.40.313 : 好的，都收拾好了 全家可以出门旅游了
2020-04-11 02:33.40.314 : 爸爸 开始穿鞋出发，我等待了2999 秒了
2020-04-11 02:33.40.314 : 妈妈 开始穿鞋出发，我等待了0 秒了
2020-04-11 02:33.40.314 : 儿子 开始穿鞋出发，我等待了2000 秒了

上面的Demo 是什么呢，是一个全家要出去完的demo,既然要出去玩，那一定要准备准备，妈妈要化妆换衣服神马的，儿子可能要选一个自己心爱的玩具带着，爸爸估计没什么事，换个衣服就好了，剩下的就要等儿子和妈妈，只有全家都好了，才能每个人穿鞋，然后步行到停车场去开车走人！
这边要注意的是，创建CyclicBarrier的线程是非阻塞的，从上面执行gogogo的日志就可以看出，阻塞的线程是执行cyclicBarrier.await的线程，就是任务线程吧，这个是要做等待的，只要等到所有的线程到执行到了一个点，才能继续执行await后面的代码

如果上面的Demo 你还是不明白，我可以这么说，拼多多的拼团大家一定知道的，比如一个商品必须拼满3个人才能成团成功享受优惠，第一个人购买时下单等待但是不能成团支付，第二个也是，直到第3个人来了才能成团成功，这个时候会通知第一个人，第二个人订单成团成功了可以享受优惠价格，现在可以支付了！当然第4个人来的时候还是要做等待。。这就是CyclicBarrier最好的理解！！！

2者的区别

从上文我加粗的文字中可以看出 2者的区别是什么，核心就在于阻塞的线程是不一样的，

CountDownLatch是创建countDownLatch对象的线程调用await并且会阻塞当前线程，其余执行任务的线程并不会阻塞，只是执行完成后修改countDownLatch里面的计数器值，当最后一个线程执行完成后，计数器值变0，这个时候才会唤醒被阻塞的创建线程

CyclicBarrier是创建CyclicBarrier对象的线程并不会阻塞，也不会调用await,只有执行任务的线程才会调用CyclicBarrier的await方法并且调用后阻塞当前线程，当CyclicBarrier里面的count属性值为0的时候，那么说明说有的线程都执行就绪了可以穿过这个屏障，唤醒之前调用await的线程，继续执行剩下的逻辑。

那我在列举一个生活中的例子，相信大家一定参加过公司的团建跑步什么的，一般都会把人分成多个组，然后每个组有个队长，就拿我公司的团建举例吧，每年都会去拉练说白了就是跑步，一跑就是十几公里，公司为了活动的趣味性，一般都是在中途设置多个打卡点做任务什么的，一般到了打卡点队长会等所有队员到齐了拍个照片做个任务什么的才会继续往下个关口走，队长点名就相当于一个CountDownLatch 为什么这么说呢，因为队长做点名等一系列的任务，这个时候他到了打卡点就要等待所有的队员到来，这个时候剩余的队员就相当于执行了一个cyclicBarrier任务，因为队员到了打卡点后，自己不能继续往下走，必须等待队伍里面其余的人到了才能继续下个任务！

说到这里大家明白了么，简单的说就是CountDownLatch是队长自己阻塞要等待大家一起走，cyclicBarrier是队员到了打卡点要自己等待别的队员一起到了才能继续下个任务！

再直白一点儿就是队长就是司机要等人齐了才能发车队员相当于乘客，自己到了在巴士里面还要等待所有人全部到了才能走

核心代码分析

CountDownLatch

先看下CountDownLatch的类结构，CountDownLatch结构很简单，有个内部类，Sync 继续了AQS类，剩下就是一个初始化方法，和2个await方法

Sync

private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;

        /**
         * 设置AQS同步器里面State的值
         */
        Sync(int count) {
            setState(count);
        }

        /**
         * 获取AQS里面State的值
         */
        int getCount() {
            return getState();
        }

        /**
         * 重写了AQS的方法，尝试获取资源  如果当前的状态值等于0了 那就返回1 否则返回-1
         * 为什么这边要这么实现呢  因为这个锁是一个倒计时锁  如果当前State不等于0 返回值是-1 说明当前线程还是需要阻塞的
         * 只有当State 等于0了 说明占用资源的线程都执行结束了 执行结束会调用countDown方法
         * 那这个时候线程就不需要阻塞了 可执行下去 而且会是
         */
        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }

        /**
         * 共享模式下的释放资源
         */
        protected boolean tryReleaseShared(int releases) {
            // Decrement count; signal when transition to zero
            for (; ; ) {//自旋
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c - 1;//每次释放的时候 同步器状态值State -1
                if (compareAndSetState(c, nextc))//为什么实用CAS操作呢 这边和Semaphore信号量  是因为存在多线程操作的问题
                    return nextc == 0;
            }
        }
    }

上面的代码我都已经做了注解，相信看下能够明白，方法很也很简单，主要就是2个重写了AQS的tryAcquireShared和tryReleaseShared的方法

内部方法

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

    /**
     * 共享模式下的AQS获取的资源的方法 响应中断的版本
     */
    public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

    /**
     * 共享模式下 获取资源 带截止时间的
     */
    public boolean await(long timeout, TimeUnit unit)
            throws InterruptedException {
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }

    /**
     * 释放同步器State资源 减1
     */
    public void countDown() {
        sync.releaseShared(1);
    }

    /**
     * 返回当前的同步器状态值
     */
    public long getCount() {
        return sync.getCount();
    }

上面就是CountDownLatch的方法也没什么好讲的都是调用了AQS里面的方法实现的，这个已经多少阐述，不说了不说了，前文都有部明白的自己看看~

CyclicBarrier

CyclicBarrier的类结构呢和CountDownLatch 是完全不一样的，它并没有直接继承AQS类，而是使用了ReentrantLock和Condition组合使用去实现功能的。

构造函数

首先看下2个构造函数：

public CyclicBarrier(int parties, Runnable barrierAction) {
        if (parties <= 0) throw new IllegalArgumentException();
        this.parties = parties;//表示开启Barrier需要的线程数量
        this.count = parties;// 当前还剩余需要开启屏障的线程数量
        this.barrierCommand = barrierAction;//开启屏障后的回调函数
    }

    /**
     * 创建CyclicBarrier   parties表示开启Barrier需要的线程数量
     */
    public CyclicBarrier(int parties) {
        this(parties, null);
    }

看完了构造函数，看下几个方法

reset方法

 /**
     * break当前的屏障 并且开启下一个屏障
     */
    public void reset() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            breakBarrier();   // 破坏当前的屏障
            nextGeneration(); // 开启一个新的屏障
        } finally {
            lock.unlock();
        }
    }
    
     private void breakBarrier() {
        generation.broken = true;//设置当前屏障的broken值，
        count = parties;//重置count值 屏障已经破坏 count值要还原 为下次屏障做准备
        trip.signalAll();//唤醒所有之前在此屏障上等待的线程
    }
    
     private void nextGeneration() {
        trip.signalAll();//唤醒之前所有在屏障上等待的线程
        count = parties;//设置开启屏障的数量
        generation = new Generation();//重置generation 里面的broken默认值false
    }

breakBarrier和nextGeneration方法做的事情差不多，breakBarrier做的主要是对当前屏障的处理，nextGeneration做的为了开启一个新的屏障，这就是为为什么叫CyclicBarrier 循环屏障了，因为可以开启了关闭然后再次开启

await方法

CyclicBarrier里面有2个pulic的await方法一个是不带时间等待的，一个是带时间等待的，核心都是调的dowait方法

 public int await() throws InterruptedException, BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen
        }
    }

    public int await(long timeout, TimeUnit unit)
            throws InterruptedException,
            BrokenBarrierException,
            TimeoutException {
        return dowait(true, unit.toNanos(timeout));
    }
   //核心方法dowait
   private int dowait(boolean timed, long nanos)
            throws InterruptedException, BrokenBarrierException,
            TimeoutException {
        final ReentrantLock lock = this.lock;//重入锁 保证执行的线程安全
        lock.lock();
        try {
            final Generation g = generation;

            if (g.broken)//如果当前的Barrier已经broken 就抛出BrokenBarrierException异常
                throw new BrokenBarrierException();

            if (Thread.interrupted()) {//如果当前线程发生了中断 就执行breakBarrier 然后抛出中断异常
                breakBarrier();
                throw new InterruptedException();
            }

            int index = --count;// 每次执行先减少count值 在判断
            if (index == 0) {  // 说明已经到了打开Barrier的条件
                boolean ranAction = false;//是否执行打开当前屏障成功
                try {
                    final Runnable command = barrierCommand;//开启Barrier的执行的Runnable方法
                    if (command != null)
                        command.run();
                    ranAction = true;
                    nextGeneration();//重置下一个Barrier的条件
                    return 0;
                } finally {
                    if (!ranAction)
                        breakBarrier();//如果已经到达了开启Barrier条件 但是没有执行成功 就破坏掉Barrier
                }
            }

            /**
             * loop until
             * tripped, 被最后达到Barrier的线程唤醒
             * broken, 当前的Barrier遭到了破坏
             * interrupted,阻塞等待的线程 被中断 包括当前线程
             * or timed out 或者其中一个等待的线程 超时了
             * */
            for (; ; ) {//这是一个自旋
                try {
                    if (!timed)//是否有等待的时间  如果没有等待的时间 就直接阻塞当前线程 当前线程进入tripConditionQueue中
                        trip.await();
                    else if (nanos > 0L)//如果有等待时间并且等待时间大于0
                        nanos = trip.awaitNanos(nanos);
                } catch (InterruptedException ie) {//捕获线程等待期间的中断
                    if (g == generation && !g.broken) {// 如果generation没有发生变化 说明在还在之前的Barrier中 且Barrier没有破坏
                        breakBarrier();//那就执行破坏Barrier的操作  为了唤醒等待的线程
                        throw ie;
                    } else {
                        // We're about to finish waiting even if we had not
                        // been interrupted, so this interrupt is deemed to
                        // "belong" to subsequent execution.
                        Thread.currentThread().interrupt();//重新设置中断标识
                    }
                }

                if (g.broken)//唤醒后 发现Barrier已经破坏 抛出BrokenBarrierException异常
                    throw new BrokenBarrierException();

                /**
                 *唤醒后发现等待时的Barrier和唤醒后的Barrier已经不一致了 Barrier已经换代了  返回之前的index
                 * 因为一个线程可以开启多个Barrier 比如reset后会唤醒阻塞的线程  这个时候当前的generation对象是new了一个
                 * 就明显不是一个对象了
                 */
                if (g != generation)
                    return index;

                if (timed && nanos <= 0L) {//如果等待的时间已经到了 就破坏屏障 返回TimeoutException
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock();
        }
    }

CyclicBarrier的方法核心都在这边了有兴趣的可以对照下源码自己尝试着理解一下

总结

CyclicBarrier和CountDownLatch 都能做到是的多个线程等待然后再开始做下一步的动作，只是实施的主体不同，通过上面的代码和Demo分析，CountDownLatch实施的主体和进行下一步的操作是创建者本身线程且不可重复利用，CyclicBarrier实施继续操作的主体是其他线程，而且可以重复的使用

原文链接:https://www.cnblogs.com/burg-xun/p/12681580.html
如有疑问请与原作者联系

标签：

版权申明：本站文章部分自网络，如有侵权，请联系：west999com@outlook.com
特别注意：本站所有转载文章言论不代表本站观点，本站所提供的摄影照片，插画，设计作品，如需使用，请与原作者联系，版权归原作者所有