本文为博主原创,转载请注明出处:
实现异步任务时,经常使用 FutureTask 来实现;一个简单的示例代码如下:
public static void main(String[] args) throws ExecutionException, InterruptedException { //构建线程池 ExecutorService executorService = Executors.newFixedThreadPool(2); FutureTask<Integer> futureTask1 = (FutureTask<Integer>) executorService.submit(()->doTask1()); FutureTask<Integer> futureTask2 = (FutureTask<Integer>) executorService.submit(()->doTask2()); // 获取电结果并异步保存 executorService.execute(()->save(futureTask1.get())); // 获取结果并异步保存 executorService.execute(()->save(futureTask2.get()); }
上面代码如果 futureTask1 的任务需要执行很长时间,而 futureTask2 执行很短时间,上面代码在执行的过程中,futureTask2 任务的执行也的先等 futureTask1.get() 执行结束后 ,才能保存 futureTask2.get() ;因为这个主线程都阻塞在 futureTask1.get() 的操作上;严重降低了效率。
此时可以使用 CompletionService 来解决这个问题
CompletionService 接口的功能是以异步的方式一边生产新的任务,一边处理已完成任务的结果,这样就可以将执行任务与处理任务分离开。
CompletionService的一个实现是ExecutorCompletionService,它是Executor和BlockingQueue功能的融合体,Executor完成计算任务,BlockingQueue负责保存异步任务的执行结果;先执行完的先进入阻塞队列,利用这个特性,你可以轻松实现后续处理的有序性,避免无谓的等待。在执行大量相互独立和同构的任务时,可以使用CompletionService;
该实现类定义的三个属性:
在类的注释上有使用的示例,可以参考学习
使用 CompletionService 实现的示例如下:
public static void main(String[] args) { // 创建线程池 ExecutorService executor = Executors.newFixedThreadPool(3); // 创建CompletionService CompletionService<Integer> cs = new ExecutorCompletionService<>(executor); // 用于保存Future对象 List<Future<Integer>> futures = new ArrayList<>(3); //提交异步任务,并保存future到futures futures.add(cs.submit(()->doTask1())); futures.add(cs.submit(()->doTask2())); futures.add(cs.submit(()->doTask3())); // 获取最快返回的任务执行结果 Integer r = 0; try { // 只要有一个成功返回,则break for (int i = 0; i < 3; ++i) { r = cs.take().get(); //简单地通过判空来检查是否成功返回 if (r != null) { break; } } } finally { //取消所有任务 for(Future<Integer> f : futures) f.cancel(true); } // 返回结果 }