JAVA并发编程—CountDownLatch原理剖析

大家好，欢迎来到IT知识分享网。

简介：

CountDownLatch就是我们常说的门闩或者同步计数器，常用于一个或多个线程等待其它线程完成操作的场景。在CountDownLatch出现之前我们一般都使用线程的join()方法来实现类似功能，但是很多时候join不够灵活，很难满足业务场景，所以后面JDK就为我们提供了CountDo wnLatch这个类。

简单使用案例：

假设我们此时正在使用迅雷下载一部大桥老师高达10G的经典影视作品，在不关心网速的情况下为了提升下载速度迅雷后台开启5个线程每个任务分2gb为我们下载。在这5个线程下载任务都结束后将数据进行拼接同时滴滴滴提醒用户下载完成。下面我们就用CountDownLatch简单模拟一下这个场景。

/ * @author haichi * @version 1.0 */ public class Example4 { 
    @Test public void xunlei() throws InterruptedException { 
    CountDownLatch countDownLatch = new CountDownLatch(5); Random random = new Random(System.currentTimeMillis()); IntStream.range(0,5).forEach( ics -> { 
    new Thread( () -> { 
    System.out.println(Thread.currentThread().getName() + "正在下载"); try { 
    Thread.sleep(random.nextInt(5 * 1000)); System.out.println(Thread.currentThread().getName() + "下载完毕"); countDownLatch.countDown(); } catch (InterruptedException e) { 
    e.printStackTrace(); } }).start(); }); countDownLatch.await(); System.out.println("大桥未久合集下载完毕，滴滴滴滴滴"); } } 

上述案例中我们同时开启5个线程分别进行任务的处理，使用随机数来模拟任务进行过程。从运行结果来看我们可以清晰的体会到Count DownLatch在这个案例中的作用，相当于我们在一个门上加了5把锁，只有当我们把所有的锁都打开后才能开启大门执行后面的方法。至于
Doug Lea到底是怎么实现此功能的呢，我们一起到源码中分析下。

源码分析

首先我们从构造函数入手看一下我们设置的数值到底是怎么一回事

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

Sync是CountDownLatch的一个内部类，同时它也是AQS的一个子类。实际上是吧计数器的值赋给了AQS的状态变量state，也就是使用AQS的状态值来表示计数器值。

 private static final class Sync extends AbstractQueuedSynchronizer { 
    private static final long serialVersionUID = L; Sync(int count) { 
    setState(count); } int getCount() { 
    return getState(); } 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; if (compareAndSetState(c, nextc)) return nextc == 0; } } } 

设置状态变量state，其中state是个volatile 用于保证可见性

函数探究

await()：

从上面的案例中我们可以看到当调用await()方法时当前线程会被阻塞，只有当计数器的值为0，或者是其它线程调用了当前线程的interrypt()方法中断了当前线程，当前线程就会抛出InterruptedException异常然后返回。

 public void await() throws InterruptedException { 
    sync.acquireSharedInterruptibly(1); } // 获取共享资源时可被中断的方法 public final void acquireSharedInterruptibly(int arg) throws InterruptedException { 
    if (Thread.interrupted()) // 如果当前线程被中断则抛出异常 throw new InterruptedException(); if (tryAcquireShared(arg) < 0) // 如果State = 0则返回1，State != 0时返回-1 执行 doAcquireSharedInterruptibly(arg); doAcquireSharedInterruptibly(arg); } // 实现AQS中的方法 protected int tryAcquireShared(int acquires) { 
    return (getState() == 0) ? 1 : -1; } 

由上述代码可以看出线程获取资源时可以被中断，首先会判断当前线程是否被中断（Thread.interrupted()），若是则抛出异常。否则调用内部类Sync中的tryAcquireShared方法获取AQS中state的值，判断当前计数器是否为0。如果为0直接返回，否则调用AQS的doAcquireInterru ptibly方法让当前线程阻塞。

 //该方法使当前线程一直等待，直到当前线程获取到共享锁或被中断才返回 private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { 
    //根据当前线程创建一个共享模式的Node节点，并把这个节点添加到等待队列的尾部 final Node node = addWaiter(Node.SHARED); boolean failed = true; try { 
    for (;;) { 
    // 获取新建节点的前驱节点 final Node p = node.predecessor(); // 如果前驱节点是头结点 if (p == head) { 
    // 尝试获取共享锁，调用CountDownLatch中实现 // 如果当前内部计数器等于零返回1，否则返回-1 // 内部计数器等于零表示可以获取共享锁，否则不可以 int r = tryAcquireShared(arg); // 获取到共享锁 if (r >= 0) { 
    // 将前驱节点从等待队列中释放 // 同时使用LockSupport.unpark方法唤醒前驱节点的后继节点中的线程 setHeadAndPropagate(node, r); p.next = null; // help GC 将头节点的next为空彻底断开与下一个节点的连接，失去引用后方便垃圾回收 failed = false; return; } } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) throw new InterruptedException(); } } finally { 
    if (failed) cancelAcquire(node); } } 

 / Marker to indicate a node is waiting in shared mode */ static final Node SHARED = new Node(); private Node addWaiter(Node mode) { 
    Node node = new Node(Thread.currentThread(), mode); // Try the fast path of enq; backup to full enq on failure Node pred = tail; // 如果尾节点不为空(等待队列不为空)，则新节点的前驱节点指向这个尾节点, 同时尾节点指向新节点简单来说就是加入队列 if (pred != null) { 
    // 添加到尾结点 node.prev = pred; if (compareAndSetTail(pred, node)) { 
    pred.next = node; return node; } } // 如果尾节点为空(等待队列是空的) // 执行enq方法将节点插入到等待队列尾部 enq(node); return node; } 

// //判断当前节点里的线程是否需要被阻塞 private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { 
    // 前驱节点线程的状态 int ws = pred.waitStatus; // 如果前驱节点线程的状态是SIGNAL，返回true，需要阻塞线程 if (ws == Node.SIGNAL) /* * This node has already set status asking a release * to signal it, so it can safely park. */ return true; // 如果前驱节点线程的状态是CANCELLED，则设置当前节点的前驱节点为"原前驱节点的前驱节点" // 因为当前节点的前驱节点线程已经被取消了  if (ws > 0) { 
    /* * Predecessor was cancelled. Skip over predecessors and * indicate retry. */ do { 
    node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; } else { 
    //其它状态的都设置前驱节点为SIGNAL状态 /* * waitStatus must be 0 or PROPAGATE. Indicate that we * need a signal, but don't park yet. Caller will need to * retry to make sure it cannot acquire before parking. */ compareAndSetWaitStatus(pred, ws, Node.SIGNAL); } return false; } 

 // 通过使用LockSupport.park阻塞当前线程 // 同时返回当前线程是否中断 private final boolean parkAndCheckInterrupt() { 
    LockSupport.park(this); return Thread.interrupted(); } 

countDown():

/ * 此方法的作用就是将count的值-1，如果count等于0了，就唤醒等待的线程 */ public void countDown() { 
    sync.releaseShared(1); } // 这里直接调用sync的releaseShared方法，这个方法的实现在AQS中，也是AQS提供的模板方法， // 这个方法的作用是当前线程释放锁，若释放失败，返回false，若释放成功，则返回false， // 若锁被释放成功，则当前线程会唤醒AQS同步队列中第一个被阻塞的线程，让他尝试获取锁 // 对于CountDownLatch来说，释放锁实际上就是让count - 1，只有当count被减小为0， // 锁才是真正被释放，线程才能继续向下运行 public final boolean releaseShared(int arg) { 
    //如果内部计数器状态值递减后等于零 if (tryReleaseShared(arg)) { 
    //唤醒等待队列节点中的线程 doReleaseShared(); return true; } return false; } //尝试释放共享锁，即将内部计数器值减一 protected boolean tryReleaseShared(int releases) { 
    for (;;) { 
    //获取内部计数器状态值 int c = getState(); if (c == 0) return false; //计数器减一 int nextc = c-1; //使用CAS修改state值 if (compareAndSetState(c, nextc)) return nextc == 0; } } private void doReleaseShared() { 
    for (;;) { 
    //从头结点开始 Node h = head; //头结点不为空，并且不是尾节点 if (h != null && h != tail) { 
    int ws = h.waitStatus; if (ws == Node.SIGNAL) { 
    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; //唤醒阻塞的线程 unparkSuccessor(h); } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; } if (h == head) break; } } private void unparkSuccessor(Node node) { 
    int ws = node.waitStatus; if (ws < 0) compareAndSetWaitStatus(node, ws, 0); Node s = node.next; if (s == null || s.waitStatus > 0) { 
    s = null; for (Node t = tail; t != null && t != node; t = t.prev) if (t.waitStatus <= 0) s = t; } if (s != null) //通过使用LockSupport.unpark唤醒线程 LockSupport.unpark(s.thread); }