ExecutorService与CompletionService

标签：

我们现在在Java中使用多线程通常不会直接用Thread对象了，而是会用到java.util.concurrent包下的ExecutorService类来初始化一个线程池供我们使用。

之前我一直习惯自己维护一个list保存submit的callable task所返回的Future对象。

在主线程中遍历这个list并调用Future的get()方法取到Task的返回值。

但是，我在很多地方会看到一些代码通过CompletionService包装ExecutorService，然后调用其take()方法去取Future对象。以前没研究过这两者之间的区别。

这两者最主要的区别在于CompletionService的submit的task不一定是按照加入自己维护的list顺序完成的。

ExecutorService：从list中遍历的每个Future对象并不一定处于完成状态，这时调用get()方法就会被阻塞住，如果系统是设计成每个线程完成后就能根据其结果继续做后面的事，这样对于处于list后面的但是先完成的线程就会增加了额外的等待时间。

   而CompletionService的实现是维护一个保存Future对象的BlockingQueue。只有当这个Future对象状态是结束的时候，才会加入到这个Queue中，take()方法其实就是Producer-Consumer中的Consumer。它会从Queue中取出Future对象，如果Queue是空的，就会阻塞在那里，直到有完成的Future对象加入到Queue中。

所以，先完成的必定先被取出。这样就减少了不必要的等待时间

public class TestCompletionService {

	public static void test() throws InterruptedException, ExecutionException{
		ExecutorService pool =Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()*5);
		List<Future<Integer>> resultList = new ArrayList<Future<Integer>>();
		
		 for(int j=0;j<10;j++){
			 final int i=j;
			 Future<Integer> f= (Future) pool.submit(new Callable<Integer>() {
					@Override
					public Integer call() throws Exception {
						  Thread.sleep(200);
						  if(i==5){
							  Thread.sleep(2000);
						  }
						     return i;
					}
				});
			resultList.add(f);
		 }
		 
		 
		 for(int i=0;i<resultList.size();i++){
			System.out.println(resultList.get(i).get());
		 }
	}
	
	public static void test2() throws InterruptedException, ExecutionException{
		ExecutorService pool =Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()*5);
		CompletionService cs =new ExecutorCompletionService(pool);
		List<Future<Integer>> resultList = new ArrayList<Future<Integer>>();
		
		 for(int j=0;j<10;j++){
			 final int i=j;
			 Future<Integer> f= (Future) cs.submit(new Callable<Integer>() {
					@Override
					public Integer call() throws Exception {
						  Thread.sleep(200);
						  if(i==5){
							  Thread.sleep(2000);
						  }
						     return i;
					}
				});
			resultList.add(f);
		 }
		 
		 for(int i=0;i<resultList.size();i++){
			System.out.println(cs.take().get());
		 }
	}
	
	public static void main(String[] args) {
		try {
			test2();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (ExecutionException e) {
			e.printStackTrace();
		}
	}
	
}

1

2

3

4

等到一段时间

5

6

7

8

9

执行test2时会

3
9
7
4
0
6
2
1
8
5

ExecutorService与CompletionService

标签：

原文地址：http://blog.csdn.net/liyantianmin/article/details/42808089