文档章节

ExecutorCompletionService的源码解读

写代码的奥特曼
 写代码的奥特曼
发布于 2017/05/30 12:16
字数 661
阅读 28
收藏 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();
    }

 

© 著作权归作者所有

共有 人打赏支持
写代码的奥特曼
粉丝 11
博文 86
码字总数 109060
作品 0
杭州
高级程序员
从静态代码扫描引擎PMD源码学习-多线程任务模型和File过滤设计

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

phinehasz
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
源码|批量执行invokeAll()&&多选一invokeAny()

ExecutorService中定义了两个批量执行任务的方法,invokeAll()和invokeAny(),在批量执行或多选一的业务场景中非常方便。invokeAll()在所有任务都完成(包括成功/被中断/超时)后才会返回,i...

猴子007
2017/12/04
0
0
源码 | 批量执行invokeAll() && 多选一invokeAny()

原文出处:猴子007 ExecutorService中定义了两个批量执行任务的方法,invokeAll()和invokeAny(),在批量执行或多选一的业务场景中非常方便。invokeAll()在所有任务都完成(包括成功/被中断/...

猴子007
2017/12/02
0
0
Android6.0源码解读之Activity点击事件分发机制

本篇博文是Android点击事件分发机制系列博文的第四篇,主要是从解读Activity类的源码入手,根据源码理清Activity点击事件分发原理,并掌握Activity点击事件分法机制。特别声明的是,本源码解...

mynameishuangshuai
2016/10/23
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

kernel version does not match DSO version

错误信息: kernel version 384.11 does not match DSO version 384.130.0 原因是: cuda driver版本太低,不匹配DSO 简单有效的修复方法,升级nvidia driver, 步骤如下: 1. google seach ...

刘小米
今天
0
0
maven坐标和依赖

一、maven坐标详解 <groupId>com.fgt.club</groupId><artifactId>club-common-service-facade</artifactId><version>3.0.0</version><packaging>jar</packaging> maven的坐标元素说......

老韭菜
今天
1
0
springmvc-servlet.xml配置表功能解释

问:<?xml version="1.0" encoding="UTF-8" ?> 答: xml version="1.0"表示是此xml文件的版本是1.0 encoding="UTF-8"表示此文件的编码方式是UTF-8 问:<!DOCTYPE beans PUBLIC "-//SPRING//......

隐士族隐逸
今天
1
0
基于TP5的微信的公众号获取登录用户信息

之前讲过微信的公众号自动登录的菜单配置,这次记录一下在TP5项目中获取自动登录的用户信息并存到数据库的操作 基本的流程为:微信设置自动登录的菜单—>访问的URL指定的函数里获取用户信息—...

月夜中徘徊
今天
0
0
youTrack

package jetbrains.teamsys.license.runtime; 计算lis package jetbrains.ring.license.reader; 验证lis 安装后先不要生成lis,要把相关文件进行替换 ring-license-checker-1.0.41.jar char......

max佩恩
今天
2
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部