CompletionService 使用小结

completionservice,使用,小结 · 浏览次数 : 37

小编点评

**实现异步任务的示例** ```java public static void main(String[] args) throws ExecutionException, InterruptedException { // 创建线程池 ExecutorService executorService = Executors.newFixedThreadPool(3); // 创建CompletionService CompletionService<Integer> cs = new ExecutorCompletionService<>(executorService); // 用于保存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); } // 返回结果 System.out.println(r); } ``` **使用 CompletionService 实现的示例** ```java 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); } // 返回结果 System.out.println(r); } ``` **注意：** * `ExecutorCompletionService` 是 `ExecutorService` 和 `BlockingQueue` 的融合体。 * `executorService` 是一个用于执行计算任务的线程池。 * `cs` 是一个用于保存 `Future` 的 `CompletionService`。 * `futures` 是一个存储 `Future` 的列表。 * `doTask1()`, `doTask2()` 和 `doTask3()` 是实际执行任务的函数。

正文

本文为博主原创，转载请注明出处：

　　实现异步任务时，经常使用 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);
        }
        // 返回结果
    }

与CompletionService 使用小结相似的内容：

CompletionService 使用小结

本文为博主原创，转载请注明出处：实现异步任务时，经常使用 FutureTask 来实现；一个简单的示例代码如下： public static void main(String[] args) throws ExecutionException, InterruptedException { //构