关注小程序 找一找教程网-随时随地学编程

C/C++教程

CompletionService 介绍

介绍

java.util.concurrent.CompletionService 是对 ExecutorService 的一个功能增强封装,优化了获取异步操作结果的接口。

使用场景

假设我们要向线程池提交一批任务,并获取任务结果。一般的方式是提交任务后,从线程池得到一批 Future 对象集合,然后依次调用其 get() 方法。

这里有个问题:因为我们会要按固定的顺序来遍历 Future 元素,而 get() 方法又是阻塞的,因此如果某个 Future 对象执行时间太长,会使得我们的遍历过程阻塞在该元素上,无法及时从后面早已完成的 Future 当中取得结果。

CompletionService 解决了这个问题。它本身不包含线程池,创建一个 CompletionService 需要先创建一个 Executor。下面是一个例子:

ExecutorService executor = Executors.newFixedThreadPool(4);
CompletionService<String> completionService = new ExecutorCompletionService<>(executor);

CompletionService 提交任务的方式与 ExecutorService 一样:

completionService.submit(() -> "Hello");

当你需要获得结果的时候,就不同了。有了 CompletionService,你不需要再持有 Future 集合。如果要得到最早的执行结果,只需要像下面这样:

String result = completionService.take().get();

这个 take() 方法返回的是最早完成的任务的结果,这个就解决了一个任务被另一个任务阻塞的问题。下面是一个完整的例子:

示例

public static void main(String[] args) throws Exception {

    ExecutorService executor;
    CompletionService<String> completionService;

    // 创建一个指定执行时长的任务的方法
    BiFunction<Integer, Integer, Callable<String>> createTask = (id, duration) -> () -> {
        log("Task " + id + " started, duration=" + duration);
        Thread.sleep(duration);
        log("Task " + id + " completed.");
        return "Result of task " + id;
    };

    ///////////////////////////////////////////////////////////////////
    System.out.println("// 示例1:像使用 ExecutorService 一样使用 CompletionService");

    // 初始化 executor 和 completionService
    executor = Executors.newFixedThreadPool(4);
    completionService = new ExecutorCompletionService<>(executor);

    // 提交任务
    List<Future<String>> results = Arrays.asList(
            completionService.submit(createTask.apply(1, 1000)),
            completionService.submit(createTask.apply(2, 800)),
            completionService.submit(createTask.apply(3, 600)),
            completionService.submit(createTask.apply(4, 400))
    );

    // 取结果
    for (Future<String> result : results) {
        log(result.get());
    }

    executor.shutdown();

    ///////////////////////////////////////////////////////////////////
    System.out.println("// 示例2:按标准方式使用 CompletionService");

    // 初始化 executor 和 completionService
    executor = Executors.newFixedThreadPool(4);
    completionService = new ExecutorCompletionService<>(executor);

    // 提交任务
    completionService.submit(createTask.apply(5, 1000));
    completionService.submit(createTask.apply(6, 800));
    completionService.submit(createTask.apply(7, 600));
    completionService.submit(createTask.apply(8, 400));

    // 取结果
    for (int i = 0; i < 4; i++) {
        log(completionService.take().get());
    }

    ///////////////////////////////////////////////////////////////////
    executor.shutdown();
}

这个例子的执行结果如下所示:

// 示例1:像使用 ExecutorService 一样使用 CompletionService
10:22:32:271 - Task 4 started, duration=400
10:22:32:271 - Task 3 started, duration=600
10:22:32:271 - Task 2 started, duration=800
10:22:32:271 - Task 1 started, duration=1000
10:22:32:687 - Task 4 completed.
10:22:32:888 - Task 3 completed.
10:22:33:089 - Task 2 completed.
10:22:33:303 - Task 1 completed.
10:22:33:303 - Result of task 1
10:22:33:303 - Result of task 2
10:22:33:303 - Result of task 3
10:22:33:303 - Result of task 4
// 示例2:按标准方式使用 CompletionService
10:22:33:305 - Task 5 started, duration=1000
10:22:33:305 - Task 7 started, duration=600
10:22:33:305 - Task 6 started, duration=800
10:22:33:305 - Task 8 started, duration=400
10:22:33:718 - Task 8 completed.
10:22:33:718 - Result of task 8
10:22:33:918 - Task 7 completed.
10:22:33:918 - Result of task 7
10:22:34:119 - Task 6 completed.
10:22:34:119 - Result of task 6
10:22:34:320 - Task 5 completed.
10:22:34:320 - Result of task 5

可以看出,在示例 1 中,虽然 Task 4 执行时间只有 400ms,但因为我们是按照 1-2-3-4 的顺序依次取结果,因此 Task 4 完成后并没有马上打印出结果来。而在示例 2 中,对每个 Task 都是在完成时立刻就将结果打印出来了。这就是 CompletionService 的优势所在。

原理解释

CompletionService 之所以能够做到这点,是因为它没有采取依次遍历 Future 的方式,而是在中间加上了一个结果队列,任务完成后马上将结果放入队列,那么从队列中取到的就是最早完成的结果。

如果队列为空,那么 take() 方法会阻塞直到队列中出现结果为止。此外 CompletionService 还提供一个 poll() 方法,返回值与 take() 方法一样,不同之处在于它不会阻塞,如果队列为空则立刻返回 null。这算是给用户多一种选择。