CountDownLatch源碼之a(chǎn)wait()有什么用

這篇文章主要介紹CountDownLatch源碼之a(chǎn)wait()有什么用，文中介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們一定要看完！

具體內(nèi)容如下

我們已經(jīng)知道await 能夠讓當前線程處于阻塞狀態(tài)，直到鎖存器計數(shù)為零（或者線程中斷）。

下面是它的源碼。

end.await(); 
  ↓
public void await() throws InterruptedException {
  sync.acquireSharedInterruptibly(1);
}

sync 是CountDownLatch的內(nèi)部類。下面是它的定義。

private static final class Sync extends AbstractQueuedSynchronizer {
　　...
}

它繼承了AbstractQueuedSynchronizer。AbstractQueuedSynchronizer 這個類在java線程中屬于一個非常重要的類。

它提供了一個框架來實現(xiàn)阻塞鎖，以及依賴FIFO等待隊列的相關(guān)同步器（比如信號、事件等）。

繼續(xù)走下去，就跳到 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 中實現(xiàn)的。

AbstractQueuedSynchronizer 中雖然也有對它的實現(xiàn)，但是默認的實現(xiàn)是拋一個異常。

tryAcquireShared 這個方法是用來查詢當前對象的狀態(tài)是否能夠被允許獲取鎖。

我們可以看到Sync 中是通過判斷state 是否為0 來返回對應(yīng)的 int 值的。

那么 state 又代表什么？ 

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

上面代碼很清楚的表明 state 是表示同步的狀態(tài) 。

需要注意的是 state 使用 volatile 關(guān)鍵字修飾。

volatile 關(guān)鍵字能夠保證 state 的修改立即被更新到主存，當有其他線程需要讀取時，會去內(nèi)存中讀取新值。

也就是保證了state的可見性。是最新的數(shù)據(jù)。

走到這里 state 是多少呢？

這里我們就需要看一看CountDownLatch 的 構(gòu)造函數(shù)了。

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

原來構(gòu)造函數(shù)中的數(shù)字就是這個作用啊，用來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，這段代碼有點長，里面還調(diào)用了幾個函數(shù)。我們一行一行的看。

第一行 出現(xiàn)了一個新的類 Node。

Node 是AQS（AbstractQueuedSynchronizer）類中的內(nèi)部類，定義了一種鏈式結(jié)構(gòu)。如下所示。

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

千萬記住這個結(jié)構(gòu)。

第一行代碼中還有一個方法 addWaiter(Node.SHARED) 。

addWaiter(Node.SHARED) //Node.SHARED 表示該結(jié)點處于共享模式
  ↓
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;
}

首先是構(gòu)造了一個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 就會被設(shè)置為 update 的值，也就是一個新的結(jié)點。

 tail = head;  那么此時 tail 也不再是null了。進入下一次的循環(huán)。

這次首先將node 的 prev 指針指向 tail ，然后通過一個CAS 操作將node 設(shè)置為尾部，并返回了隊列的 tail ，也就是 node 。

等待隊列的模型變化如下

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

ok，到了這里await 方法 就返回了，是一個 thread 等于當前線程的Node。

返回到 doAcquireSharedInterruptibly(int arg) 中，進入下面循環(huán)。

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

這個時候假設(shè)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 設(shè)置為 Node.SIGNAL 。

Node.SIGNAL 表示后續(xù)結(jié)點中的線程需要被unparking（類似被喚醒的意思）。該方法返回false。

經(jīng)過這輪循環(huán)，隊列模型變成下面狀態(tài)

+--------------------------+  prev      +------------------+
head  | waitStatus = Node.SIGNAL | <---- node(tail) | currentThread  |
    +--------------------------+         +------------------+

因為shouldParkAfterFailedAcquire返回的是false，所以后面這個條件就不再看了。繼續(xù) for (;;)  中的循環(huán)。

如果state仍然大于0，再次進入到 shouldParkAfterFailedAcquire。

這次因為head 中的waitStatus 為 Node.SIGNAL ，所以 shouldParkAfterFailedAcquire 返回true。

這次就需要看parkAndCheckInterrupt 這個方法了。

 private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
  }

ok，線程沒有被中斷，所以，返回false。繼續(xù) for (;;)  中的循環(huán)。

如果state 一直大于0，并且線程一直未被中斷，那么就一直在這個循環(huán)中。也就是我們上篇文章說的裁判一直不愿意宣布比賽結(jié)束的情況。

那么什么情況下跳出循環(huán)呢？也就是什么情況下state 會 小于0呢？ 下一篇文章 我將說明。

總結(jié)一下，await()  方法 其實就是初始化一個隊列，將需要等待的線程（state > 0）加入一個隊列中，并用waitStatus 標記后繼結(jié)點的線程狀態(tài)。

以上是“CountDownLatch源碼之a(chǎn)wait()有什么用”這篇文章的所有內(nèi)容，感謝各位的閱讀！希望分享的內(nèi)容對大家有幫助，更多相關(guān)知識，歡迎關(guān)注億速云行業(yè)資訊頻道！