java并发 ExecutorCpmpletionService

实现ExecutorCompletionService的步骤

简介

在Java并发编程中，ExecutorCompletionService是一个方便的工具类，用于处理一批任务并获取它们的结果。它能够将多个任务提交给Executor执行，并将它们的结果按照完成的顺序返回给调用者。本文将详细介绍如何使用ExecutorCompletionService实现这个功能。

流程概览

下面是使用ExecutorCompletionService的基本流程：

1. 创建ExecutorCompletionService对象；
2. 提交任务到ExecutorCompletionService；
3. 获取已完成的任务结果。

接下来，我们将逐步介绍这些步骤和需要用到的代码。

步骤详解

1. 创建ExecutorCompletionService对象

首先，我们需要创建一个ExecutorCompletionService对象，它是Executor和CompletionService的组合。我们可以使用ThreadPoolExecutor来创建Executor，并将其传递给ExecutorCompletionService的构造函数。

Executor executor = Executors.newFixedThreadPool(nThreads);
ExecutorCompletionService<T> completionService = new ExecutorCompletionService<>(executor);

上述代码创建了一个固定大小的线程池，创建了一个ExecutorCompletionService对象。

2. 提交任务到ExecutorCompletionService

接下来，我们需要将任务提交给ExecutorCompletionService来执行。我们可以使用submit方法来提交任务。

Future<T> future = completionService.submit(new Callable<T>() {
    @Override
    public T call() throws Exception {
        // 任务的具体实现
        return result;
    }
});

上述代码创建了一个Callable对象，并将其提交给ExecutorCompletionService来执行。任务的具体实现可以在call方法中编写，返回值将作为任务的结果。

3. 获取已完成的任务结果

最后，我们需要获取已完成的任务结果。我们可以使用take方法来获取已完成的任务结果，它会阻塞直到有任务完成。

try {
    Future<T> completedFuture = completionService.take();
    T result = completedFuture.get();
    // 处理任务结果
} catch (InterruptedException e) {
    // 处理中断异常
} catch (ExecutionException e) {
    // 处理执行异常
}

上述代码使用take方法获取已完成的任务结果，并使用get方法获取任务的执行结果。如果任务执行过程中发生中断或执行异常，我们可以在catch块中进行相应的处理。

完整示例代码

下面是一个完整的示例代码，展示了如何使用ExecutorCompletionService实现并发任务的执行和结果获取。

import java.util.concurrent.*;

public class ExecutorCompletionServiceExample {
    public static void main(String[] args) {
        // 创建一个线程池
        Executor executor = Executors.newFixedThreadPool(3);
        // 创建ExecutorCompletionService
        ExecutorCompletionService<Integer> completionService = new ExecutorCompletionService<>(executor);

        // 提交任务
        completionService.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                Thread.sleep(5000); // 模拟任务执行时间
                return 1;
            }
        });

        completionService.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                Thread.sleep(3000); // 模拟任务执行时间
                return 2;
            }
        });

        completionService.submit(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                Thread.sleep(1000); // 模拟任务执行时间
                return 3;
            }
        });

        // 获取已完成的任务结果
        try {
            for (int i = 0; i < 3; i++) {
                Future<Integer> completedFuture = completionService.take();
                Integer result = completedFuture.get();
                System.out.println("Task " + (i + 1) + " completed with result: " + result);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        // 关闭线程池
        ((ThreadPoolExecutor) executor).shutdown();
    }
}

上述代码创建了一个线程池，使用ExecutorCompletionService提交了三个任务，并获取了它们的结果。任务的执行时间分别为5秒、3秒和1秒，因此结果的输出顺序应该是3、2、1。

序列图

下面是使用ExecutorCompletionService的流程的序列图表示：

sequenceDiagram
    participant Developer as 开发者
    participant ExecutorCompletionService as ExecutorCompletionService
    participant Executor as Executor
    participant ThreadPoolExecutor as ThreadPoolExecutor
    participant Callable as Callable
    participant Future