CountDownLatch源码解析之await()

countdownlatch 源码解析—— await()，具体内容如下

说了一下countdownlatch的使用方法。这篇文章就从源码层面说一下await() 的原理。

我们已经知道await 能够让当前线程处于阻塞状态，直到锁存器计数为零（或者线程中断）。

下面是它的源码。

end.await(); 
  ↓
public void await() throws interruptedexception {
  sync.acquiresharedinterruptibly(1);
}

sync 是countdownlatch的内部类。下面是它的定义。

private static final class sync extends abstractqueuedsynchronizer {
　　...
}

它继承了abstractqueuedsynchronizer。abstractqueuedsynchronizer 这个类在java线程中属于一个非常重要的类。

它提供了一个框架来实现阻塞锁，以及依赖fifo等待队列的相关同步器（比如信号、事件等）。

继续走下去，就跳到 abstractqueuedsynchronizer 这个类中。

sync.acquiresharedinterruptibly(1); 
  ↓
public final void acquiresharedinterruptibly(int arg) //abstractqueuedsynchronizer
      throws interruptedexception {
  if (thread.interrupted())
    throw new interruptedexception();
  if (tryacquireshared(arg) < 0)
    doacquiresharedinterruptibly(arg);
}

这里有两个判断，首先判断线程是否中断，然后再进行下一个判断，这里我们主要看看第二个判断。 

protected int tryacquireshared(int acquires) {
  return (getstate() == 0) ? 1 : -1;
}

需要注意的是 tryacquireshared 这个方法是在sync 中实现的。

abstractqueuedsynchronizer 中虽然也有对它的实现，但是默认的实现是抛一个异常。

tryacquireshared 这个方法是用来查询当前对象的状态是否能够被允许获取锁。

我们可以看到sync 中是通过判断state 是否为0 来返回对应的 int 值的。

那么 state 又代表什么？ 

/**
 * the synchronization state.
 */
  private volatile int state;

上面代码很清楚的表明 state 是表示同步的状态 。

需要注意的是 state 使用 volatile 关键字修饰。

volatile 关键字能够保证 state 的修改立即被更新到主存，当有其他线程需要读取时，会去内存中读取新值。

也就是保证了state的可见性。是最新的数据。

走到这里 state 是多少呢？

这里我们就需要看一看countdownlatch 的 构造函数了。

countdownlatch end = new countdownlatch(2);
  ↓
public countdownlatch(int count) {
  if (count < 0) throw new illegalargumentexception("count < 0");
  this.sync = new sync(count);
}
  ↓
sync(int count) {
  setstate(count);
}

原来构造函数中的数字就是这个作用啊，用来set state 。

所以我们这里state == 2 了。tryacquireshared 就返回 -1。进入到下面

doacquiresharedinterruptibly(arg);
  ↓
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);
    }
  }

ok，这段代码有点长，里面还调用了几个函数。我们一行一行的看。

第一行 出现了一个新的类 node。

node 是aqs（abstractqueuedsynchronizer）类中的内部类，定义了一种链式结构。如下所示。

   +------+ prev +-----+    +-----+
head |   | <---- |   | <---- |   | tail
   +------+    +-----+    +-----+

千万记住这个结构。

第一行代码中还有一个方法 addwaiter(node.shared) 。

addwaiter(node.shared) //node.shared 表示该结点处于共享模式
  ↓
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; // private transient volatile node tail;
  if (pred != null) {
    node.prev = pred;
    if (compareandsettail(pred, node)) {
      pred.next = node;
      return node;
    }
  }
  enq(node);
  return node;
}

首先是构造了一个node，将当前的线程存进去了，模式是共享模式。

tail 表示 这个等待队列的队尾，此刻是null. 所以 pred == null ，进入到enq(node) ;

enq(node)
  ↓
private node enq(final node node) {
  for (;;) {
    node t = tail;
    if (t == null) { // must initialize
      if (compareandsethead(new node()))
        tail = head;
    } else {
      node.prev = t;
      if (compareandsettail(t, node)) {
        t.next = node;
        return t;
      }
    }
  }
}

同样tail 为 null , 进入到 compareandsethead 。

compareandsethead(new node())
  ↓
/**
 * cas head field. used only by enq.
 */
private final boolean compareandsethead(node update) {
  return unsafe.compareandswapobject(this, headoffset, null, update);
}

这是一个cas操作，如果head 是 null 的话，等待队列的 head 就会被设置为 update 的值，也就是一个新的结点。

 tail = head;  那么此时 tail 也不再是null了。进入下一次的循环。

这次首先将node 的 prev 指针指向 tail ，然后通过一个cas 操作将node 设置为尾部，并返回了队列的 tail ，也就是 node 。

等待队列的模型变化如下

      +------+ prev   +----------------+
head(tail) |   | <---- node | currentthread |
      +------+      +----------------+
      
          ↓
          
    +------+ prev      +----------------+
head  |   | <---- node(tail) | currentthread |
    +------+         +----------------+

ok，到了这里await 方法 就返回了，是一个 thread 等于当前线程的node。

返回到 doacquiresharedinterruptibly(int arg) 中，进入下面循环。

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();
}

这个时候假设state 仍然大于0，那么此时 r < 0，所以进入到 shouldparkafterfailedacquire 这个方法 。

shouldparkafterfailedacquire(p, node)
  ↓
private static boolean shouldparkafterfailedacquire(node pred, node node) {
  int ws = pred.waitstatus;
  if (ws == node.signal) //static final int signal  = -1;
    /*
     * this node has already set status asking a release
     * to signal it, so it can safely park.
     */
    return true;
  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 {
    /*
     * 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;
}
  ↓
/**
 * cas waitstatus field of a node.
 */
private static final boolean compareandsetwaitstatus(node node,
                           int expect,
                           int update) {
  return unsafe.compareandswapint(node, waitstatusoffset,
                  expect, update);
}

可以看到 shouldparkafterfailedacquire  也是一路走，走到 compareandsetwaitstatus。

compareandsetwaitstatus 将 prev 的 waitstatus 设置为 node.signal 。

node.signal 表示后续结点中的线程需要被unparking（类似被唤醒的意思）。该方法返回false。

经过这轮循环，队列模型变成下面状态

    +--------------------------+  prev      +------------------+
head  | waitstatus = node.signal | <---- node(tail) | currentthread  |
    +--------------------------+         +------------------+

因为shouldparkafterfailedacquire返回的是false，所以后面这个条件就不再看了。继续 for (;;)  中的循环。

如果state仍然大于0，再次进入到 shouldparkafterfailedacquire。

这次因为head 中的waitstatus 为 node.signal ，所以 shouldparkafterfailedacquire 返回true。

这次就需要看parkandcheckinterrupt 这个方法了。

 private final boolean parkandcheckinterrupt() {
    locksupport.park(this);
    return thread.interrupted();
  }

ok，线程没有被中断，所以，返回false。继续 for (;;)  中的循环。

如果state 一直大于0，并且线程一直未被中断，那么就一直在这个循环中。也就是我们上篇文章说的裁判一直不愿意宣布比赛结束的情况。

那么什么情况下跳出循环呢？也就是什么情况下state 会 小于0呢？ 下一篇文章 我将说明。

总结一下，await()  方法 其实就是初始化一个队列，将需要等待的线程（state > 0）加入一个队列中，并用waitstatus 标记后继结点的线程状态。

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。