Java并发(7):CountDownLatch、CyclicBarrier、Callable、Future

  CountDownLatch、CyclicBarrier、Callable、Future  都位于java.util.concurrent包下,其中CountDownLatch和CyclicBarrier属于该包中的tools分支,Callable和Future属于该包中的executer分支。

一.CountDownLatch

  CountDownLatch类位于java.util.concurrent包下,利用它可以实现类似计数器的功能。比如有一个任务A,它要等待其他4个任务执行完毕之后才能执行,此时就可以利用CountDownLatch来实现这种功能了。

CountDownLatch类只提供了一个构造器:

1 public CountDownLatch(int count) {  };  //参数count为计数值

然后下面这3个方法是CountDownLatch类中最重要的方法:

1 public void await() throws InterruptedException { };   //调用await()方法的线程会被挂起,它会等待直到count值为0才继续执行
2 public boolean await(long timeout, TimeUnit unit) throws InterruptedException { };  //和await()类似,只不过等待一定的时间后count值还没变为0的话就会继续执行
3 public void countDown() { };  //将count值减1

例子:假如有Thread1、Thread2、Thread3、Thread4四条线程分别统计C、D、E、F四个盘的大小,所有线程都统计完毕交给Thread5线程去做汇总,实现代码如下:

 1 public class CountDownLatchTest {
 2     private static CountDownLatch count = new CountDownLatch(4);
 3     private static ExecutorService service = Executors.newFixedThreadPool(6);
 4     public static void main(String args[]) throws InterruptedException {
 5         for (int i = 0; i < 4; i++) {
 6             service.execute(() -> {
 7                 // 模拟任务耗时
 8                 try {
 9                     int timer = new Random().nextInt(5);
10                     TimeUnit.SECONDS.sleep(timer);
11                     System.out.printf("%s时完成磁盘的统计任务,耗费%d秒.\n", new Date().toString(), timer);
12                     // 任务完成之后,计数器减一
13                     count.countDown();
14                 } catch (InterruptedException e) {
15                     e.printStackTrace();
16                 }
17             });
18         }
19         // 主线程一直被阻塞,知道count的计数器被设置为0
20         count.await();
21         System.out.printf("%s时全部任务都完成,执行合并计算.\n", new Date().toString());
22         service.shutdown();
23     }
24 }

  当问题已经分解为许多部分,每个线程都被分配一部分计算时,CountdownLatch 非常有用。在工作线程结束时,它们将减少计数,协调线程可以在闩锁处等待当前这一批计算结束,然后继续移至下一批计算。
  相反地,具有计数 1 的 CountdownLatch 类可以用作"启动大门",来立即启动一组线程;工作线程可以在闩锁处等待,协调线程减少计数,从而立即释放所有工作线程。下例使用两个 CountdownLatche。一个作为启动大门,一个在所有工作线程结束时释放线程。

二.CyclicBarrier用法

  字面意思回环栅栏,通过它可以实现让一组线程等待至某个状态之后再全部同时执行。叫做回环是因为当所有等待线程都被释放以后,CyclicBarrier可以被重用。我们暂且把这个状态就叫做barrier,当调用await()方法之后,线程就处于barrier了。

  CyclicBarrier类位于java.util.concurrent包下,CyclicBarrier提供2个构造器:

1 public CyclicBarrier(int parties, Runnable barrierAction) {
2
3 }
4
5 public CyclicBarrier(int parties) {
6
7 }

  参数parties指让多少个线程或者任务等待至barrier状态;参数barrierAction为当这些线程都达到barrier状态时会执行的内容。

  然后CyclicBarrier中最重要的方法就是await方法,它有2个重载版本:

1 public int await() throws InterruptedException, BrokenBarrierException { };
2 public int await(long timeout, TimeUnit unit)throws InterruptedException,BrokenBarrierException,TimeoutException { };

  第一个版本比较常用,用来挂起当前线程,直至所有线程都到达barrier状态再同时执行后续任务;

  第二个版本是让这些线程等待至一定的时间,如果还有线程没有到达barrier状态就直接让到达barrier的线程执行后续任务。

  例子:假若有若干个线程都要进行写数据操作,并且只有所有线程都完成写数据操作之后,这些线程才能继续做后面的事情,实现代码如下:

 1 public class Test {
 2     public static void main(String[] args) {
 3         int N = 4;
 4         CyclicBarrier barrier  = new CyclicBarrier(N);
 5         for(int i=0;i<N;i++)
 6             new Writer(barrier).start();
 7     }
 8     static class Writer extends Thread{
 9         private CyclicBarrier cyclicBarrier;
10         public Writer(CyclicBarrier cyclicBarrier) {
11             this.cyclicBarrier = cyclicBarrier;
12         }
13
14         @Override
15         public void run() {
16             System.out.println("线程"+Thread.currentThread().getName()+"正在写入数据...");
17             try {
18                 Thread.sleep(5000);      //以睡眠来模拟写入数据操作
19                 System.out.println("线程"+Thread.currentThread().getName()+"写入数据完毕,等待其他线程写入完毕");
20                 cyclicBarrier.await();
21             } catch (InterruptedException e) {
22                 e.printStackTrace();
23             }catch(BrokenBarrierException e){
24                 e.printStackTrace();
25             }
26             System.out.println("所有线程写入完毕,继续处理其他任务...");
27         }
28     }
29 }

执行结果:

线程Thread-0正在写入数据...
线程Thread-3正在写入数据...
线程Thread-2正在写入数据...
线程Thread-1正在写入数据...
线程Thread-2写入数据完毕,等待其他线程写入完毕
线程Thread-0写入数据完毕,等待其他线程写入完毕
线程Thread-3写入数据完毕,等待其他线程写入完毕
线程Thread-1写入数据完毕,等待其他线程写入完毕
所有线程写入完毕,继续处理其他任务...
所有线程写入完毕,继续处理其他任务...
所有线程写入完毕,继续处理其他任务...
所有线程写入完毕,继续处理其他任务...

  从上面输出结果可以看出,每个写入线程执行完写数据操作之后,就在等待其他线程写入操作完毕。当所有线程线程写入操作完毕之后,所有线程就继续进行后续的操作了。更多的其他相关例子详见http://www.cnblogs.com/dolphin0520/p/3920397.html

  CountdownLatch 类与 CyclicBarrier 相似,因为它的角色是对已经在它们中间分摊了问题的一组线程进行协调。它也是使用整型变量构造的,指明计数的初始值,但是与 CyclicBarrier 不同的是,CountdownLatch 不能重新使用。其中,CyclicBarrier 是到达屏障的所有线程的大门,只有当所有线程都已经到达屏障或屏障被打破时,才允许这些线程通过,CountdownLatch 将到达和等待功能分离。任何线程都可以通过调用 countDown() 减少当前计数,这种不会阻塞线程,而只是减少计数。await() 方法的行为与 CyclicBarrier.await() 稍微有所不同,调用 await() 任何线程都会被阻塞,直到闩锁计数减少为零,在该点等待的所有线程才被释放,对 await() 的后续调用将立即返回。

  CountDownLatch和CyclicBarrier都能够实现线程之间的等待,只不过它们侧重点不同:CountDownLatch一般用于某个线程A等待若干个其他线程执行完任务之后,它才执行;而CyclicBarrier一般用于一组线程互相等待至某个状态,然后这一组线程再同时执行;另外,CountDownLatch是不能够重用的,而CyclicBarrier是可以重用的。

三. Callable

  先说一下java.lang.Runnable吧,它是一个接口,在它里面只声明了一个run()方法:

1 public interface Runnable {
2     public abstract void run();
3 }

  由于run()方法返回值为void类型,所以在执行完任务之后无法返回任何结果。

  Callable位于java.util.concurrent包下,它也是一个接口,在它里面也只声明了一个方法,只不过这个方法叫做call():

1 public interface Callable<V> {
2     V call() throws Exception;
3 }

  可以看到,这是一个泛型接口,call()函数返回的类型就是传递进来的V类型。

  Callable的使用一般情况下是配合ExecutorService来使用的,在ExecutorService接口中声明了若干个submit方法的重载版本:

1 <T> Future<T> submit(Callable<T> task);
2 <T> Future<T> submit(Runnable task, T result);
3 Future<?> submit(Runnable task);

  第一个submit方法里面的参数类型就是Callable。暂时只需要知道Callable一般是和ExecutorService配合来使用的,具体的使用方法讲在后面讲述。

  一般情况下我们使用第一个submit方法和第三个submit方法,第二个submit方法很少使用。

四. Future

4.1 Future

  Future就是对于具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果。必要时可以通过get方法获取执行结果,该方法会阻塞直到任务返回结果。

  Future类位于java.util.concurrent包下,它是一个接口:

1 public interface Future<V> {
2     boolean cancel(boolean mayInterruptIfRunning);
3     boolean isCancelled();
4     boolean isDone();
5     V get() throws InterruptedException, ExecutionException;
6     V get(long timeout, TimeUnit unit)
7         throws InterruptedException, ExecutionException, TimeoutException;
8 }

在Future接口中声明了5个方法,下面依次解释每个方法的作用:

  • cancel方法用来取消任务,如果取消任务成功则返回true,如果取消任务失败则返回false。参数mayInterruptIfRunning表示是否允许取消正在执行却没有执行完毕的任务,如果设置true,则表示可以取消正在执行过程中的任务。如果任务已经完成,则无论mayInterruptIfRunning为true还是false,此方法肯定返回false,即如果取消已经完成的任务会返回false;如果任务正在执行,若mayInterruptIfRunning设置为true,则返回true,若mayInterruptIfRunning设置为false,则返回false;如果任务还没有执行,则无论mayInterruptIfRunning为true还是false,肯定返回true。
  • isCancelled方法表示任务是否被取消成功,如果在任务正常完成前被取消成功,则返回 true。
  • isDone方法表示任务是否已经完成,若任务完成,则返回true;
  • get()方法用来获取执行结果,这个方法会产生阻塞,会一直等到任务执行完毕才返回;
  • get(long timeout, TimeUnit unit)用来获取执行结果,如果在指定时间内,还没获取到结果,就直接返回null。

也就是说Future提供了三种功能:

  1. 判断任务是否完成;
  2. 能够中断任务;
  3. 能够获取任务执行结果。

4.2 FutureTask

  FutureTask是Future接口的一个唯一实现类。

先来看一下FutureTask的实现:

1 public class FutureTask<V> implements RunnableFuture<V>

FutureTask类实现了RunnableFuture接口,我们看一下RunnableFuture接口的实现:

1 public interface RunnableFuture<V> extends Runnable, Future<V> {
2     void run();
3 }

  可以看出RunnableFuture继承了Runnable接口和Future接口,而FutureTask实现了RunnableFuture接口。所以它既可以作为Runnable被线程执行,又可以作为Future得到Callable的返回值。

FutureTask提供了2个构造器:

1 public FutureTask(Callable<V> callable) {
2 }
3 public FutureTask(Runnable runnable, V result) {
4 }

  Callable+FutureTask的使用示例如下:

 1 public class Test {
 2     public static void main(String[] args) {
 3         //第一种方式
 4         ExecutorService executor = Executors.newCachedThreadPool();
 5         Task task = new Task();
 6         FutureTask<Integer> futureTask = new FutureTask<Integer>(task);
 7         executor.submit(futureTask);
 8         executor.shutdown();
 9
10         //第二种方式,注意这种方式和第一种方式效果是类似的,只不过一个使用的是ExecutorService,一个使用的是Thread
11         /*Task task = new Task();
12         FutureTask<Integer> futureTask = new FutureTask<Integer>(task);
13         Thread thread = new Thread(futureTask);
14         thread.start();*/
15
16         try {
17             Thread.sleep(1000);
18         } catch (InterruptedException e1) {
19             e1.printStackTrace();
20         }
21         System.out.println("主线程在执行任务");
22         try {
23             System.out.println("task运行结果"+futureTask.get());
24         } catch (InterruptedException e) {
25             e.printStackTrace();
26         } catch (ExecutionException e) {
27             e.printStackTrace();
28         }
29         System.out.println("所有任务执行完毕");
30     }
31 }
32 class Task implements Callable<Integer>{
33     @Override
34     public Integer call() throws Exception {
35         System.out.println("子线程在进行计算");
36         Thread.sleep(3000);
37         int sum = 0;
38         for(int i=0;i<100;i++)
39             sum += i;
40         return sum;
41     }
42 }

参考:http://www.cnblogs.com/dolphin0520/p/3949310.html

时间: 2025-01-01 20:48:49

Java并发(7):CountDownLatch、CyclicBarrier、Callable、Future的相关文章

Java 并发工具CountDownLatch和CyclicBarrier 原理解析

[TOC] Java 并发工具CountDownLatch和CyclicBarrier 原理解析 一,简介 CountDownLatch 允许一个或者多个线程等待其他线程完成操作. CyclicBarrier 的字面意思是可循环使用(Cyclic)的屏障(Barrier).它要做的事情是,让一组线程达到一个屏障(也可以叫同步点)时被阻塞,直到最后一个线程到达屏障时,屏障才会开门,所有被屏障拦截的线程才会继续运行. 二,代码演示 CountDownLatchDemo public class Co

Java并发工具类CyclicBarrier

CyclicBarrier同步屏障 java并发工具类中有一个叫做CyclicBarrier的类,与CountDownLatch类似,都可以实现线程间的同步,但是差别是CyclicBarrier是可重置的同步屏障. 想象一个场景,有N个人不同时间走到一扇门,因为门需要N个人合力才能推开,所以人不足N个时,只能阻塞在此,等到N个人都到了之后,可以推开门,继续进行之前的工作.CyclicBarrier就是这扇门. 看看下面的代码,定义了一个线程数为2的,CyclicBarrier,并在主线程和另外一

JAVA并发,CountDownLatch使用

该文章转自:http://www.itzhai.com/the-introduction-and-use-of-a-countdownlatch.html CountDownLatch 1.类介绍 一个同步辅助类,在完成一组正在其他线程中执行的操作之前,它允许一个或多个线程一直等待.用给定的计数 初始化 CountDownLatch.由于调用了 countDown() 方法,所以在当前计数到达零之前,await 方法会一直受阻塞.之后,会释放所有等待的线程,await 的所有后续调用都将立即返回

Java并发编程实践:Callable异步回调Future、FutureTask用法

Callable接口类似于Runnable,从名字就可以看出来了,但是Runnable不会返回结果,并且无法抛出返回结果的异常,而Callable功能更强大一些,被线程执行后,可以返回值,这个返回值可以被Future拿到.FutureTask实现了两个接口,Runnable和Future,所以它既可以作为Runnable被线程执行,又可以作为Future得到Callable的返回值,那么这个组合的使用有什么好处呢?假设有一个很耗时的返回值需要计算,并且这个返回值不是立刻需要的话,那么就可以使用这

Java 并发专题 : CyclicBarrier 打造一个安全的门禁系统

继续并发专题~ 这次介绍CyclicBarrier:看一眼API的凝视: /** * A synchronization aid that allows a set of threads to all wait for * each other to reach a common barrier point. CyclicBarriers are * useful in programs involving a fixed sized party of threads that * must o

java并发编程之CyclicBarrier

一个同步辅助类,它允许一组线程互相等待,直到到达某个公共屏障点 (common barrier point).在涉及一组固定大小的线程的程序中,这些线程必须不时地互相等待,此时 CyclicBarrier 很有用.因为该 barrier 在释放等待线程后可以重用,所以称它为循环 的 barrier. CyclicBarrier 支持一个可选的 Runnable 命令,在一组线程中的最后一个线程到达之后(但在释放所有线程之前),该命令只在每个屏障点运行一次.若在继续所有参与线程之前更新共享状态,此

java并发编程之三--CyclicBarrier的使用

CyclicBarrier 允许一组线程全部等待彼此达到共同屏障点的同步辅助. 循环阻塞在涉及固定大小的线程方的程序中很有用,这些线程必须偶尔等待彼此. 屏障被称为循环 ,因为它可以在等待的线程被释放之后重新使用. A CyclicBarrier支持一个可选的Runnable命令,每个屏障点运行一次,在派对中的最后一个线程到达之后,但在任何线程释放之前. 在任何一方继续进行之前,此屏障操作对更新共享状态很有用. 实现原理:在CyclicBarrier的内部定义了一个Lock对象,每当一个线程调用

Java并发编程--CountDownLatch

概述 CountDownLatch是一个同步工具类,它允许一个或多个线程一直等待,直到其他线程的操作执行完后再执行. CountDownLatch使用一个计数器count实现,构建CountDownLatch时需要使用给定的count初始化CountDownLatch.在count到达0之前,调用await()方法的线程将一直阻塞,当count到达0时,会唤醒所有阻塞的线程.注意:计数器count无法被重置,即只能实现一次这种功能,这也是CountDownLatch与CyclicBarrier的

Java并发——线程间的等待与通知

前言: 前面讲完了一些并发编程的原理,现在我们要来学习的是线程之间的协作.通俗来说就是,当前线程在某个条件下需要等待,不需要使用太多系统资源.在某个条件下我们需要去唤醒它,分配给它一定的系统资源,让它继续工作.这样能更好的节约资源. 一.Object的wait()与notify() 基本概念: 一个线程因执行目标动作的条件未能满足而被要求暂停就是wait,而一个线程满足执行目标动作的条件之后唤醒被暂停的线程就是notify. 基本模板: synchronized (obj){ //保护条件不成立

Java并发编程:Callable、Future和FutureTask(转)

Java并发编程:Callable.Future和FutureTask 在前面的文章中我们讲述了创建线程的2种方式,一种是直接继承Thread,另外一种就是实现Runnable接口. 这2种方式都有一个缺陷就是:在执行完任务之后无法获取执行结果. 如果需要获取执行结果,就必须通过共享变量或者使用线程通信的方式来达到效果,这样使用起来就比较麻烦. 而自从Java 1.5开始,就提供了Callable和Future,通过它们可以在任务执行完毕之后得到任务执行结果. 今天我们就来讨论一下Callabl