JAVA中如何使用线程池来管理并行任务

大多数时候，我们都可以采取多线程+线程池的方式，来优化我们程序的处理效率，JAVA在JDK1.5后的并发包，提供了很多方便快捷的并发工具辅助类，来简化并发编程。

今天散仙，简单的描述下怎么使用CompletionService和Future来快捷的处理多个并行的任务。

需求如下：现在某个盘符的目录下有10个txt文件，每行一个单词，让你利用多线程来快速，去重和并发去重后的结果，写入一个总的txt文件。
（注意，去重有2部分，先是单个txt里面的去重，然后是合并后的去重）

如果的这个需求，利用线程池来处理，是非常方便的。

下面散仙的这两个例子，是模拟如下需求来测试的。
代码如下：

package com.dic.merge;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * 测试线程池提交任务
 * @author qindongliang
 * 
 * 
 * 
 * **/
public class TestFutureTask {

	
	
	public static void main(String[] args) throws Exception{
		
		TestFutureTask tft=new TestFutureTask();
		tft.testFutureTask();//测试FutrueTask
	//	tft.testCompletionService();//测试CompletionService
	}
	
	
	
	   /*
	    * 线程池
	    * 
	    * **/
	   ExecutorService es=Executors.newFixedThreadPool(5);
	
	   /*
	    * 
	    * 管理异步提交的task
	    * 
	    * 
	    * **/
	   private   CompletionService<List<String>> cs=new ExecutorCompletionService<List<String>>(es);
	  
	   
	   /**
	    * 异步进行，并获取结果
	    * 
	    * **/
	   public void testCompletionService()throws Exception{
		   
		
		   for(int i=0;i<3;i++){
			   
			  cs.submit(new MyTask());
			 //  Future<List<String>> f=  cs.submit(new MyTask());
			   //在此获取结果是如果第一个任务没有执行完，则其他的任务一直阻塞
//			   if(f.get()!=null){
//				   System.out.println("完成了....: "+f.get());
//			   }
		   }
		   
		   List<Future<List<String>>> list=new ArrayList<Future<List<String>>>();
		   
		   for(int i=0;i<3;i++){
			   list.add(cs.take());
		   }
		   
		   for(Future<List<String>> l:list){
				 System.out.println("有任务执行完毕： "+l.get());
			 }
//		   
		   es.shutdown();
	   }
	public  void testFutureTask()throws Exception{
		List<Future<List<String>>> list=new ArrayList<Future<List<String>>>();
		for(int i=0;i<3;i++){
			 list.add(es.submit(new MyTask()));
			//Future<List<String>> f=es.submit(new MyTask());
			// es.submit(new MyTask());
//			if(f.get()!=null){
//				System.out.println("有任务完成:  "+f.get());
//			}
		}   
		 for(Future<List<String>> l:list){
			 System.out.println("有任务执行完毕： "+l.get());
		 }
//		
		es.shutdown();
		
	}
	
	
	
	
	
	
	/**
	 * 实现Callable接口，带任务返回值的线程类
	 * 
	 * 
	 * ***/
	private class MyTask implements Callable<List<String>>{
		private Random r=new Random();
		@Override
		public List<String> call() throws Exception {
			int count=0;
			 String name=Thread.currentThread().getName();
			System.out.println(Thread.currentThread().getName()+"  开启启动任务......");
				List<String> list=new ArrayList<String>();
				
				for(int i=0;i<5;i++){
					int data=r.nextInt(5);
					//System.out.println(name+"    ==>　"+data);
					count+=data;
					Thread.sleep(data*1000);
					list.add(name+"  "+data+"");
			System.out.println(name+"  线程沉睡"+data+" 秒，放入List数据");
				}
			System.out.println(name+"任务完成,总共沉睡时间为: "+count+"s .");
			
			return list;
		}
	}
	
	
	
	
	
	
	
	
	
	
	
}

执行testFutureTask方法的运行输出：

pool-1-thread-2  开启启动任务......
pool-1-thread-3  开启启动任务......
pool-1-thread-1  开启启动任务......
pool-1-thread-3  线程沉睡1 秒，放入List数据
pool-1-thread-1  线程沉睡2 秒，放入List数据
pool-1-thread-3  线程沉睡2 秒，放入List数据
pool-1-thread-2  线程沉睡4 秒，放入List数据
pool-1-thread-3  线程沉睡1 秒，放入List数据
pool-1-thread-1  线程沉睡2 秒，放入List数据
pool-1-thread-3  线程沉睡3 秒，放入List数据
pool-1-thread-1  线程沉睡3 秒，放入List数据
pool-1-thread-2  线程沉睡3 秒，放入List数据
pool-1-thread-2  线程沉睡0 秒，放入List数据
pool-1-thread-1  线程沉睡1 秒，放入List数据
pool-1-thread-2  线程沉睡2 秒，放入List数据
pool-1-thread-2  线程沉睡0 秒，放入List数据
pool-1-thread-2任务完成,总共沉睡时间为: 9s .
pool-1-thread-3  线程沉睡3 秒，放入List数据
pool-1-thread-3任务完成,总共沉睡时间为: 10s .
pool-1-thread-1  线程沉睡3 秒，放入List数据
pool-1-thread-1任务完成,总共沉睡时间为: 11s .
有任务执行完毕： [pool-1-thread-1  2, pool-1-thread-1  2, pool-1-thread-1  3, pool-1-thread-1  1, pool-1-thread-1  3]
有任务执行完毕： [pool-1-thread-2  4, pool-1-thread-2  3, pool-1-thread-2  0, pool-1-thread-2  2, pool-1-thread-2  0]
有任务执行完毕： [pool-1-thread-3  1, pool-1-thread-3  2, pool-1-thread-3  1, pool-1-thread-3  3, pool-1-thread-3  3]

执行testCompletionService方法的输出如下：

pool-1-thread-1  开启启动任务......
pool-1-thread-3  开启启动任务......
pool-1-thread-2  开启启动任务......
pool-1-thread-3  线程沉睡2 秒，放入List数据
pool-1-thread-2  线程沉睡2 秒，放入List数据
pool-1-thread-1  线程沉睡4 秒，放入List数据
pool-1-thread-1  线程沉睡1 秒，放入List数据
pool-1-thread-3  线程沉睡3 秒，放入List数据
pool-1-thread-2  线程沉睡3 秒，放入List数据
pool-1-thread-1  线程沉睡2 秒，放入List数据
pool-1-thread-3  线程沉睡2 秒，放入List数据
pool-1-thread-2  线程沉睡3 秒，放入List数据
pool-1-thread-2  线程沉睡1 秒，放入List数据
pool-1-thread-1  线程沉睡3 秒，放入List数据
pool-1-thread-3  线程沉睡3 秒，放入List数据
pool-1-thread-2  线程沉睡2 秒，放入List数据
pool-1-thread-2任务完成,总共沉睡时间为: 11s .
pool-1-thread-3  线程沉睡2 秒，放入List数据
pool-1-thread-3任务完成,总共沉睡时间为: 12s .
pool-1-thread-1  线程沉睡2 秒，放入List数据
pool-1-thread-1任务完成,总共沉睡时间为: 12s .
有任务执行完毕： [pool-1-thread-2  2, pool-1-thread-2  3, pool-1-thread-2  3, pool-1-thread-2  1, pool-1-thread-2  2]
有任务执行完毕： [pool-1-thread-3  2, pool-1-thread-3  3, pool-1-thread-3  2, pool-1-thread-3  3, pool-1-thread-3  2]
有任务执行完毕： [pool-1-thread-1  4, pool-1-thread-1  1, pool-1-thread-1  2, pool-1-thread-1  3, pool-1-thread-1  2]

观察上面的两个结果，我们就会发现，Future类的先执行完的任务，并不一定是先输出的，而在CompletionService里面，我们可以发现，最先完成的任务，一定是最先输出的，这一点需要注意一下，CompletionService提供了，异步分离任务的机制，将生产新的异步任务与使用已完成任务的结果分离开来的服务。生产者 submit 执行的任务。使用者 take 已完成的任务，并按照完成这些任务的顺序处理它们的结果。例如，CompletionService 可以用来管理异步 IO ，执行读操作的任务作为程序或系统的一部分提交，然后，当完成读操作时，会在程序的不同部分执行其他操作，执行操作的顺序可能与所请求的顺序不同。

大部分情况下，这两种方式，我们都可以任意采用，如果，我们需要先完成的任务先输出，那么我们应该优先选择CompletionService 类，如果不在意任务的先后顺序，仅仅是看到任务的结果，那么我们即可以使用Future，也可以使用CompletionService。

评论

1 楼 JavaCFW 2014-05-10