详解CountDownLatch

时间  2019-11-26
标签 详解 countdownlatch 繁體版
原文   原文链接

CountDownLatch是什么？ 

jdk1.5开始concurrent包里提供的，并发编程工具类。 node

CountDownLatch这个类可以使一个线程等待其余线程完成各自的工做后再执行。CountDownLatch容许一个或多个线程等待其余线程完成操做。 算法

例如，应用程序的主线程但愿在负责启动框架服务的线程已经启动全部的框架服务以后再执行。编程

CountDownLatch很是适合于对任务进行拆分，使其并行执行，好比某个任务执行2s，其对数据的请求能够分为五个部分，那么就能够将这个任务拆分为5个子任务，分别交由五个线程执行，执行完成以后再由主线程进行汇总，此时，总的执行时间将决定于执行最慢的任务，平均来看，仍是大大减小了总的执行时间。bash

CountDownLatch是不能复用的，不可能从新初始化或者修改CountDownLatch对象的内部计数器的值。
多线程

CountDownLatch如何工做？

CountDownLatch是经过维护一个计数器 cnt 来实现的，计数器的初始值为线程的数量。每当一个线程完成了本身的任务后，调用 countDown() 方法会让计数器的值就会减1。当计数器值到达0时，它表示全部的线程已经完成了任务，那些由于调用 await() 方法而在等待的线程就会被唤醒。
API
  • countDownLatch.countDown(); //使CountDownLatch初始值N减1；
  • countDownLatch.await(); //调用该方法的线程等到构造方法传入的N减到0的时候，才能继续往下执行；
  • await(long timeout, TimeUnit unit); //与上面的await方法功能一致，只不过这里有了时间限制，调用该方法的线程等到指定的timeout时间后，无论N是否减至为0，都会继续往下执行；
  • long getCount(); //获取当前CountDownLatch维护的值；

CountDownLatch底层原理

CountDownLatch经过AQS（AbstractQueuedSynchronizer）里面的共享锁来实现的。
ReentrantLock也是使用AQS

CountDownLatch是基于AbstractQueuedSynchronizer实现的，在AbstractQueuedSynchronizer中维护了一个volatile类型的整数state，volatile能够保证多线程环境下该变量的修改对每一个线程均可见，而且因为该属性为整型，于是对该变量的修改也是原子的。 建立一个CountDownLatch对象时，所传入的整数n就会赋值给state属性，当countDown()方法调用时，该线程就会尝试对state减一，而调用await()方法时，当前线程就会判断state属性是否为0，若是为0，则继续往下执行，若是不为0，则使当前线程进入等待状态，直到某个线程将state属性置为0，其就会唤醒在await()方法中等待的线程。以下是countDown()方法的源代码： 

public void countDown() {
  sync.releaseShared(1);
}复制代码
这里sync也即一个继承了AbstractQueuedSynchronizer的类实例，该类是CountDownLatch的一个内部类，其声明以下： 

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

  Sync(int count) {
    setState(count);
  }

  int getCount() {
    return getState();
  }

  protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1;
  }

  protected boolean tryReleaseShared(int releases) {
    for (;;) {
      int c = getState();   // 获取当前state属性的值
      if (c == 0)   // 若是state为0，则说明当前计数器已经计数完成，直接返回
        return false;
      int nextc = c-1;
      if (compareAndSetState(c, nextc)) // 使用CAS算法对state进行设置
        return nextc == 0;  // 设置成功后返回当前是否为最后一个设置state的线程
    }
  }
}复制代码
      这里tryReleaseShared(int)方法即对state属性进行减一操做的代码。能够看到，CAS也即compare and set的缩写，jvm会保证该方法的原子性，其会比较state是否为c，若是是则将其设置为nextc（自减1），若是state不为c，则说明有另外的线程在getState()方法和compareAndSetState()方法调用之间对state进行了设置，当前线程也就没有成功设置state属性的值，其会进入下一次循环中，如此往复，直至其成功设置state属性的值，即countDown()方法调用成功。

在countDown()方法中调用的sync.releaseShared(1)调用时实际仍是调用的tryReleaseShared(int)方法，以下是releaseShared(int)方法的实现： 

public final boolean releaseShared(int arg) {
  if (tryReleaseShared(arg)) {
    doReleaseShared();
    return true;
  }
  return false;
}复制代码
        能够看到，在执行sync.releaseShared(1)方法时，其在调用tryReleaseShared(int)方法时会在无限for循环中设置state属性的值，设置成功以后其会根据设置的返回值（此时state已经自减了一），即当前线程是否为将state属性设置为0的线程，来判断是否执行if块中的代码。doReleaseShared()方法主要做用是唤醒调用了await()方法的线程。须要注意的是，若是有多个线程调用了await()方法，这些线程都是以共享的方式等待在await()方法处的，试想，若是以独占的方式等待，那么当计数器减小至零时，就只有一个线程会被唤醒执行await()以后的代码，这显然不符合逻辑。以下是doReleaseShared()方法的实现代码： 

private void doReleaseShared() {
  for (;;) {
    Node h = head;  // 记录等待队列中的头结点的线程
    if (h != null && h != tail) {   // 头结点不为空，且头结点不等于尾节点
      int ws = h.waitStatus;
      if (ws == Node.SIGNAL) {  // SIGNAL状态表示当前节点正在等待被唤醒
        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))    // 清除当前节点的等待状态
          continue;
        unparkSuccessor(h); // 唤醒当前节点的下一个节点
      } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
        continue;
    }
    if (h == head)  // 若是h仍是指向头结点，说明前面这段代码执行过程当中没有其余线程对头结点进行过处理
      break;
  }
}复制代码
在doReleaseShared()方法中（始终注意当前方法是最后一个执行countDown()方法的线程执行的），首先判断头结点不为空，且不为尾节点，说明等待队列中有等待唤醒的线程，这里须要说明的是，在等待队列中，头节点中并无保存正在等待的线程，其只是一个空的Node对象，真正等待的线程是从头节点的下一个节点开始存放的，于是会有对头结点是否等于尾节点的判断。在判断等待队列中有正在等待的线程以后，其会清除头结点的状态信息，而且调用unparkSuccessor(Node)方法唤醒头结点的下一个节点，使其继续往下执行。以下是unparkSuccessor(Node)方法的具体实现： 

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的等待状态大于0说明该节点中的线程已经被外部取消等待了
    s = null;
    // 从队列尾部往前遍历，找到最后一个处于等待状态的节点，用s记录下来
    for (Node t = tail; t != null && t != node; t = t.prev)
      if (t.waitStatus <= 0)
        s = t;
  }
  if (s != null)
    LockSupport.unpark(s.thread);   // 唤醒离传入节点最近的处于等待状态的节点线程
}复制代码
能够看到，unparkSuccessor(Node)方法的做用是唤醒离传入节点最近的一个处于等待状态的线程，使其继续往下执行。前面咱们讲到过，等待队列中的线程可能有多个，而调用countDown()方法的线程只唤醒了一个处于等待状态的线程，这里剩下的等待线程是如何被唤醒的呢？其实这些线程是被当前唤醒的线程唤醒的。具体的咱们能够看看await()方法的具体执行过程。以下是await()方法的代码： 

public void await() throws InterruptedException {
  sync.acquireSharedInterruptibly(1);
}
复制代码
     await()方法实际仍是调用了Sync对象的方法acquireSharedInterruptibly(int)方法，以下是该方法的具体实现： 

public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
  if (Thread.interrupted())
    throw new InterruptedException();
  if (tryAcquireShared(arg) < 0)
    doAcquireSharedInterruptibly(arg);
}复制代码

     能够看到acquireSharedInterruptibly(int)方法判断当前线程是否须要以共享状态获取执行权限，这里tryAcquireShared(int)方法是AbstractQueuedSynchronizer中的一个模板方法，其具体实如今前面的Sync类中，能够看到，其主要是判断state是否为零，若是为零则返回1，表示当前线程不须要进行权限获取，可直接执行后续代码，返回-1则表示当前线程须要进行共享权限。具体的获取执行权限的代码在doAcquireSharedInterruptibly(int)方法中，以下是该方法的具体实现： 

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) {   // 大于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);
  }
}复制代码
      在doAcquireSharedInterruptibly(int)方法中，首先使用当前线程建立一个共享模式的节点。而后在一个for循环中判断当前线程是否获取到执行权限，若是有（r >= 0判断）则将当前节点设置为头节点，而且唤醒后续处于共享模式的节点；若是没有，则对调用shouldParkAfterFailedAcquire(Node, Node)和parkAndCheckInterrupt()方法使当前线程处于“搁置”状态，该“搁置”状态是由操做系统进行的，这样能够避免该线程无限循环而获取不到执行权限，形成资源浪费，这里也就是线程处于等待状态的位置，也就是说当线程被阻塞的时候就是阻塞在这个位置。当有多个线程调用await()方法而进入等待状态时，这几个线程都将等待在此处。这里回过头来看前面将的countDown()方法，其会唤醒处于等待队列中离头节点最近的一个处于等待状态的线程，也就是说该线程被唤醒以后会继续从这个位置开始往下执行，此时执行到tryAcquireShared(int)方法时，发现r大于0（由于state已经被置为0了），该线程就会调用setHeadAndPropagate(Node, int)方法，而且退出当前循环，也就开始执行awat()方法以后的代码。这里咱们看看setHeadAndPropagate(Node, int)方法的具体实现： 

private void setHeadAndPropagate(Node node, int propagate) {
  Node h = head;
  setHead(node);    // 将当前节点设置为头节点
  // 检查唤醒过程是否须要往下传递，而且检查头结点的等待状态
  if (propagate > 0 || h == null || h.waitStatus < 0 ||
      (h = head) == null || h.waitStatus < 0) {
    Node s = node.next;
    if (s == null || s.isShared())  // 若是下一个节点是尝试以共享状态获取获取执行权限的节点，则将其唤醒
      doReleaseShared();
  }
}复制代码

     setHeadAndPropagate(Node, int)方法主要做用是设置当前节点为头结点，而且将唤醒工做往下传递，在传递的过程当中，其会判断被传递的节点是不是以共享模式尝试获取执行权限的，若是不是，则传递到该节点处为止（通常状况下，等待队列中都只会都是处于共享模式或者处于独占模式的节点）。也就是说，头结点会依次唤醒后续处于共享状态的节点，这也就是共享锁与独占锁的实现方式。这里doReleaseShared()方法也就是咱们前面讲到的会将离头结点最近的一个处于等待状态的节点唤醒的方法。

CountDownLatch的demo

public class CountdownLatchExample {
    public static void main(String[] args) throws InterruptedException {
        final int totalThread = 10;
        CountDownLatch countDownLatch = new CountDownLatch(totalThread);
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < totalThread; i++) {
            executorService.execute(() -> {
                System.out.print("run..");
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        System.out.println("end");
        executorService.shutdown();
    }
}
run..run..run..run..run..run..run..run..run..run..end复制代码

总结

CountDownLatch的做用就是容许一个或多个线程等待其余线程完成操做，看起来有点相似join() 方法，但其提供了比 join() 更加灵活的API。CountDownLatch能够手动控制在n个线程里调用n次countDown方法使计数器进行减一操做，也能够在一个线程里调用n次执行减一操做。而 join() 的实现原理是不停检查join线程是否存活，若是 join 线程存活则让当前线程永远等待。因此二者之间相对来讲仍是CountDownLatch使用起来较为灵活。并发


参考自：《Java并发编程的艺术》和www.jianshu.com/p/128476015…框架

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程