latch.await java有什么作用_java相关：CountDownLatch源码解析之await()
java相關：CountDownLatch源碼解析之await()
發(fā)布于 2020-6-18|
復制鏈接
摘記: CountDownLatch 源碼解析—— await()，具體內容如下上一篇文章說了一下CountDownLatch的使用方法。這篇文章就從源碼層面說一下await() 的原理。我們已經知道awa ..
CountDownLatch 源碼解析—— await()，具體內容如下上一篇文章說了一下CountDownLatch的使用方法。這篇文章就從源碼層面說一下await() 的原理。我們已經知道await 能夠讓當前線程處于阻塞狀態(tài)，直到鎖存器計數(shù)為零(或者線程中斷)。下面是它的源碼。
```java
end.await();
↓
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
```
sync 是CountDownLatch的內部類。下面是它的定義。
```java
private static final class Sync extends AbstractQueuedSynchronizer {
...
}
```
它繼承了AbstractQueuedSynchronizer。AbstractQueuedSynchronizer 這個類在java線程中屬于一個非常重要的類。它提供了一個框架來實現(xiàn)阻塞鎖，以及依賴FIFO等待隊列的相關同步器(比如信號、事件等)。繼續(xù)走下去，就跳到 AbstractQueuedSynchronizer 這個類中。
```java
sync.acquireSharedInterruptibly(1);
↓
public final void acquireSharedInterruptibly(int arg) //AbstractQueuedSynchronizer
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) 這里有兩個判斷，首先判斷線程是否中斷，然后再進行下一個判斷，這里我們主要看看第二個判斷。
```java
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 來返回對應的 int 值的。那么 state 又代表什么？
```java
/**
* The synchronization state.
*/
private volatile int state;
```
上面代碼很清楚的表明 state 是表示同步的狀態(tài) 。需要注意的是 state 使用 volatile 關鍵字修飾。volatile 關鍵字能夠保證 state 的修改立即被更新到主存，當有其他線程需要讀取時，會去內存中讀取新值。也就是保證了state的可見性。是最新的數(shù)據(jù)。走到這里 state 是多少呢？這里我們就需要看一看CountDownLatch 的 構造函數(shù)了。
```java
CountDownLatch end = new CountDownLatch(2);
↓
public CountDownLatch(int count) {
if (count 原來構造函數(shù)中的數(shù)字就是這個作用啊，用來set state 。所以我們這里state == 2 了。tryAcquireShared 就返回 -1。進入到下面
```java
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，這段代碼有點長，里面還調用了幾個函數(shù)。我們一行一行的看。第一行 出現(xiàn)了一個新的類 Node。Node 是AQS(AbstractQueuedSynchronizer)類中的內部類，定義了一種鏈式結構。如下所示。
```java
+------+ prev +-----+ +-----+
head | | 千萬記住這個結構。第一行代碼中還有一個方法 addWaiter(Node.SHARED) 。
```java
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) ;
```java
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 。
```java
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了。進入下一次的循環(huán)。這次首先將node 的 prev 指針指向 tail ，然后通過一個CAS 操作將node 設置為尾部，并返回了隊列的 tail ，也就是 node 。等待隊列的模型變化如下
```java
+------+ prev +----------------+
head(tail) | | ok，到了這里await 方法 就返回了，是一個 thread 等于當前線程的Node。返回到 doAcquireSharedInterruptibly(int arg) 中，進入下面循環(huán)。
```java
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
```java
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 表示后續(xù)結點中的線程需要被unparking(類似被喚醒的意思)。該方法返回false。經過這輪循環(huán)，隊列模型變成下面狀態(tài)
```java
+--------------------------+ prev +------------------+
head | waitStatus = Node.SIGNAL | 因為shouldParkAfterFailedAcquire返回的是false，所以后面這個條件就不再看了。繼續(xù) for (;;)? 中的循環(huán)。如果state仍然大于0，再次進入到 shouldParkAfterFailedAcquire。這次因為head 中的waitStatus 為 Node.SIGNAL ，所以 shouldParkAfterFailedAcquire 返回true。這次就需要看parkAndCheckInterrupt 這個方法了。
```java
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
```
總結
以上是生活随笔為你收集整理的latch.await java有什么作用_java相关：CountDownLatch源码解析之await()的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 鸭脖为什么不能吃？
- 下一篇： java 使用ehcache,ehcac