当前位置： 技术文章>> Java 中如何使用 ExecutorCompletionService？

文章标题：Java 中如何使用 ExecutorCompletionService？
  • 文章分类: 后端
  • 8697 阅读

在Java并发编程中，ExecutorCompletionService 是一个非常有用的工具类，它封装了 Executor（执行器）服务，以便能够异步执行任务并获取这些任务的结果。它特别适合用于处理多个异步任务的场景，尤其是当你需要按照任务完成的顺序来处理结果时。下面，我们将深入探讨 ExecutorCompletionService 的使用方式，包括其基本原理、使用场景、以及一个详细的示例，展示如何在实际项目中应用它。

基本原理

ExecutorCompletionService 内部维护了一个阻塞队列（如 BlockingQueue），用于存储已完成任务的 Future 对象。当你提交一个任务给 ExecutorCompletionService 时，这个任务会被封装成一个 Future 对象，并提交到背后的 Executor 上执行。当任务完成后，它的 Future 对象会被放入到之前提到的阻塞队列中。这样，你就可以通过调用 take() 或 poll() 方法从阻塞队列中获取已完成任务的 Future 对象，进而获取任务执行的结果。

使用场景

ExecutorCompletionService 的使用场景非常广泛，特别是在需要并行处理多个任务，并且这些任务的执行时间可能不同，但你希望按照它们完成的顺序来处理结果的场景下。例如：

  • 数据并行处理：当你需要从多个数据源并行加载数据，但处理这些数据时希望按照它们加载完成的顺序来执行。
  • 网络请求：在并发发送多个网络请求时，你可能希望先处理先返回结果的请求。
  • 批量任务执行：在批处理任务时，如果任务之间没有依赖关系，但你想尽快地处理完所有任务并获取结果。

示例：使用 ExecutorCompletionService 处理多个异步任务

接下来，我们通过一个具体的示例来演示如何使用 ExecutorCompletionService 来处理多个异步任务。

准备工作

首先，定义一个简单的任务类，这个类实现了 Callable 接口，以便能够返回执行结果。

import java.util.concurrent.Callable;

public class MyTask implements Callable<String> {
    private final int taskId;
    private final int duration; // 模拟任务执行时间

    public MyTask(int taskId, int duration) {
        this.taskId = taskId;
        this.duration = duration;
    }

    @Override
    public String call() throws InterruptedException {
        // 模拟任务执行时间
        Thread.sleep(duration);
        return "Task " + taskId + " completed after " + duration + " ms.";
    }
}

使用 ExecutorCompletionService

然后，我们编写主程序，使用 ExecutorCompletionService 来执行多个 MyTask 实例。

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class ExecutorCompletionServiceExample {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        // 创建一个固定大小的线程池
        ExecutorService executor = Executors.newFixedThreadPool(4);

        // 创建一个ExecutorCompletionService实例，传入上面的线程池
        ExecutorCompletionService<String> completionService = new ExecutorCompletionService<>(executor);

        // 准备任务列表
        List<Callable<String>> tasks = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            // 假设每个任务的执行时间不同
            int duration = (int) (Math.random() * 1000);
            tasks.add(new MyTask(i, duration));
        }

        // 提交所有任务到ExecutorCompletionService
        for (Callable<String> task : tasks) {
            completionService.submit(task);
        }

        // 关闭ExecutorService（注意：这不会立即停止正在执行的任务）
        executor.shutdown();

        // 等待所有任务完成，并按完成顺序处理结果
        try {
            for (int i = 0; i < tasks.size(); i++) {
                // take() 会阻塞，直到有任务完成
                Future<String> future = completionService.take();
                System.out.println(future.get()); // 获取并打印任务结果
            }
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

        // 等待ExecutorService中的线程完全终止
        if (!executor.isTerminated()) {
            executor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
        }
    }
}

分析

在上面的示例中，我们首先创建了一个固定大小的线程池（ExecutorService），并用它初始化了一个 ExecutorCompletionService 实例。然后，我们创建了一个包含10个 MyTask 任务的列表，这些任务的执行时间随机生成。接下来，我们将这些任务提交给 ExecutorCompletionService，并通过调用 take() 方法按任务完成的顺序获取并处理结果。注意，take() 方法会阻塞当前线程，直到有任务完成。

最后，我们通过调用 shutdown() 方法来启动线程池的关闭过程（注意，这不会立即停止正在执行的任务），并通过 awaitTermination() 方法等待线程池中的所有线程完全终止。

注意事项

  • 当你使用 ExecutorCompletionService 时，应该确保在适当的时机关闭背后的 ExecutorService，以避免资源泄露。
  • 如果任务执行过程中抛出异常，这些异常会被封装在 ExecutionException 中，你需要捕获并处理这个异常来获取任务失败的具体原因。
  • ExecutorCompletionService 的 take() 方法会阻塞调用线程，直到有任务完成。如果你不想阻塞当前线程，可以使用 poll(long timeout, TimeUnit unit) 方法，该方法会等待指定的时间后返回，如果没有任务完成则返回 null。

结语

ExecutorCompletionService 是Java并发编程中一个非常强大的工具，它能够帮助你以高效且灵活的方式处理多个异步任务的结果。通过上面的示例，你应该已经对如何使用 ExecutorCompletionService 有了清晰的认识。希望这个示例能够对你的项目开发有所帮助，并激发你对Java并发编程更深入的探索。在码小课网站上，你可以找到更多关于Java并发编程的资源和教程，帮助你进一步提升编程技能。

推荐文章