文档章节

ExecutorCompletionService的源码解读

写代码的奥特曼
 写代码的奥特曼
发布于 2017/05/30 12:16
字数 661
阅读 29
收藏 0

ExecutorCompletionService分析及使用

 

当我们通过Executor提交一组并发执行的任务,并且希望在每一个任务完成后能立即得到结果,有两种方式可以采取:

 

方式一:

通过一个list来保存一组future,然后在循环中轮训这组future,直到每个future都已完成。如果我们不希望出现因为排在前面的任务阻塞导致后面先完成的任务的结果没有及时获取的情况,那么在调用get方式时,需要将超时时间设置为0 

Java代码 

public class CompletionServiceTest {

	static class Task implements Callable<String>{
		private int i;
		
		public Task(int i){
			this.i = i;
		}

		@Override
		public String call() throws Exception {
			Thread.sleep(10000);
			return Thread.currentThread().getName() + "执行完任务:" + i;
		}	
	}
	
	public static void main(String[] args){
		testUseFuture();
	}
	
	private static void testUseFuture(){
		int numThread = 5;
		ExecutorService executor = Executors.newFixedThreadPool(numThread);
		List<Future<String>> futureList = new ArrayList<Future<String>>();
		for(int i = 0;i<numThread;i++ ){
			Future<String> future = executor.submit(new CompletionServiceTest.Task(i));
			futureList.add(future);
		}
				
		while(numThread > 0){
			for(Future<String> future : futureList){
				String result = null;
				try {
					result = future.get(0, TimeUnit.SECONDS);
				} catch (InterruptedException e) {
					e.printStackTrace();
				} catch (ExecutionException e) {
					e.printStackTrace();
				} catch (TimeoutException e) {
					//超时异常直接忽略
				}
				if(null != result){
					futureList.remove(future);
					numThread--;
					System.out.println(result);
					//此处必须break,否则会抛出并发修改异常。(也可以通过将futureList声明为CopyOnWriteArrayList类型解决)
					break;
				}
			}
		}
	}
}

 方式二:

第一种方式显得比较繁琐,通过使用ExecutorCompletionService,则可以达到代码最简化的效果。

Java代码 

public class CompletionServiceTest {

	static class Task implements Callable<String>{
		private int i;
		
		public Task(int i){
			this.i = i;
		}

		@Override
		public String call() throws Exception {
			Thread.sleep(10000);
			return Thread.currentThread().getName() + "执行完任务:" + i;
		}	
	}
	
	public static void main(String[] args) throws InterruptedException, ExecutionException{
		testExecutorCompletionService();
	}
	
	private static void testExecutorCompletionService() throws InterruptedException, ExecutionException{
		int numThread = 5;
		ExecutorService executor = Executors.newFixedThreadPool(numThread);
		CompletionService<String> completionService = new ExecutorCompletionService<String>(executor);
		for(int i = 0;i<numThread;i++ ){
			completionService.submit(new CompletionServiceTest.Task(i));
		}
}
		
		for(int i = 0;i<numThread;i++ ){		
			System.out.println(completionService.take().get());
		}
		
	}

ExecutorCompletionService分析:

 CompletionService是Executor和BlockingQueue的结合体。

Java代码 

public ExecutorCompletionService(Executor executor) {
        if (executor == null)
            throw new NullPointerException();
        this.executor = executor;
        this.aes = (executor instanceof AbstractExecutorService) ?
            (AbstractExecutorService) executor : null;
        this.completionQueue = new LinkedBlockingQueue<Future<V>>();
    }

 任务的提交和执行都是委托给Executor来完成。当提交某个任务时,该任务首先将被包装为一个QueueingFuture,

Java代码 

public Future<V> submit(Callable<V> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<V> f = newTaskFor(task);
        executor.execute(new QueueingFuture(f));
        return f;
    }

 QueueingFuture是FutureTask的一个子类,通过改写该子类的done方法,可以实现当任务完成时,将结果放入到BlockingQueue中。

 

private class QueueingFuture extends FutureTask<Void> {
        QueueingFuture(RunnableFuture<V> task) {
            super(task, null);
            this.task = task;
        }
        protected void done() { completionQueue.add(task); }
        private final Future<V> task;
    }

 而通过使用BlockingQueue的take或poll方法,则可以得到结果。在BlockingQueue不存在元素时,这两个操作会阻塞,一旦有结果加入,则立即返回。

Java代码 

    public Future<V> take() throws InterruptedException {
        return completionQueue.take();
    }

    public Future<V> poll() {
        return completionQueue.poll();
    }

 

© 著作权归作者所有

共有 人打赏支持
写代码的奥特曼
粉丝 12
博文 87
码字总数 109550
作品 0
杭州
高级程序员
私信 提问
Java多线程6 CompletionService

CompletionService 1 CompletionService介绍 CompletionService用于提交一组Callable任务,其take方法返回已完成的一个Callable任务对应的Future对象。 如果你向Executor提交了一个批处理任务...

香沙小熊
2018/11/27
0
0
java.util.concurrent.ExecutorCompletionService 源码

线程池相关 源码: 类 ExecutorCompletionService<V> 所有已实现的接口: CompletionService<V> 使用提供的 来执行任务的 。 此类将安排那些完成时提交的任务,把它们放置在可使用 take 访问...

狼王黄师傅
2018/11/29
0
0
Java使用Executor执行Callable任务时的几种方法

多线程在需要返回值时,我们知道需要用到Callable和Future。Callable的cell方法可以返回一个值并且可抛出异常,是对Runnable的很好的补充;Future表示了一个任务的周期,它提供了判断任务状态...

东都大狼狗
2018/09/09
0
0
从静态代码扫描引擎PMD源码学习-多线程任务模型和File过滤设计

不知不觉在工作中研究PMD并定制规则已经4个月左右了。其实PMD有许多值得我学习的源码,不过出于时间并不曾动笔。今天简单记录总结一下PMD的多线程和File过滤设计的源码。 1 public class Mul...

phinehasz
2018/07/21
0
0
java - concurrent 之 CompletionService

CompletionService /** A service that decouples the production of new asynchronous tasks from the consumption of the results of completed tasks. Producers {@code submit} tasks fo......

yuzn
2016/08/17
9
0

没有更多内容

加载失败,请刷新页面

加载更多

iOS个人中心渐变动画、微信对话框、标签选择器、自定义导航栏、短信验证输入框等源码

iOS精选源码 简单的个人中心页面-自定义导航栏并予以渐变动画 程序员取悦女票的正确姿势---Tip1(iOS美容篇) iOS 前台重启应用和清除角标的问题 微信原生提醒对话框3.0 JHLikeButton - 有趣...

Android爱开源
16分钟前
1
0
Yii2使用驼峰命名的形式访问控制器

yii2在使用的时候,访问控制器的时候,如果控制器的名称是驼峰命名法,那访问的url中要改成横线的形式。例如: public function actionRoomUpdate(){//}//访问的时候就要www.test.co...

dragon_tech
19分钟前
1
0
Navicat使用教程:使用Navicat Query Analyzer优化查询性能(第2部分)

下载Navicat Monitor最新版本 Navicat Monitor 是一套安全、简单而且无代理的远程服务器监控工具。它具有强大的功能使你的监控发挥最大效用。受监控的服务器包括 MySQL、MariaDB 和 Percona ...

电池盒
25分钟前
1
0
Python3 读写utf-8文本文件

with open('testRead.txt', 'r', encoding='utf-8') as f: for each_line in f: Passwith open('testWrite.txt', 'w', encoding='utf-8') as f: f.write('写入的内容'......

编程老陆
28分钟前
1
0
Linux syslog相关函数详解

介绍 syslog是Unix系统的日志系统。可以将日志记录在本地系统中。 一个完整的syslong日志包含如下信息:程序模块 | 严重性 | 时间 | 主机名 | 进程名 | 进程ID | 正文。 syslong相关函数 1....

RongJinhui0
33分钟前
0
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部