- 浏览: 460179 次
- 性别:
- 来自: 杭州
文章分类
最新评论
-
ty1972873004:
sunwang810812 写道我运行了这个例子,怎么结果是这 ...
Java并发编程: 使用Semaphore限制资源并发访问的线程数 -
lgh1992314:
simpleDean 写道请问,Logger.setLevel ...
Java内置Logger详解 -
sunwang810812:
我运行了这个例子,怎么结果是这样的:2号车泊车6号车泊车5号车 ...
Java并发编程: 使用Semaphore限制资源并发访问的线程数 -
jp260715007:
nanjiwubing123 写道参考你的用法,用如下方式实现 ...
面试题--三个线程循环打印ABC10次的几种解决方法 -
cb_0312:
SurnameDictionary文章我没看完,现在懂了
中文排序
本文将介绍CountDownLatch工具类,并采用这个工具类给出一个实例。
1. CountDownLatch工具类介绍
CountDownLatch是一个同步工具类,它允许一个或多个线程处于等待状态直到在其它线程中运行的一组操作完成为止。CountDownLatch用一个给定的计数来实现初始化。Await方法会一直处于阻塞状态,直到countDown方法调用而使当前计数达到零。当计数为零之后,所有处于等待的线程将被释放,await的任何后续调用将立即返回。这种现象只出现一次,计数是不能被重置的。如果你需要一个可以重置计数的版本,需要考虑使用CyclicBarrie.
上面的介绍来自于CountDownLatch类的注释。
/** * A synchronization aid that allows one or more threads to wait until * a set of operations being performed in other threads completes. * * <p>A {@code CountDownLatch} is initialized with a given [i]count[/i]. * The {@link #await await} methods block until the current count reaches * zero due to invocations of the {@link #countDown} method, after which * all waiting threads are released and any subsequent invocations of * {@link #await await} return immediately. This is a one-shot phenomenon * -- the count cannot be reset. If you need a version that resets the * count, consider using a {@link CyclicBarrier}. * */
CountDownLatch中定义了一个内部类Sync,该类继承AbstractQueuedSynchronizer。从代码中可以看出,CountDownLatch的await,countDown以及getCount方法都调用了Sync的方法。CountDownLatch工具类相关的类图以及详细代码如下:
/* * @(#)CountDownLatch.java 1.5 04/02/09 * * Copyright 2004 Sun Microsystems, Inc. All rights reserved. * SUN PROPRIETARY/CONFIDENTIAL. Use is subject to license terms. */ package java.util.concurrent; import java.util.concurrent.locks.*; import java.util.concurrent.atomic.*; /** * A synchronization aid that allows one or more threads to wait until * a set of operations being performed in other threads completes. * * <p>A <tt>CountDownLatch</tt> is initialized with a given * [i]count[/i]. The {@link #await await} methods block until the current * {@link #getCount count} reaches zero due to invocations of the * {@link #countDown} method, after which all waiting threads are * released and any subsequent invocations of {@link #await await} return * immediately. This is a one-shot phenomenon -- the count cannot be * reset. If you need a version that resets the count, consider using * a {@link CyclicBarrier}. * * <p>A <tt>CountDownLatch</tt> is a versatile synchronization tool * and can be used for a number of purposes. A * <tt>CountDownLatch</tt> initialized with a count of one serves as a * simple on/off latch, or gate: all threads invoking {@link #await await} * wait at the gate until it is opened by a thread invoking {@link * #countDown}. A <tt>CountDownLatch</tt> initialized to [i]N[/i] * can be used to make one thread wait until [i]N[/i] threads have * completed some action, or some action has been completed N times. * <p>A useful property of a <tt>CountDownLatch</tt> is that it * doesn't require that threads calling <tt>countDown</tt> wait for * the count to reach zero before proceeding, it simply prevents any * thread from proceeding past an {@link #await await} until all * threads could pass. * * <p><b>Sample usage:</b> Here is a pair of classes in which a group * of worker threads use two countdown latches: * [list] * <li>The first is a start signal that prevents any worker from proceeding * until the driver is ready for them to proceed; * <li>The second is a completion signal that allows the driver to wait * until all workers have completed. * [/list] * * <pre> * class Driver { // ... * void main() throws InterruptedException { * CountDownLatch startSignal = new CountDownLatch(1); * CountDownLatch doneSignal = new CountDownLatch(N); * * for (int i = 0; i < N; ++i) // create and start threads * new Thread(new Worker(startSignal, doneSignal)).start(); * * doSomethingElse(); // don't let run yet * startSignal.countDown(); // let all threads proceed * doSomethingElse(); * doneSignal.await(); // wait for all to finish * } * } * * class Worker implements Runnable { * private final CountDownLatch startSignal; * private final CountDownLatch doneSignal; * Worker(CountDownLatch startSignal, CountDownLatch doneSignal) { * this.startSignal = startSignal; * this.doneSignal = doneSignal; * } * public void run() { * try { * startSignal.await(); * doWork(); * doneSignal.countDown(); * } catch (InterruptedException ex) {} // return; * } * * void doWork() { ... } * } * * </pre> * * <p>Another typical usage would be to divide a problem into N parts, * describe each part with a Runnable that executes that portion and * counts down on the latch, and queue all the Runnables to an * Executor. When all sub-parts are complete, the coordinating thread * will be able to pass through await. (When threads must repeatedly * count down in this way, instead use a {@link CyclicBarrier}.) * * <pre> * class Driver2 { // ... * void main() throws InterruptedException { * CountDownLatch doneSignal = new CountDownLatch(N); * Executor e = ... * * for (int i = 0; i < N; ++i) // create and start threads * e.execute(new WorkerRunnable(doneSignal, i)); * * doneSignal.await(); // wait for all to finish * } * } * * class WorkerRunnable implements Runnable { * private final CountDownLatch doneSignal; * private final int i; * WorkerRunnable(CountDownLatch doneSignal, int i) { * this.doneSignal = doneSignal; * this.i = i; * } * public void run() { * try { * doWork(i); * doneSignal.countDown(); * } catch (InterruptedException ex) {} // return; * } * * void doWork() { ... } * } * * </pre> * * @since 1.5 * @author Doug Lea */ public class CountDownLatch { /** * Synchronization control For CountDownLatch. * Uses AQS state to represent count. */ private static final class Sync extends AbstractQueuedSynchronizer { Sync(int count) { setState(count); } int getCount() { return getState(); } public int tryAcquireShared(int acquires) { return getState() == 0? 1 : -1; } public boolean tryReleaseShared(int releases) { // 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; } } } private final Sync sync; /** * Constructs a <tt>CountDownLatch</tt> initialized with the given * count. * * @param count the number of times {@link #countDown} must be invoked * before threads can pass through {@link #await}. * * @throws IllegalArgumentException if <tt>count</tt> is less than zero. */ public CountDownLatch(int count) { if (count < 0) throw new IllegalArgumentException("count < 0"); this.sync = new Sync(count); } /** * Causes the current thread to wait until the latch has counted down to * zero, unless the thread is {@link Thread#interrupt interrupted}. * * <p>If the current {@link #getCount count} is zero then this method * returns immediately. * <p>If the current {@link #getCount count} is greater than zero then * the current thread becomes disabled for thread scheduling * purposes and lies dormant until one of two things happen: * [list] * <li>The count reaches zero due to invocations of the * {@link #countDown} method; or * <li>Some other thread {@link Thread#interrupt interrupts} the current * thread. * [/list] * <p>If the current thread: * [list] * <li>has its interrupted status set on entry to this method; or * <li>is {@link Thread#interrupt interrupted} while waiting, * [/list] * then {@link InterruptedException} is thrown and the current thread's * interrupted status is cleared. * * @throws InterruptedException if the current thread is interrupted * while waiting. */ public void await() throws InterruptedException { sync.acquireSharedInterruptibly(1); } /** * Causes the current thread to wait until the latch has counted down to * zero, unless the thread is {@link Thread#interrupt interrupted}, * or the specified waiting time elapses. * * <p>If the current {@link #getCount count} is zero then this method * returns immediately with the value <tt>true</tt>. * * <p>If the current {@link #getCount count} is greater than zero then * the current thread becomes disabled for thread scheduling * purposes and lies dormant until one of three things happen: * [list] * <li>The count reaches zero due to invocations of the * {@link #countDown} method; or * <li>Some other thread {@link Thread#interrupt interrupts} the current * thread; or * <li>The specified waiting time elapses. * [/list] * <p>If the count reaches zero then the method returns with the * value <tt>true</tt>. * <p>If the current thread: * [list] * <li>has its interrupted status set on entry to this method; or * <li>is {@link Thread#interrupt interrupted} while waiting, * [/list] * then {@link InterruptedException} is thrown and the current thread's * interrupted status is cleared. * * <p>If the specified waiting time elapses then the value <tt>false</tt> * is returned. * If the time is * less than or equal to zero, the method will not wait at all. * * @param timeout the maximum time to wait * @param unit the time unit of the <tt>timeout</tt> argument. * @return <tt>true</tt> if the count reached zero and <tt>false</tt> * if the waiting time elapsed before the count reached zero. * * @throws InterruptedException if the current thread is interrupted * while waiting. */ public boolean await(long timeout, TimeUnit unit) throws InterruptedException { return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout)); } /** * Decrements the count of the latch, releasing all waiting threads if * the count reaches zero. * <p>If the current {@link #getCount count} is greater than zero then * it is decremented. If the new count is zero then all waiting threads * are re-enabled for thread scheduling purposes. * <p>If the current {@link #getCount count} equals zero then nothing * happens. */ public void countDown() { sync.releaseShared(1); } /** * Returns the current count. * <p>This method is typically used for debugging and testing purposes. * @return the current count. */ public long getCount() { return sync.getCount(); } /** * Returns a string identifying this latch, as well as its state. * The state, in brackets, includes the String * "Count =" followed by the current count. * @return a string identifying this latch, as well as its * state */ public String toString() { return super.toString() + "[Count = " + sync.getCount() + "]"; } }
2. CountDownLatch工具类的使用案例
CountDownLatch的作用是控制一个计数器,每个线程在运行完毕后执行countDown,表示自己运行结束,这对于多个子任务的计算特别有效,比如一个异步任务需要拆分成10个子任务执行,主任务必须知道子任务是否完成,所有子任务完成后才能进行合并计算,从而保证了一个主任务逻辑的正确性。(此段摘自于<<改善Java程序的151个建议>>, P254)
CountDownLatch最重要的方法是countDown()和await(),前者主要是倒数一次,后者是等待倒数到0,如果没有到达0,就只有阻塞等待了。
本实例主要使用CountDownLatch工具类来实现10个线程对1~100的求和,每个线程对10个数进行求和。第一个线程对1 – 10求和
第二个线程对 11 – 20求和
第三个线程对21 – 30 求和
…..
第十个线程对91 – 100求和。
具体的代码如下:
package my.concurrent.countdown; import java.util.concurrent.Callable; import java.util.concurrent.CountDownLatch; public class Calculator implements Callable<Integer> { //开始信号 private final CountDownLatch startSignal; //结束信号 private final CountDownLatch doneSignal; private int groupNumber = 0; /** * @param startSignal * @param endSignal * @param groupId */ public Calculator(CountDownLatch startSignal, CountDownLatch doneSignal, int groupNumber) { this.startSignal = startSignal; this.doneSignal = doneSignal; this.groupNumber = groupNumber; } public Integer call() throws Exception { startSignal.await(); Integer result = sum(groupNumber); printCompleteInfor(groupNumber,result); doneSignal.countDown(); return result; } private Integer sum(int groupNumber) { if (groupNumber < 1) { throw new IllegalArgumentException(); } int sum = 0; int start = (groupNumber - 1) * 10 + 1; int end = groupNumber * 10; for (int i = start; i <= end; i++) { sum += i; } return sum; } private void printCompleteInfor(int groupNumber, int sum) { System.out.println(String.format("Group %d is finished, the sum in this gropu is %d", groupNumber, sum)); } }
package my.concurrent.countdown; import java.util.ArrayList; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; public class CountDownLatchTest { public static void main(String[] args) throws Exception { /** * 1-100求和,分10个线程来计算,每个线程对10个数求和。 */ int numOfGroups = 10; CountDownLatch startSignal = new CountDownLatch(1); CountDownLatch doneSignal = new CountDownLatch(numOfGroups); ExecutorService service = Executors.newFixedThreadPool(numOfGroups); List<Future<Integer>> futures = new ArrayList<Future<Integer>>(); submit(futures, numOfGroups, service, startSignal, doneSignal); /** * 开始,让所有的求和计算线程运行 */ startSignal.countDown(); /** * 阻塞,知道所有计算线程完成计算 */ doneSignal.await(); shutdown(service); printResult(futures); } private static void submit(List<Future<Integer>> futures, int numOfGroups, ExecutorService service, CountDownLatch startSignal, CountDownLatch doneSignal) { for (int groupNumber = 1; groupNumber <= numOfGroups; groupNumber++) { futures.add(service.submit(new Calculator(startSignal, doneSignal, groupNumber))); } } private static int getResult(List<Future<Integer>> futures) throws InterruptedException, ExecutionException { int result = 0; for (Future<Integer> f : futures) { result += f.get(); } return result; } private static void printResult(List<Future<Integer>> futures) throws InterruptedException, ExecutionException { System.out.println("[1,100] Sum is :" + getResult(futures)); } private static void shutdown(ExecutorService service) { service.shutdown(); } }
一次的执行结果如下:
Group 8 is finished, the sum in this gropu is 755
Group 2 is finished, the sum in this gropu is 155
Group 10 is finished, the sum in this gropu is 955
Group 5 is finished, the sum in this gropu is 455
Group 7 is finished, the sum in this gropu is 655
Group 3 is finished, the sum in this gropu is 255
Group 9 is finished, the sum in this gropu is 855
Group 1 is finished, the sum in this gropu is 55
Group 4 is finished, the sum in this gropu is 355
Group 6 is finished, the sum in this gropu is 555
[1,100] Sum is :5050
评论
8 楼
julydave
2013-08-05
MouseLearnJava 写道
freezingsky 写道
内容基本与《java并发编程实践》很相似。不过,在这文章里如果能增加CyclicBarrier的内容,并做出相应的比较,会更好。
谢谢你的建议。 我会在下一篇介绍CuclicBarrier内容的时候,做出一些比较的。
可参见此篇文章 http://www.molotang.com/articles/487.html
7 楼
julydave
2013-08-05
freezingsky 写道
内容基本与《java并发编程实践》很相似。不过,在这文章里如果能增加CyclicBarrier的内容,并做出相应的比较,会更好。
可以看看这篇 http://www.molotang.com/articles/487.html
6 楼
MouseLearnJava
2013-07-31
freezingsky 写道
内容基本与《java并发编程实践》很相似。不过,在这文章里如果能增加CyclicBarrier的内容,并做出相应的比较,会更好。
谢谢你的建议。 我会在下一篇介绍CuclicBarrier内容的时候,做出一些比较的。
5 楼
freezingsky
2013-07-30
内容基本与《java并发编程实践》很相似。不过,在这文章里如果能增加CyclicBarrier的内容,并做出相应的比较,会更好。
4 楼
MouseLearnJava
2013-07-30
/** * 开始,让所有的求和计算线程运行 */ startSignal.countDown(); /** * 阻塞,直到所有计算线程完成计算 */ doneSignal.await(); shutdown(service); /** * 计算并打印出求和内容 */ printResult(futures);
doneSignal.await();是用于阻塞,直到所有子线程完成计算。
在这之后,主线程才去求和,本文的代码中可以看出,计算和打印结果放在最后了,这时候,所有的线程都有了计算结果。
3 楼
fisher123
2013-07-30
doneSignal.await(); 不也是阻塞线程直到所有线程执行完毕么?
2 楼
MouseLearnJava
2013-07-30
文中所述的方法是采用CountDownLatch工具类来做的,该例子中,1-100的求和分十个子线程来做,主线程是在所有线程计算好结果之后,才把所有子线程得到的结果累加起来,主线程求和过程中不会有线程等待的情况。
如果直接采用future.get()方法来做,比如一个for循环中采用sum+=future.get()去求和, 可能会有等待的现象,因为多线程下执行的顺序是不定的。比如:可能先执行2,4,5,再执行1,那么尝试用future.get()取第一个线程产生的结果的时候,第一个线程的结果可能还没有计算好,这样就会产生等待的情况了。
Future接口的get方法如下:
如果直接采用future.get()方法来做,比如一个for循环中采用sum+=future.get()去求和, 可能会有等待的现象,因为多线程下执行的顺序是不定的。比如:可能先执行2,4,5,再执行1,那么尝试用future.get()取第一个线程产生的结果的时候,第一个线程的结果可能还没有计算好,这样就会产生等待的情况了。
Future接口的get方法如下:
/** * Waits if necessary for the computation to complete, and then * retrieves its result. * * @return the computed result * @throws CancellationException if the computation was cancelled * @throws ExecutionException if the computation threw an * exception * @throws InterruptedException if the current thread was interrupted * while waiting */ V get() throws InterruptedException, ExecutionException; /** * Waits if necessary for at most the given time for the computation * to complete, and then retrieves its result, if available. * * @param timeout the maximum time to wait * @param unit the time unit of the timeout argument * @return the computed result * @throws CancellationException if the computation was cancelled * @throws ExecutionException if the computation threw an * exception * @throws InterruptedException if the current thread was interrupted * while waiting * @throws TimeoutException if the wait timed out */ V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
1 楼
fisher123
2013-07-30
请问一下LZ,文中所述方法和直接调用future.get(),有何不同之处呢?求教
发表评论
-
Java编程练手100题
2014-12-11 17:13 6670本文给出100道Java编程练手的程序。 列表如下: 面 ... -
The Code Sample 代码实例 个人博客开通
2014-09-04 18:48 1373个人博客小站开通 http://thecodesample. ... -
三道多线程编程题
2013-10-14 14:30 5839本文记录三道多线程编程题。 第一道: 有四个线程1、2、3 ... -
面试题 -- 一个线程打印 1~52,另一个线程打印字母A-Z。打印顺序为12A34B56C……5152Z。
2013-10-01 13:03 4956题目:一个线程打印 1~52,另一个线程打印字母A-Z。打印顺 ... -
面试题--三个线程循环打印ABC10次的几种解决方法
2013-09-29 20:01 10191最近发现公司有份笔试试卷中有道多线程的题目:有三个线程分别打印 ... -
Java并发编程-Condition使用实例
2013-09-27 21:35 4823本文将简单介绍用于线程协作的Condition, 并给出一个例 ... -
Java并发编程: 使用Exchanger实现线程间的数据交换
2013-08-08 12:13 3724本文介绍Exchanger工具类, 然后采用Exchanger ... -
Java并发编程: 使用Semaphore限制资源并发访问的线程数
2013-08-07 16:23 4669本文将介绍用来控制资源同时访问个数的Semaphore工具类, ... -
Java并发编程: Volatile不能保证数据同步
2013-08-05 12:47 5955在本篇博文中,将给出一 ... -
Java并发编程: CyclicBarrier让多线程齐步走
2013-08-01 11:12 4499以前在<<编写高质量代码-改善Java程序的151 ... -
Java并发编程-Executor框架+实例
2013-07-18 21:23 3788Executor框架是指java 5中 ...
相关推荐
龙果 java并发编程原理实战 第2节理解多线程与并发的之间的联系与区别 [免费观看] 00:11:59分钟 | 第3节解析多线程与多进程的联系以及上下文切换所导致资源浪费问题 [免费观看] 00:13:03分钟 | 第4节学习并发的四...
Java并发编程基础主要包括以下几个核心方面: 线程与线程状态:理解Java中线程的基本概念,包括线程的创建、启动、暂停、恢复和终止。熟悉线程的生命周期及其不同状态,如新建、就绪、运行、阻塞和死亡。 线程同步...
Java并发编程一CountDownLatch、CyclicBarrier、Semaphore初使用 CountDownLatch、CyclicBarrier、Semaphore这些线程协作工具类是基于AQS的,看完这篇博客后可以去看下面这篇博客,了解它们是如何实现的。 Java并发...
多种创建线程的方式案例演示(二)使用线程池.mp4 Spring对并发的支持:Spring的异步任务.mp4 使用jdk8提供的lambda进行并行计算.mp4 了解多线程所带来的安全风险.mp4 从线程的优先级看饥饿问题.mp4 从Java字节码的...
1、本资源包含并发编程基础知识的使用案例,包括:线程创建、Synchronized和Reentrantlock锁的使用、线程安全问题演示、Condition的应用、CountDownLatch的应用、Cyclicbarrier的应用、Semaphore的应用、线程池的...
7、并发工具类CountDownLatch 、CyclicBarrier和Semaphore底层实现原理 8、线程池原理和如何使用线程池 9、ThreadLocal 为什么会内存泄漏 10、Volatile底层实现原理 11、AQS源码分析 12、CAS原理分析和使用场景 13、...
第2节理解多线程与并发的之间的联系与区别 [免费观看] 00:11:59分钟 | 第3节解析多线程与多进程的联系以及上下文切换所导致资源浪费问题 [免费观看] 00:13:03分钟 | 第4节学习并发的四个阶段并推荐学习并发的资料 ...
java并发编程原理实战 第2节理解多线程与并发的之间的联系与区别 [免费观看] 00:11:59分钟 | 第3节解析多线程与多进程的联系以及上下文切换所导致资源浪费问题 [免费观看] 00:13:03分钟 | 第4节学习并发的四个...
Java并发编程实战 本书深入浅出地介绍了Java线程和并发,是一本完美的Java并发参考手册。书中从并发性和线程安全性的基本概念出发,介绍了如何使用类库提供的基本并发构建块,用于避免并发危险、构造线程安全的类及...
│ Java并发编程.png │ ppt+源码.rar │ 高并发编程第二阶段01讲、课程大纲及主要内容介绍.wmv │ 高并发编程第二阶段02讲、介绍四种Singleton方式的优缺点在多线程情况下.wmv │ 高并发编程第二阶段03讲、...
为了方便编写出线程安全的程序,Java里面提供了一些线程安全类和并发工具,比如:同步容器、并发容器、阻塞队列、Synchronizer(比如CountDownLatch)。我们来讨论下同步容器。 一.为什么会出现同步容器? 在...
Java并发编程 背景介绍 并发历史 必要性 进程 资源分配的最小单位 线程 CPU调度的最小单位 线程的优势 (1)如果设计正确,多线程程序可以通过提高处理器资源的利用率来提升系统吞吐率 ...
第2节理解多线程与并发的之间的联系与区别 [免费观看] 00:11:59分钟 | 第3节解析多线程与多进程的联系以及上下文切换所导致资源浪费问题 [免费观看] 00:13:03分钟 | 第4节学习并发的四个阶段并推荐学习并发的资料 ...
之所以把这一章节叫做AQS简介而不是叫AQS详解,是因为已经有大神写过详解的文章Java并发之AQS详解,这篇文章对AQS的源码解析很透彻,博主读了之后受益匪浅,鉴于对原作者的尊重,所以如上附上原文的链接。...
【2018最新最详细】并发多线程教程,课程结构如下 ...25.大白话说java并发工具类-CountDownLatch,CyclicBarrier 26.大白话说java并发工具类-Semaphore,Exchanger 27.一篇文章,让你彻底弄懂生产者--消费者问题
1、使用线程的经验:设置名称、响应中断、使用ThreadLocal 2、Executor :ExecutorService和Future ☆☆☆ 3、阻塞队列: put和take、offer和poll、drainTo 4、线程间的协调手段:lock、condition、wait、notify、...
JUC(Java Util Concurrent)是Java中用于并发编程的工具包,提供了一组接口和类,用于处理多线程和并发操作。JUC提供了一些常用的并发编程模式和工具,如线程池、并发集合、原子操作等。 JUC的主要特点包括: ...