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

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

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

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

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

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

<pre name="code" class="java">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&lt;List&lt;String&gt;&gt; cs=new ExecutorCompletionService&lt;List&lt;String&gt;&gt;(es);
 
  
   /**
    * 异步进行，并获取结果
    *
    * **/
   public void testCompletionService()throws Exception{
  

   for(int i=0;i&lt;3;i++){
  
  cs.submit(new MyTask());
//  Future&lt;List&lt;String&gt;&gt; f=  cs.submit(new MyTask());
   //在此获取结果是如果第一个任务没有执行完，则其他的任务一直阻塞
//    if(f.get()!=null){
//    System.out.println("完成了....: "+f.get());
//    }
   }
  
   List&lt;Future&lt;List&lt;String&gt;&gt;&gt; list=new ArrayList&lt;Future&lt;List&lt;String&gt;&gt;&gt;();
  
   for(int i=0;i&lt;3;i++){
   list.add(cs.take());
   }
  
   for(Future&lt;List&lt;String&gt;&gt; l:list){
System.out.println("有任务执行完毕： "+l.get());
}
//   
   es.shutdown();
   }
public  void testFutureTask()throws Exception{
List&lt;Future&lt;List&lt;String&gt;&gt;&gt; list=new ArrayList&lt;Future&lt;List&lt;String&gt;&gt;&gt;();
for(int i=0;i&lt;3;i++){
list.add(es.submit(new MyTask()));
//Future&lt;List&lt;String&gt;&gt; f=es.submit(new MyTask());
// es.submit(new MyTask());
// if(f.get()!=null){
// System.out.println("有任务完成:  "+f.get());
// }
}  
for(Future&lt;List&lt;String&gt;&gt; l:list){
System.out.println("有任务执行完毕： "+l.get());
}
//
es.shutdown();

}






/**
* 实现Callable接口，带任务返回值的线程类
*
*
* ***/
private class MyTask implements Callable&lt;List&lt;String&gt;&gt;{
private Random r=new Random();
@Override
public List&lt;String&gt; call() throws Exception {
int count=0;
String name=Thread.currentThread().getName();
System.out.println(Thread.currentThread().getName()+"  开启启动任务......");
List&lt;String&gt; list=new ArrayList&lt;String&gt;();

for(int i=0;i&lt;5;i++){
int data=r.nextInt(5);
//System.out.println(name+"    ==&gt;　"+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;
}
}











}
</pre>
执行testFutureTask方法的运行输出：
<pre name="code" class="java">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]
</pre>
执行testCompletionService方法的输出如下：
<pre name="code" class="java">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]
</pre>

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

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