Java多线程系列–“JUC锁”09之 CountDownLatch原理和示例

  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 }