CountDownLatch 源碼解析—— countDown()
上一篇文章從源碼層面說了一下CountDownLatch 中 await() 的原理。這篇文章說一下countDown() 。
public void countDown() { //CountDownLatch sync.releaseShared(1);} ↓public final boolean releaseShared(int arg) { //AQS if (tryReleaseShared(arg)) { doReleaseShared(); return true; } return false;} ↓protected boolean tryReleaseShared(int releases) { //CountDownLatch.Sync // Decrement count; signal when transition to zero for (;;) { int c = getState(); if (c == 0) return false; int nextc = c-1; if (compareAndSetState(c, nextc)) return nextc == 0; }}通過構造器 CountDownLatch end = new CountDownLatch(2); state 被設置為2，所以c == 2，nextc = 2-1,
然后通過下面這個CAS操作將state設置為1。
protected final boolean compareAndSetState(int expect, int update) { // See below for intrinsics setup to support this return unsafe.compareAndSwapInt(this, stateOffset, expect, update); }此時nextc還不為0，返回false。一直等到countDown() 方法被調用兩次，state == 0，nextc ==0,此時返回true。
進入doReleaseShared()方法。
doReleaseShared(); ↓private void doReleaseShared() { /* * Ensure that a release propagates, even if there are other * in-progress acquires/releases. This proceeds in the usual * way of trying to unparkSuccessor of head if it needs * signal. But if it does not, status is set to PROPAGATE to * ensure that upon release, propagation continues. * Additionally, we must loop in case a new node is added * while we are doing this. Also, unlike other uses of * unparkSuccessor, we need to know if CAS to reset status * fails, if so rechecking. */ for (;;) { Node h = head; if (h != null && h != tail) { int ws = h.waitStatus; if (ws == Node.SIGNAL) { if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; // loop to recheck cases unparkSuccessor(h); } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; // loop on failed CAS } if (h == head) // loop if head changed break; }}回顧一下此時的等待隊列模型。
+--------------------------+ prev +------------------+head | waitStatus = Node.SIGNAL | <---- node(tail) | currentThread | +--------------------------+ +------------------+
此時head 不為null，也不為tail，waitStatus == Node.SIGNAL，所以進入 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) 這個判斷。
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) ↓ /** * CAS waitStatus field of a node. */private static final boolean compareAndSetWaitStatus(Node node, int expect, int update) { return unsafe.compareAndSwapInt(node, waitStatusOffset, expect, update);}這個CAS 操作將 state 設置為 0 ，也就是說此時Head 中的 waitStatus 是0.此時隊列模型如下所示
+----------------+ prev +------------------+head | waitStatus = 0 | <---- node(tail) | currentThread | +----------------+ +------------------+
該方法返回true。進入unparkSuccessor(h);
unparkSuccessor(h); ↓private void unparkSuccessor(Node node) { /* * If status is negative (i.e., possibly needing signal) try * to clear in anticipation of signalling. It is OK if this * fails or if status is changed by waiting thread. */ int ws = node.waitStatus; if (ws < 0) compareAndSetWaitStatus(node, ws, 0); /* * Thread to unpark is held in successor, which is normally * just the next node. But if cancelled or apparently null, * traverse backwards from tail to find the actual * non-cancelled successor. */ Node s = node.next; if (s == null || s.waitStatus > 0) { s = null; for (Node t = tail; t != null && t != node; t = t.prev) if (t.waitStatus <= 0) s = t; } if (s != null) LockSupport.unpark(s.thread);}s 就是head的后繼結點，也就是裝有當前線程的結點。s != null ，并且s.waitStatus ==0 ，所以進入 LockSupport.unpark(s.thread);
public static void unpark(Thread thread) { if (thread != null) UNSAFE.unpark(thread); }也就是unlock 被阻塞的線程。裁判被允許吹哨了！
countDown() 的原理就此就非常清晰了。
每執行一次countDown() 方法，state 就是減1，直到state == 0，則開始釋放被阻塞在隊列中的線程，根據前驅結點中waitStatus的狀態，釋放后續結點中的線程。
OK,回到上一篇文章的問題，什么時候跳出下面這個循環（await方法中的循環）
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，所以進入 setHeadAndPropagate 方法。
setHeadAndPropagate(node, r); ↓private void setHeadAndPropagate(Node node, int propagate) { Node h = head; // Record old head for check below setHead(node); /* * Try to signal next queued node if: * Propagation was indicated by caller, * or was recorded (as h.waitStatus either before * or after setHead) by a previous operation * (note: this uses sign-check of waitStatus because * PROPAGATE status may transition to SIGNAL.) * and * The next node is waiting in shared mode, * or we don't know, because it appears null * * The conservatism in both of these checks may cause * unnecessary wake-ups, but only when there are multiple * racing acquires/releases, so most need signals now or soon * anyway. */ if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) { Node s = node.next; if (s == null || s.isShared()) doReleaseShared(); }} ↓private void setHead(Node node) { head = node; node.thread = null; node.prev = null;}這個方法將head 的后繼結點變為head。該方法過后，又將node的next結點設置為null，模型變成下圖
prev +---------+ nextnull <---- node(tail/head) | null | ----> null +---------+
也就是node head tail 什么的都被置為null，等待GC回收了，這個時候return，跳出了for循環，隊列被清空。
下面演示一下整個過程
setHeadAndPropagate(node, r); +----------------+ head(tail) | waitStatus=0 | | thread =null | +----------------+ ↓ +----------------+ +----------------+ | waitStatus=0 | prev | waitStatus=0 |head(tail) | thread =null | <---- node | currentThread | +----------------+ +----------------+ ↓ +----------------+ +----------------+ | waitStatus=0 | prev | waitStatus=0 |head | thread =null | <---- node(tail) | currentThread | +----------------+ +----------------+ ↓ +----------------+ +----------------+ | Node.SIGNAL | prev | waitStatus=0 |head | thread =null | <---- node(tail) | currentThread | +----------------+ +----------------+ ↓ +----------------+ +----------------+ | waitStatus=0 | prev | waitStatus=0 |head | thread =null | <---- node(tail) | currentThread | +----------------+ +----------------+ ↓ +----------------+ prev | waitStatus=0 | nextnull <---- node(tail/head) | null | ----> null +----------------+
CountDownLatch 的核心就是一個阻塞線程隊列，這是由鏈表構造而成的隊列，里面包含thread 和 waitStatus，其中waitStatus說明了后繼結點線程狀態。
state 是一個非常重要的標志，構造時，設置為對應的n值，如果n != 0，阻塞隊列將一直阻塞，除非中斷線程。
每次調用countDown() 方法，就是將state-1，而調用await() 方法就是將調用該方法的線程加入到阻塞隊列，直到state==0，才能釋放線程。
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持VeVb武林網。
新聞熱點
疑難解答
圖片精選
