关于java:CompletionService-介绍

2次阅读

共计 3462 个字符，预计需要花费 9 分钟才能阅读完成。

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。这算是给用户多一种抉择。

正文完