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

迈不过友情╰ 2022-04-13 10:18 358阅读 0赞

\[b\]\[color=olive\]\[size=large\]大多数时候，我们都可以采取多线程+线程池的方式，来优化我们程序的处理效率，JAVA在JDK1.5后的并发包，提供了很多方便快捷的并发工具辅助类，来简化并发编程。  
  
今天散仙，简单的描述下怎么使用CompletionService和Future来快捷的处理多个并行的任务。  
  
需求如下：现在某个盘符的目录下有10个txt文件，每行一个单词，让你利用多线程来快速，去重和并发去重后的结果，写入一个总的txt文件。  
（注意，去重有2部分，先是单个txt里面的去重，然后是合并后的去重）  
  
如果的这个需求，利用线程池来处理，是非常方便的。  
  
下面散仙的这两个例子，是模拟如下需求来测试的。  
代码如下：  
\[/size\]\[/color\]\[/b\]

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;
    		}
    	}
    
    
    
    
    
    
    
    
    
    
    
    }

\[b\]\[color=green\]\[size=large\]执行testFutureTask方法的运行输出：\[/size\]\[/color\]\[/b\]

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]

\[b\]\[color=green\]\[size=large\]执行testCompletionService方法的输出如下：\[/size\]\[/color\]\[/b\]

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]

\[b\]\[color=green\]\[size=large\]观察上面的两个结果，我们就会发现，Future类的先执行完的任务，并不一定是先输出的，而在CompletionService里面，我们可以发现，最先完成的任务，一定是最先输出的，这一点需要注意一下，CompletionService提供了，异步分离任务的机制，将生产新的异步任务与使用已完成任务的结果分离开来的服务。生产者 submit 执行的任务。使用者 take 已完成的任务，并按照完成这些任务的顺序处理它们的结果。例如，CompletionService 可以用来管理异步 IO ，执行读操作的任务作为程序或系统的一部分提交，然后，当完成读操作时，会在程序的不同部分执行其他操作，执行操作的顺序可能与所请求的顺序不同。  
  
大部分情况下，这两种方式，我们都可以任意采用，如果，我们需要先完成的任务先输出，那么我们应该优先选择CompletionService 类，如果不在意任务的先后顺序，仅仅是看到任务的结果，那么我们即可以使用Future，也可以使用CompletionService。  
\[/size\]\[/color\]\[/b\]