CountDownLatch源码解析加流程图详解--AQS类注释翻译

阅读量：4292 次

发布时间：2019-05-27

本文共 8922 字，大约阅读时间需要 29 分钟。

功能定义

这个类是一个叫名Doug Lea的java 并发大神开发的。是在jdk1.5中添加的。其官方解释为：

A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.

他的意思是CountDownLatch是一个同步工具，他的功能是允许一个或多个线程处于等待状态，直到其他的线程执行完成一组操作后。

CountDownLatch的初始化需要一个计数器, 调用await()方法会被锁住直到当前的计数器由于调用countDown(）方法到达0的时候，所有的等待线程会被立即释放并且随后在使用await()方法也会马上结束。不会在有等待效果。这个方法只能使用一次，无法重置计数。如果你想使用重复计数，可以使用CyclicBarrier。

CountDownLatch 是一个多用途同步工具可以用在很多场景中。使用计数器一来初始化CountDownlatch来提供一个插销。可以把这个过程比如一个门：所有的线程调用await方法就会在门那里待等，直接到有一个线程调用countDown()方法，拔到了插销，门就打开了。 CountDownLatch 初始计数器设置为N的话，可以让一个线程一直等待，直到有N个线程完成一些工作并调用countDown()方法，或者一些线程完成一些工作后调用N次countDown()方法。

这里有一个问题：这句话是CountDownLatch 源码中的英文注释，我不太会翻译。如果有能翻译的，请在下方留言。万分感谢。

A useful property of a {@code CountDownLatch} is that it
doesn’t require that threads calling {@code countDown} wait for
the count to reach zero before proceeding, it simply prevents any
thread from proceeding past an {@link #await await} until all
threads could pass.

下面一个案例：

/** * description: * author: 田培融 * date: 2020-08-07 10:18 */public class Driver {
       public static void main(String[] args) throws InterruptedException {
           int n = 5;        CountDownLatch startSignal = new CountDownLatch(1);        CountDownLatch doneSignal = new CountDownLatch(n);        for (int i = 0; i < n; i++) {
               new Thread(new Worker(startSignal,doneSignal)).start();        }        System.out.println("开始工作");        startSignal.countDown();        doneSignal.await();        System.out.println("工作结束 ");    }    static class Worker implements  Runnable{
           private final  CountDownLatch startSignal;        private final  CountDownLatch doneSignal;        public Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
               this.startSignal = startSignal;            this.doneSignal = doneSignal;        }        @Override        public void run() {
               try {
                   startSignal.await();                doWork();                doneSignal.countDown();            } catch (InterruptedException e) {
                   e.printStackTrace();            }        }        void doWork(){
               System.out.println("工作...");        }    }}

解析全生命周期

创建

创建只需要一个 new 出一个新对象来就可以了。

CountDownLatch startSignal = new CountDownLatch(1);

源码

public CountDownLatch(int count) {
           if (count < 0) throw new IllegalArgumentException("count < 0");        this.sync = new Sync(count);    }

count如果小于零就会抛出异常。通过count构建出Sync对象。

Sync做为CountDownLatch的内部类，继承AbstractQueueSynchronizer。这个类为CountDownLatch提供同步控制，使用AQS 状态来维护计数器。

创建Sync的时候需要传一个值，这个值最终被设置到AQS中的state中。

在这里插入图片描述

这段代码就是AbstractQueuedSynchronizer.java中的赋值操作. state 被volatitle修饰来保证内存的可见性。先不介绍AQS中state的具体意思，下面的内容会有介绍。

/**     * The synchronization state.     */    private volatile int state;    /**     * Sets the value of synchronization state.     * This operation has memory semantics of a {@code volatile} write.     * @param newState the new state value     */    protected final void setState(int newState) {
           state = newState;    }

这样初始化的操作就完成了。

await() 等待方法

先来看看await()在CountDownLatch中的代码

/**     * Causes the current thread to wait until the latch has counted down to     * zero, unless the thread is {@linkplain Thread#interrupt interrupted}.     *     * If the current count is zero then this method returns immediately.     *     * 
If the current count is greater than zero then the current     * thread becomes disabled for thread scheduling purposes and lies     * dormant until one of two things happen:     * 
   
         * 
    The count reaches zero due to invocations of the     * {@link #countDown} method; or     * 
    Some other thread {@linkplain Thread#interrupt interrupts}     * the current thread.     * 
   
     *     * If the current thread:     * 
   
         * 
    has its interrupted status set on entry to this method; or     * 
    is {@linkplain Thread#interrupt interrupted} while waiting,     * 
   
     * then {@link InterruptedException} is thrown and the current thread's     * interrupted status is cleared.     *     * @throws InterruptedException if the current thread is interrupted     *         while waiting     */    public void await() throws InterruptedException {
           sync.acquireSharedInterruptibly(1);    }

我们来翻译一些这个英文注释。

让当前线程一直等待，直到 count计数器减到0。除非线程被打断，就会抛出异常。

如果当前计数器是0的话，这个await()方法就会直接放行。

如果当前计数器比0大的话那么当前线程就会停止重新调度并处于休眠状态，直到触发下面的条件：

调用countDown()方法，使用当前计数器减到0。

其他的线程调用 intertrupt()方法打断这个线程。

哪果当前线程在进入await()方法之前已经处于中断状态，或者等待时被中断那么就会抛出一个InterruptedException异常并且这个线程的打断状态也会被清除。

以上就是官方注释对这个方法的解释，翻译的不是很准确包括了我自己的一点理解。

await()方法中调用了 AbstractQueueSynchronizer.java 中的 sync.acquireSharedInterruptibly(1); 从方法名上简单理解是 获取同步共享等待。

下面我们就到AbstractQueueSynchronizer.java中去看一看这个方法。

/**     * Acquires in shared mode, aborting if interrupted.  Implemented     * by first checking interrupt status, then invoking at least once     * {@link #tryAcquireShared}, returning on success.  Otherwise the     * thread is queued, possibly repeatedly blocking and unblocking,     * invoking {@link #tryAcquireShared} until success or the thread     * is interrupted.     * @param arg the acquire argument.     * This value is conveyed to {@link #tryAcquireShared} but is     * otherwise uninterpreted and can represent anything     * you like.     * @throws InterruptedException if the current thread is interrupted     */    public final void acquireSharedInterruptibly(int arg)            throws InterruptedException {
           if (Thread.interrupted())            throw new InterruptedException();        if (tryAcquireShared(arg) < 0)            doAcquireSharedInterruptibly(arg);    }

以共享模式获取，如果调用interrupted()则中止。首先需要通过检查当前线程的interrupted状态，然后调用tryAcquiredShared(arg)方法。返回值，检查大小。否则线程就会等待，可能会反复的阻塞和取消阻塞直接调用 tryAcquireShared成功或者被打断。

在这里他会调用两个方法，这里我们先看第一个。tryAcquireShared() 这个方法在 CountDownLatch.java中的内部类被重写了。

这里会获取CountDownLatch刚创建的时候传递给AQS的计数器，这里在重复一遍。 CountDownLatch内部内Sync继承了AQS，在创建CountDownLatch的时候会将计数器传递给AQS。AQS就是使用的这个state来接收的。在JUC的其他并发工具中也有很多继承使用了AQS，都使用了state，但是代表的意思在不同的工具中是不同的。

这里表示的是如果计数器等于0的话就会返回1如果不等0的话就会返回-1。 我们初始化的时候给计数器设置的值是3，此时获取到的state的值就是3，返回的就是-1。条件就会成立，并执行doAcquireSharedInterruptibly(arg)方法。

if (tryAcquireShared(arg) < 0)            doAcquireSharedInterruptibly(arg);

下面我们就看看这个private void doAcquireSharedInterruptibly(int arg) 方法做了什么。

/**     * Acquires in shared interruptible mode.     * @param arg the acquire argument     */    private void doAcquireSharedInterruptibly(int arg)        throws InterruptedException {
           final Node node = addWaiter(Node.SHARED);        boolean failed = true;        try {
               for (;;) {
                   final Node p = node.predecessor();                if (p == head) {
                       int r = tryAcquireShared(arg);                    if (r >= 0) {
                           setHeadAndPropagate(node, r);                        p.next = null; // help GC                        failed = false;                        return;                    }                }                if (shouldParkAfterFailedAcquire(p, node) &&                    parkAndCheckInterrupt())                    throw new InterruptedException();            }        } finally {
               if (failed)                cancelAcquire(node);        }    }

我们先看开头的方法注释，获取共享可中断模型。这里的参数arg的值是1，是在CountDownLatch方法中传入的。他这里会进入一个自循环阻塞当前线程，直到state 减到0为止，或者线程被中断。先整体的理解一下这断代码的意思。下面我们就开始开始梳理一下这个方法中用到的AQS中的元素。

梳理一下AQS来解释doAcquireSharedInterruptibly方法

首先我们到AQS的类最开始的位置看到有很多英文，我们就先来翻译一下这些英文。看看 Doug Lea大神给这个类的定义。

提供一个框架用于开发依赖于先进先出等待队列的阻塞锁和相关容器（信号量，事件等等)。这个类被设计为一些依赖于单个原子整型表示状态的同步容器的基础。（也就是说我们在设计开发同步容器的时候，如果使用了单个原子整型表示状态了，就可以使用AQS。）。子类必须定义一个protected方法来修改状态，并定义这个状态在这个状态中减或者加是什么意思。（此处翻译的感觉不太对劲。。。）。

根据这些，AQS类中的其他方法执行其他的队列和阻塞机制。子类可以维护其他的状态字段，关于同步操作状态值必须使用getState setState``compareAndSetState这三个方法。

子类应该定义一个非public的内部类，用于初始化封闭类的同步属性。AbstractQueuedSynchronizer类不实现任何同步接口。（在CountDownLatch中就是Sync类哪下图

在这里插入图片描述

）

相反，它定义了方法比如acquireInterruptibly()来调用适合的具体哪种锁和相关同步容器来实现公共方法。

AQS类介绍

这个类支持一个或者两个独占模型和共享模型。当获取独占模型，其他线程不能在试图获取。共享模型可以多个线程获取。当一个共享模型获取他们的区别是机械意义上的。下一个等待线程也必须确定是否可以获取成功。在不同的的模型下线程共享相同的FIFO队列。通常，实现的子类只支持一种模型，但是在ReadWriteLock中两种模型都是可以的。只支持独占模型或者只支持共享模型的不需要定义未使用的模型方法。

ConditionObject类

AQS内部定义了一个内容部ConditionObject，它实现了Condition类，支持isHeldExclusively 方法，报告这个线程是否是独占的。

release 方法：调用getState方法和acquire方法修改保存当前状态值。最终恢复这个对象以前获取到的状态值。没有其他方法来创造这样的条件，如果这个条件不被满足请不要使用它。 ConditionObject的行为取决行实现容器的语义。

在次回到AQS中

这个类内部队列提供检查，检测，监控的方法，以及类似于条件对象的方法。可以根据需要将他们导入到类中，以实现他们的同步机制。

这个类的序列化对象中存储底层的原子整型维护状态，因此反序列化对象有一个空的线程队列。需要序列化的典型子类将定义一个readObject方法，在反序列化的时恢复到已知的状态。

案例介绍

使用这个类作为同步容器的基础类，需要重新定义以下方法，如检查或者修改同步状态使用getState方法，setState方法，或者compareAndSetState。下面的这些方法在juc的工具有很多都有实现如ReentrantLock.java中，可以在里面看到实现方法。会发现有使用到getState方法。