Java并发专题 带返回结果的批量任务执行 CompletionService

    xiaoxiao2025-12-18  16

    转自 :http://blog.csdn.net/lmj623565791/article/details/27250059

    一般情况下,我们使用Runnable作为基本的任务表示形式,但是Runnable是一种有很大局限的抽象,run方法中只能记录日志,打印,或者把数据汇总入某个容器(一方面内存消耗大,另一方面需要控制同步,效率很大的限制),总之不能返回执行的结果;比如同时1000个任务去网络上抓取数据,然后将抓取到的数据进行处理(处理方式不定),我觉得最好的方式就是提供回调接口,把处理的方式最为回调传进去;但是现在我们有了更好的方式实现:CompletionService + Callable

    Callable的call方法可以返回执行的结果;

    CompletionService将Executor(线程池)和BlockingQueue(阻塞队列)结合在一起,同时使用Callable作为任务的基本单元,整个过程就是生产者不断把Callable任务放入阻塞对了,Executor作为消费者不断把任务取出来执行,并返回结果;

    优势:

    a、阻塞队列防止了内存中排队等待的任务过多,造成内存溢出(毕竟一般生产者速度比较快,比如爬虫准备好网址和规则,就去执行了,执行起来(消费者)还是比较慢的)

    b、CompletionService可以实现,哪个任务先执行完成就返回,而不是按顺序返回,这样可以极大的提升效率;

    1、CompletionService : Executor + BlockingQueue 

    下面看个例子:

    [java]  view plain copy package com.zhy.concurrency.completionService;      import java.util.Random;   import java.util.concurrent.BlockingQueue;   import java.util.concurrent.Callable;   import java.util.concurrent.CompletionService;   import java.util.concurrent.ExecutionException;   import java.util.concurrent.ExecutorCompletionService;   import java.util.concurrent.ExecutorService;   import java.util.concurrent.Executors;   import java.util.concurrent.Future;   import java.util.concurrent.LinkedBlockingDeque;      /**   * 将Executor和BlockingQueue功能融合在一起,可以将Callable的任务提交给它来执行, 然后使用take()方法获得已经完成的结果   *    * @author zhy   *    */   public class CompletionServiceDemo   {          public static void main(String[] args) throws InterruptedException,               ExecutionException       {           /**           * 内部维护11个线程的线程池           */           ExecutorService exec = Executors.newFixedThreadPool(11);           /**           * 容量为10的阻塞队列           */           final BlockingQueue<Future<Integer>> queue = new LinkedBlockingDeque<Future<Integer>>(                   10);           //实例化CompletionService           final CompletionService<Integer> completionService = new ExecutorCompletionService<Integer>(                   exec, queue);              /**           * 模拟瞬间产生10个任务,且每个任务执行时间不一致           */           for (int i = 0; i < 10; i++)           {               completionService.submit(new Callable<Integer>()               {                   @Override                   public Integer call() throws Exception                   {                       int ran = new Random().nextInt(1000);                       Thread.sleep(ran);                       System.out.println(Thread.currentThread().getName()                               + " 休息了 " + ran);                       return ran;                   }               });           }                      /**           * 立即输出结果           */           for (int i = 0; i < 10; i++)           {               try               {                      //谁最先执行完成,直接返回                   Future<Integer> f = completionService.take();                   System.out.println(f.get());               } catch (InterruptedException e)               {                   e.printStackTrace();               } catch (ExecutionException e)               {                   e.printStackTrace();               }           }              exec.shutdown();          }      }   输出结果:

    [java]  view plain copy pool-1-thread-4 休息了 52   52   pool-1-thread-1 休息了 59   59   pool-1-thread-10 休息了 215   215   pool-1-thread-9 休息了 352   352   pool-1-thread-5 休息了 389   389   pool-1-thread-3 休息了 589   589   pool-1-thread-2 休息了 794   794   pool-1-thread-7 休息了 805   805   pool-1-thread-6 休息了 909   909   pool-1-thread-8 休息了 987   987   最先执行完成的直接返回,并不需要按任务提交的顺序执行,如果需要写个高并发的程序,且每个任务需要返回执行结果,这是个相当不错的选择!
    最新回复(0)