這篇文章主要介紹CountDownLatch源碼之a(chǎn)wait()有什么用,文中介紹的非常詳細(xì),具有一定的參考價值,感興趣的小伙伴們一定要看完!
創(chuàng)新互聯(lián)堅持“要么做到,要么別承諾”的工作理念,服務(wù)領(lǐng)域包括:網(wǎng)站設(shè)計、成都網(wǎng)站制作、企業(yè)官網(wǎng)、英文網(wǎng)站、手機端網(wǎng)站、網(wǎng)站推廣等服務(wù),滿足客戶于互聯(lián)網(wǎng)時代的石臺網(wǎng)站設(shè)計、移動媒體設(shè)計的需求,幫助企業(yè)找到有效的互聯(lián)網(wǎng)解決方案。努力成為您成熟可靠的網(wǎng)絡(luò)建設(shè)合作伙伴!
具體內(nèi)容如下
我們已經(jīng)知道await 能夠讓當(dāng)前線程處于阻塞狀態(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); }
這里有兩個判斷,首先判斷線程是否中斷,然后再進(jìn)行下一個判斷,這里我們主要看看第二個判斷。
protected int tryAcquireShared(int acquires) { return (getState() == 0) ? 1 : -1; }
需要注意的是 tryAcquireShared 這個方法是在Sync 中實現(xiàn)的。
AbstractQueuedSynchronizer 中雖然也有對它的實現(xiàn),但是默認(rèn)的實現(xiàn)是拋一個異常。
tryAcquireShared 這個方法是用來查詢當(dāng)前對象的狀態(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 的修改立即被更新到主存,當(dāng)有其他線程需要讀取時,會去內(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。進(jìn)入到下面
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)部類,定義了一種鏈?zhǔn)浇Y(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,將當(dāng)前的線程存進(jìn)去了,模式是共享模式。
tail 表示 這個等待隊列的隊尾,此刻是null. 所以 pred == null ,進(jìn)入到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 , 進(jìn)入到 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了。進(jìn)入下一次的循環(huán)。
這次首先將node 的 prev 指針指向 tail ,然后通過一個CAS 操作將node 設(shè)置為尾部,并返回了隊列的 tail ,也就是 node 。
等待隊列的模型變化如下
+------+ prev +----------------+ head(tail) | | <---- node | currentThread | +------+ +----------------+ ↓ +------+ prev +----------------+ head | | <---- node(tail) | currentThread | +------+ +----------------+
ok,到了這里await 方法 就返回了,是一個 thread 等于當(dāng)前線程的Node。
返回到 doAcquireSharedInterruptibly(int arg) 中,進(jìn)入下面循環(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,所以進(jìn)入到 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,再次進(jìn)入到 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 標(biāo)記后繼結(jié)點的線程狀態(tài)。
以上是“CountDownLatch源碼之a(chǎn)wait()有什么用”這篇文章的所有內(nèi)容,感謝各位的閱讀!希望分享的內(nèi)容對大家有幫助,更多相關(guān)知識,歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道!