首页 > 编程语言 > 详细

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

时间:2016-08-12 06:38:16      阅读:290      评论:0      收藏:0      [点我收藏+]

  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写入数据完毕,等待其他线程写入完毕
所有线程写入完毕,继续处理其他任务...
所有线程写入完毕,继续处理其他任务...
所有线程写入完毕,继续处理其他任务...
所有线程写入完毕,继续处理其他任务...
View Code

  从上面输出结果可以看出,每个写入线程执行完写数据操作之后,就在等待其他线程写入操作完毕。当所有线程线程写入操作完毕之后,所有线程就继续进行后续的操作了。更多的其他相关例子详见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

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

原文:http://www.cnblogs.com/Eason-S/p/5763418.html

(0)
(0)
   
举报
评论 一句话评论(0
关于我们 - 联系我们 - 留言反馈 - 联系我们:wmxa8@hotmail.com
© 2014 bubuko.com 版权所有
打开技术之扣,分享程序人生!