Java多線程系列--「JUC鎖」09之 CountDownLatch原理和示例

時間 2019-11-06

標籤 java 多線程系列 juc countdownlatch 原理示例欄目 Java 简体版

原文原文鏈接

概要

前面對"獨佔鎖"和"共享鎖"有了個大體的瞭解；本章，咱們對CountDownLatch進行學習。和ReadWriteLock.ReadLock同樣，CountDownLatch的本質也是一個"共享鎖"。本章的內容包括：
CountDownLatch簡介
 CountDownLatch數據結構
CountDownLatch源碼分析(基於JDK1.7.0_40)
CountDownLatch示例html

轉載請註明出處：http://www.cnblogs.com/skywang12345/p/3533887.htmljava

CountDownLatch簡介

CountDownLatch是一個同步輔助類，在完成一組正在其餘線程中執行的操做以前，它容許一個或多個線程一直等待。node

CountDownLatch和CyclicBarrier的區別
(01) CountDownLatch的做用是容許1或N個線程等待其餘線程完成執行；而CyclicBarrier則是容許N個線程相互等待。
(02) CountDownLatch的計數器沒法被重置；CyclicBarrier的計數器能夠被重置後使用，所以它被稱爲是循環的barrier。
關於CyclicBarrier的原理，後面一章再來學習。數據結構

CountDownLatch函數列表多線程

CountDownLatch(int count)
構造一個用給定計數初始化的 CountDownLatch。

// 使當前線程在鎖存器倒計數至零以前一直等待，除非線程被中斷。
void await()
// 使當前線程在鎖存器倒計數至零以前一直等待，除非線程被中斷或超出了指定的等待時間。
boolean await(long timeout, TimeUnit unit)
// 遞減鎖存器的計數，若是計數到達零，則釋放全部等待的線程。
void countDown()
// 返回當前計數。
long getCount()
// 返回標識此鎖存器及其狀態的字符串。
String toString()

CountDownLatch數據結構

CountDownLatch的UML類圖以下：app

CountDownLatch的數據結構很簡單，它是經過"共享鎖"實現的。它包含了sync對象，sync是Sync類型。Sync是實例類，它繼承於AQS。框架

CountDownLatch源碼分析(基於JDK1.7.0_40)

CountDownLatch完整源碼(基於JDK1.7.0_40)less

  1 /*
  2  * ORACLE PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
  3  *
  4  *
  5  *
  6  *
  7  *
  8  *
  9  *
 10  *
 11  *
 12  *
 13  *
 14  *
 15  *
 16  *
 17  *
 18  *
 19  *
 20  *
 21  *
 22  *
 23  */
 24 
 25 /*
 26  *
 27  *
 28  *
 29  *
 30  *
 31  * Written by Doug Lea with assistance from members of JCP JSR-166
 32  * Expert Group and released to the public domain, as explained at
 33  * http://creativecommons.org/publicdomain/zero/1.0/
 34  */
 35 
 36 package java.util.concurrent;
 37 import java.util.concurrent.locks.*;
 38 import java.util.concurrent.atomic.*;
 39 
 40 /**
 41  * A synchronization aid that allows one or more threads to wait until
 42  * a set of operations being performed in other threads completes.
 43  *
 44  * <p>A {@code CountDownLatch} is initialized with a given <em>count</em>.
 45  * The {@link #await await} methods block until the current count reaches
 46  * zero due to invocations of the {@link #countDown} method, after which
 47  * all waiting threads are released and any subsequent invocations of
 48  * {@link #await await} return immediately.  This is a one-shot phenomenon
 49  * -- the count cannot be reset.  If you need a version that resets the
 50  * count, consider using a {@link CyclicBarrier}.
 51  *
 52  * <p>A {@code CountDownLatch} is a versatile synchronization tool
 53  * and can be used for a number of purposes.  A
 54  * {@code CountDownLatch} initialized with a count of one serves as a
 55  * simple on/off latch, or gate: all threads invoking {@link #await await}
 56  * wait at the gate until it is opened by a thread invoking {@link
 57  * #countDown}.  A {@code CountDownLatch} initialized to <em>N</em>
 58  * can be used to make one thread wait until <em>N</em> threads have
 59  * completed some action, or some action has been completed N times.
 60  *
 61  * <p>A useful property of a {@code CountDownLatch} is that it
 62  * doesn't require that threads calling {@code countDown} wait for
 63  * the count to reach zero before proceeding, it simply prevents any
 64  * thread from proceeding past an {@link #await await} until all
 65  * threads could pass.
 66  *
 67  * <p><b>Sample usage:</b> Here is a pair of classes in which a group
 68  * of worker threads use two countdown latches:
 69  * <ul>
 70  * <li>The first is a start signal that prevents any worker from proceeding
 71  * until the driver is ready for them to proceed;
 72  * <li>The second is a completion signal that allows the driver to wait
 73  * until all workers have completed.
 74  * </ul>
 75  *
 76  * <pre>
 77  * class Driver { // ...
 78  *   void main() throws InterruptedException {
 79  *     CountDownLatch startSignal = new CountDownLatch(1);
 80  *     CountDownLatch doneSignal = new CountDownLatch(N);
 81  *
 82  *     for (int i = 0; i < N; ++i) // create and start threads
 83  *       new Thread(new Worker(startSignal, doneSignal)).start();
 84  *
 85  *     doSomethingElse();            // don't let run yet
 86  *     startSignal.countDown();      // let all threads proceed
 87  *     doSomethingElse();
 88  *     doneSignal.await();           // wait for all to finish
 89  *   }
 90  * }
 91  *
 92  * class Worker implements Runnable {
 93  *   private final CountDownLatch startSignal;
 94  *   private final CountDownLatch doneSignal;
 95  *   Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
 96  *      this.startSignal = startSignal;
 97  *      this.doneSignal = doneSignal;
 98  *   }
 99  *   public void run() {
100  *      try {
101  *        startSignal.await();
102  *        doWork();
103  *        doneSignal.countDown();
104  *      } catch (InterruptedException ex) {} // return;
105  *   }
106  *
107  *   void doWork() { ... }
108  * }
109  *
110  * </pre>
111  *
112  * <p>Another typical usage would be to divide a problem into N parts,
113  * describe each part with a Runnable that executes that portion and
114  * counts down on the latch, and queue all the Runnables to an
115  * Executor.  When all sub-parts are complete, the coordinating thread
116  * will be able to pass through await. (When threads must repeatedly
117  * count down in this way, instead use a {@link CyclicBarrier}.)
118  *
119  * <pre>
120  * class Driver2 { // ...
121  *   void main() throws InterruptedException {
122  *     CountDownLatch doneSignal = new CountDownLatch(N);
123  *     Executor e = ...
124  *
125  *     for (int i = 0; i < N; ++i) // create and start threads
126  *       e.execute(new WorkerRunnable(doneSignal, i));
127  *
128  *     doneSignal.await();           // wait for all to finish
129  *   }
130  * }
131  *
132  * class WorkerRunnable implements Runnable {
133  *   private final CountDownLatch doneSignal;
134  *   private final int i;
135  *   WorkerRunnable(CountDownLatch doneSignal, int i) {
136  *      this.doneSignal = doneSignal;
137  *      this.i = i;
138  *   }
139  *   public void run() {
140  *      try {
141  *        doWork(i);
142  *        doneSignal.countDown();
143  *      } catch (InterruptedException ex) {} // return;
144  *   }
145  *
146  *   void doWork() { ... }
147  * }
148  *
149  * </pre>
150  *
151  * <p>Memory consistency effects: Until the count reaches
152  * zero, actions in a thread prior to calling
153  * {@code countDown()}
154  * <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>
155  * actions following a successful return from a corresponding
156  * {@code await()} in another thread.
157  *
158  * @since 1.5
159  * @author Doug Lea
160  */
161 public class CountDownLatch {
162     /**
163      * Synchronization control For CountDownLatch.
164      * Uses AQS state to represent count.
165      */
166     private static final class Sync extends AbstractQueuedSynchronizer {
167         private static final long serialVersionUID = 4982264981922014374L;
168 
169         Sync(int count) {
170             setState(count);
171         }
172 
173         int getCount() {
174             return getState();
175         }
176 
177         protected int tryAcquireShared(int acquires) {
178             return (getState() == 0) ? 1 : -1;
179         }
180 
181         protected boolean tryReleaseShared(int releases) {
182             // Decrement count; signal when transition to zero
183             for (;;) {
184                 int c = getState();
185                 if (c == 0)
186                     return false;
187                 int nextc = c-1;
188                 if (compareAndSetState(c, nextc))
189                     return nextc == 0;
190             }
191         }
192     }
193 
194     private final Sync sync;
195 
196     /**
197      * Constructs a {@code CountDownLatch} initialized with the given count.
198      *
199      * @param count the number of times {@link #countDown} must be invoked
200      *        before threads can pass through {@link #await}
201      * @throws IllegalArgumentException if {@code count} is negative
202      */
203     public CountDownLatch(int count) {
204         if (count < 0) throw new IllegalArgumentException("count < 0");
205         this.sync = new Sync(count);
206     }
207 
208     /**
209      * Causes the current thread to wait until the latch has counted down to
210      * zero, unless the thread is {@linkplain Thread#interrupt interrupted}.
211      *
212      * <p>If the current count is zero then this method returns immediately.
213      *
214      * <p>If the current count is greater than zero then the current
215      * thread becomes disabled for thread scheduling purposes and lies
216      * dormant until one of two things happen:
217      * <ul>
218      * <li>The count reaches zero due to invocations of the
219      * {@link #countDown} method; or
220      * <li>Some other thread {@linkplain Thread#interrupt interrupts}
221      * the current thread.
222      * </ul>
223      *
224      * <p>If the current thread:
225      * <ul>
226      * <li>has its interrupted status set on entry to this method; or
227      * <li>is {@linkplain Thread#interrupt interrupted} while waiting,
228      * </ul>
229      * then {@link InterruptedException} is thrown and the current thread's
230      * interrupted status is cleared.
231      *
232      * @throws InterruptedException if the current thread is interrupted
233      *         while waiting
234      */
235     public void await() throws InterruptedException {
236         sync.acquireSharedInterruptibly(1);
237     }
238 
239     /**
240      * Causes the current thread to wait until the latch has counted down to
241      * zero, unless the thread is {@linkplain Thread#interrupt interrupted},
242      * or the specified waiting time elapses.
243      *
244      * <p>If the current count is zero then this method returns immediately
245      * with the value {@code true}.
246      *
247      * <p>If the current count is greater than zero then the current
248      * thread becomes disabled for thread scheduling purposes and lies
249      * dormant until one of three things happen:
250      * <ul>
251      * <li>The count reaches zero due to invocations of the
252      * {@link #countDown} method; or
253      * <li>Some other thread {@linkplain Thread#interrupt interrupts}
254      * the current thread; or
255      * <li>The specified waiting time elapses.
256      * </ul>
257      *
258      * <p>If the count reaches zero then the method returns with the
259      * value {@code true}.
260      *
261      * <p>If the current thread:
262      * <ul>
263      * <li>has its interrupted status set on entry to this method; or
264      * <li>is {@linkplain Thread#interrupt interrupted} while waiting,
265      * </ul>
266      * then {@link InterruptedException} is thrown and the current thread's
267      * interrupted status is cleared.
268      *
269      * <p>If the specified waiting time elapses then the value {@code false}
270      * is returned.  If the time is less than or equal to zero, the method
271      * will not wait at all.
272      *
273      * @param timeout the maximum time to wait
274      * @param unit the time unit of the {@code timeout} argument
275      * @return {@code true} if the count reached zero and {@code false}
276      *         if the waiting time elapsed before the count reached zero
277      * @throws InterruptedException if the current thread is interrupted
278      *         while waiting
279      */
280     public boolean await(long timeout, TimeUnit unit)
281         throws InterruptedException {
282         return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
283     }
284 
285     /**
286      * Decrements the count of the latch, releasing all waiting threads if
287      * the count reaches zero.
288      *
289      * <p>If the current count is greater than zero then it is decremented.
290      * If the new count is zero then all waiting threads are re-enabled for
291      * thread scheduling purposes.
292      *
293      * <p>If the current count equals zero then nothing happens.
294      */
295     public void countDown() {
296         sync.releaseShared(1);
297     }
298 
299     /**
300      * Returns the current count.
301      *
302      * <p>This method is typically used for debugging and testing purposes.
303      *
304      * @return the current count
305      */
306     public long getCount() {
307         return sync.getCount();
308     }
309 
310     /**
311      * Returns a string identifying this latch, as well as its state.
312      * The state, in brackets, includes the String {@code "Count ="}
313      * followed by the current count.
314      *
315      * @return a string identifying this latch, as well as its state
316      */
317     public String toString() {
318         return super.toString() + "[Count = " + sync.getCount() + "]";
319     }
320 }

View Code

CountDownLatch是經過「共享鎖」實現的。下面，咱們分析CountDownLatch中3個核心函數: CountDownLatch(int count), await(), countDown()。dom

1. CountDownLatch(int count)ide

public CountDownLatch(int count) {
    if (count < 0) throw new IllegalArgumentException("count < 0");
    this.sync = new Sync(count);
}

說明：該函數是建立一個Sync對象，而Sync是繼承於AQS類。Sync構造函數以下：

Sync(int count) {
    setState(count);
}

setState()在AQS中實現，源碼以下：

protected final void setState(long newState) {
    state = newState;
}

說明：在AQS中，state是一個private volatile long類型的對象。對於CountDownLatch而言，state表示的」鎖計數器「。CountDownLatch中的getCount()最終是調用AQS中的getState()，返回的state對象，即」鎖計數器「。

2. await()

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

說明：該函數其實是調用的AQS的acquireSharedInterruptibly(1);

AQS中的acquireSharedInterruptibly()的源碼以下：

public final void acquireSharedInterruptibly(long arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

說明：acquireSharedInterruptibly()的做用是獲取共享鎖。
若是當前線程是中斷狀態，則拋出異常InterruptedException。不然，調用tryAcquireShared(arg)嘗試獲取共享鎖；嘗試成功則返回，不然就調用doAcquireSharedInterruptibly()。doAcquireSharedInterruptibly()會使當前線程一直等待，直到當前線程獲取到共享鎖(或被中斷)才返回。

tryAcquireShared()在CountDownLatch.java中被重寫，它的源碼以下：

protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1;
}

說明：tryAcquireShared()的做用是嘗試獲取共享鎖。
若是"鎖計數器=0"，即鎖是可獲取狀態，則返回1；不然，鎖是不可獲取狀態，則返回-1。

private void doAcquireSharedInterruptibly(long arg)
    throws InterruptedException {
    // 建立"當前線程"的Node節點，且Node中記錄的鎖是"共享鎖"類型；並將該節點添加到CLH隊列末尾。
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            // 獲取上一個節點。
            // 若是上一節點是CLH隊列的表頭，則"嘗試獲取共享鎖"。
            final Node p = node.predecessor();
            if (p == head) {
                long r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            // (上一節點不是CLH隊列的表頭) 當前線程一直等待，直到獲取到共享鎖。
            // 若是線程在等待過程當中被中斷過，則再次中斷該線程(還原以前的中斷狀態)。
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

說明：
(01) addWaiter(Node.SHARED)的做用是，建立」當前線程「的Node節點，且Node中記錄的鎖的類型是」共享鎖「(Node.SHARED)；並將該節點添加到CLH隊列末尾。關於Node和CLH在"Java多線程系列--「JUC鎖」03之公平鎖(一)"已經詳細介紹過，這裏就再也不重複說明了。
(02) node.predecessor()的做用是，獲取上一個節點。若是上一節點是CLH隊列的表頭，則」嘗試獲取共享鎖「。
(03) shouldParkAfterFailedAcquire()的做用和它的名稱同樣，若是在嘗試獲取鎖失敗以後，線程應該等待，則返回true；不然，返回false。
(04) 當shouldParkAfterFailedAcquire()返回ture時，則調用parkAndCheckInterrupt()，當前線程會進入等待狀態，直到獲取到共享鎖才繼續運行。
doAcquireSharedInterruptibly()中的shouldParkAfterFailedAcquire(), parkAndCheckInterrupt等函數在"Java多線程系列--「JUC鎖」03之公平鎖(一)"中介紹過，這裏也就再也不詳細說明了。

3. countDown()

public void countDown() {
    sync.releaseShared(1);
}

說明：該函數實際上調用releaseShared(1)釋放共享鎖。

releaseShared()在AQS中實現，源碼以下：

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

說明：releaseShared()的目的是讓當前線程釋放它所持有的共享鎖。
它首先會經過tryReleaseShared()去嘗試釋放共享鎖。嘗試成功，則直接返回；嘗試失敗，則經過doReleaseShared()去釋放共享鎖。

tryReleaseShared()在CountDownLatch.java中被重寫，源碼以下：

protected boolean tryReleaseShared(int releases) {
    // Decrement count; signal when transition to zero
    for (;;) {
        // 獲取「鎖計數器」的狀態
        int c = getState();
        if (c == 0)
            return false;
        // 「鎖計數器」-1
        int nextc = c-1;
        // 經過CAS函數進行賦值。
        if (compareAndSetState(c, nextc))
            return nextc == 0;
    }
}

說明：tryReleaseShared()的做用是釋放共享鎖，將「鎖計數器」的值-1。

總結：CountDownLatch是經過「共享鎖」實現的。在建立CountDownLatch中時，會傳遞一個int類型參數count，該參數是「鎖計數器」的初始狀態，表示該「共享鎖」最多能被count給線程同時獲取。當某線程調用該CountDownLatch對象的await()方法時，該線程會等待「共享鎖」可用時，才能獲取「共享鎖」進而繼續運行。而「共享鎖」可用的條件，就是「鎖計數器」的值爲0！而「鎖計數器」的初始值爲count，每當一個線程調用該CountDownLatch對象的countDown()方法時，纔將「鎖計數器」-1；經過這種方式，必須有count個線程調用countDown()以後，「鎖計數器」才爲0，而前面提到的等待線程才能繼續運行！

以上，就是CountDownLatch的實現原理。

CountDownLatch的使用示例

下面經過CountDownLatch實現："主線程"等待"5個子線程"所有都完成"指定的工做(休眠1000ms)"以後，再繼續運行。

 1 import java.util.concurrent.CountDownLatch;
 2 import java.util.concurrent.CyclicBarrier;
 3 
 4 public class CountDownLatchTest1 {
 5 
 6     private static int LATCH_SIZE = 5;
 7     private static CountDownLatch doneSignal;
 8     public static void main(String[] args) {
 9 
10         try {
11             doneSignal = new CountDownLatch(LATCH_SIZE);
12 
13             // 新建5個任務
14             for(int i=0; i<LATCH_SIZE; i++)
15                 new InnerThread().start();
16 
17             System.out.println("main await begin.");
18             // "主線程"等待線程池中5個任務的完成
19             doneSignal.await();
20 
21             System.out.println("main await finished.");
22         } catch (InterruptedException e) {
23             e.printStackTrace();
24         }
25     }
26 
27     static class InnerThread extends Thread{
28         public void run() {
29             try {
30                 Thread.sleep(1000);
31                 System.out.println(Thread.currentThread().getName() + " sleep 1000ms.");
32                 // 將CountDownLatch的數值減1
33                 doneSignal.countDown();
34             } catch (InterruptedException e) {
35                 e.printStackTrace();
36             }
37         }
38     }
39 }

運行結果：

main await begin.
Thread-0 sleep 1000ms.
Thread-2 sleep 1000ms.
Thread-1 sleep 1000ms.
Thread-4 sleep 1000ms.
Thread-3 sleep 1000ms.
main await finished.

結果說明：主線程經過doneSignal.await()等待其它線程將doneSignal遞減至0。其它的5個InnerThread線程，每個都經過doneSignal.countDown()將doneSignal的值減1；當doneSignal爲0時，main被喚醒後繼續執行。

更多內容

1. Java多線程系列--「JUC鎖」01之框架

2. Java多線程系列--「JUC鎖」02之互斥鎖ReentrantLock

3. Java多線程系列--「JUC鎖」03之公平鎖(一)

4. Java多線程系列--「JUC鎖」04之公平鎖(二)

5. Java多線程系列--「JUC鎖」05之非公平鎖

6. Java多線程系列--「JUC鎖」06之 Condition條件

7. Java多線程系列--「JUC鎖」07之 LockSupport

8. Java多線程系列--「JUC鎖」08之共享鎖和ReentrantReadWriteLock

9. Java多線程系列目錄(共xx篇)