ExecutorCompletionService详解与使用教程

最近发现不少小伙伴都对文章很感兴趣，所以今天继续给大家介绍文章相关的知识，本文《Java ExecutorCompletionService详解与使用教程》主要内容涉及到等等知识点，希望能帮到你！当然如果阅读本文时存在不同想法，可以在评论中表达，但是请勿使用过激的措辞~

答案：ExecutorCompletionService通过将任务结果存入阻塞队列，使结果按完成顺序而非提交顺序被处理。它结合了Executor和BlockingQueue的优点，在任务执行时间不确定的场景下，避免了因等待慢任务而阻塞后续已完成任务结果的获取。与直接使用ExecutorService的Future.get()相比，后者必须按提交顺序阻塞等待，而CompletionService提供take()方法实时获取最先完成的任务结果，提升响应速度和资源利用率。典型应用场景包括爬虫请求、渐进式数据处理和资源敏感型任务。使用时需注意异常处理（如CancellationException）、手动管理线程池生命周期、正确设置结果获取循环终止条件，并权衡其额外队列开销。代码示例展示了提交10个异步任务并按完成顺序打印结果的过程，最后安全关闭线程池。

在Java中，当你需要并行执行多个任务，并且希望以任务完成的顺序（而不是提交的顺序）来处理结果时，ExecutorCompletionService是一个非常实用的工具。它本质上是Executor和BlockingQueue的结合体，让你能够异步提交任务，然后阻塞式地获取已完成的任务结果。这对于那些任务执行时间不确定，且你需要尽快处理已完成任务结果的场景来说，简直是量身定制。

解决方案

使用ExecutorCompletionService的核心在于将你的任务提交给它，然后通过它的take()方法来获取已完成的任务结果。这比直接从ExecutorService获取Future然后逐个get()要灵活得多，因为它不会让你被第一个提交但可能最慢的任务阻塞。

下面是一个基本的使用示例：

import java.util.concurrent.*;
import java.util.Random;

public class CompletionServiceDemo {

    public static void main(String[] args) {
        // 1. 创建一个线程池，作为ExecutorCompletionService的底层执行器
        // 我个人喜欢用FixedThreadPool，因为它能控制并发度，避免资源耗尽
        ExecutorService executor = Executors.newFixedThreadPool(5);

        // 2. 创建ExecutorCompletionService实例，将线程池传入
        // 这一步是关键，它将ExecutorService包装起来，提供了更高级的完成服务
        CompletionService completionService = new ExecutorCompletionService<>(executor);

        int numberOfTasks = 10;
        Random random = new Random();

        // 3. 提交Callable任务
        // 每个任务模拟不同的执行时间
        for (int i = 0; i < numberOfTasks; i++) {
            final int taskId = i;
            completionService.submit(() -> {
                long sleepTime = random.nextInt(3000) + 500; // 模拟0.5到3.5秒的执行时间
                System.out.println("任务 " + taskId + " 开始执行，预计耗时 " + sleepTime + "ms");
                Thread.sleep(sleepTime);
                return "任务 " + taskId + " 完成，耗时 " + sleepTime + "ms";
            });
        }

        // 4. 获取并处理已完成的任务结果
        // take()方法会阻塞，直到有任务完成并返回其Future
        System.out.println("\n--- 开始获取任务结果 ---\n");
        for (int i = 0; i < numberOfTasks; i++) {
            try {
                // take()方法阻塞等待第一个完成的任务
                Future future = completionService.take();
                // get()方法不会阻塞，因为任务已经完成
                String result = future.get();
                System.out.println("处理结果: " + result);
            } catch (InterruptedException e) {
                // 当前线程被中断，通常意味着需要停止处理
                Thread.currentThread().interrupt();
                System.err.println("等待任务完成时被中断: " + e.getMessage());
            } catch (ExecutionException e) {
                // 任务执行过程中抛出了异常
                System.err.println("任务执行异常: " + e.getMessage());
                // 可以进一步检查e.getCause()获取原始异常
            } catch (CancellationException e) {
                // 任务被取消了
                System.err.println("任务被取消: " + e.getMessage());
            }
        }

        // 5. 关闭线程池
        // 这一步很重要，否则程序可能不会退出
        executor.shutdown();
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow(); // 如果无法在指定时间内关闭，则强制关闭
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }
        System.out.println("\n所有任务处理完毕，线程池已关闭。");
    }
}

ExecutorCompletionService与ExecutorService直接获取Future有什么不同？

我记得刚开始接触多线程的时候，总是习惯性地把所有Future都收集起来，然后挨个get()。结果发现，如果第一个任务是个“慢郎中”，后面那些明明已经跑完了的“快枪手”，也只能干等着。那种感觉，就像是排队买票，前面有个老大爷慢慢悠悠地数钱，后面一堆急着赶火车的都干着急。ExecutorCompletionService就是来解决这个痛点的，它像是给你开辟了一条“快速通道”，谁先跑完谁先出结果，不用等前面的人。

具体来说，当你直接向ExecutorService提交任务时，submit()方法会立即返回一个Future对象。如果你把这些Future对象存储在一个列表中，然后遍历列表并调用每个Future的get()方法，那么：

1. 第一个future.get()调用会阻塞，直到第一个任务完成。
2. 第二个future.get()调用会阻塞，直到第二个任务完成（即使它可能比第一个任务早完成）。 这种方式的问题在于，你必须按照任务提交的顺序来获取结果，即使后面的任务已经早早地完成了。

而ExecutorCompletionService则不同。它内部维护了一个队列，专门存放已经完成的任务的Future对象。当你调用completionService.take()时，它会阻塞，直到队列中有可用的Future（即有任务完成了），然后立即返回那个已完成任务的Future。这意味着你可以以任务完成的实际顺序来处理结果，大大提高了程序的响应性和吞吐量，尤其是在任务执行时间差异很大的场景下。

什么时候应该考虑使用ExecutorCompletionService？

我个人在做一些爬虫项目或者数据处理管道时，特别喜欢用它。想象一下，你发出了几百个HTTP请求，有些服务器响应快，有些慢。如果我用传统的Future.get()，我可能要等最慢的那个请求才能开始处理第一个结果。但有了CompletionService，第一个请求一回来，我就可以立即解析数据，甚至继续发出新的请求，整个流程就显得非常流畅和高效。

具体来说，以下几种情况，你真的应该考虑它：

• 任务执行时间不确定且差异大： 这是最典型的场景。如果你的任务有的几毫秒完成，有的需要几秒甚至更久，并且你希望尽快处理那些已经完成的任务结果，ExecutorCompletionService能让你避免被慢任务拖累。
• 需要实时反馈或渐进式处理结果： 例如，一个批处理系统，你希望每完成一个子任务就更新进度条，或者进行下一步操作（比如将数据写入数据库）。take()方法能让你实时获取到完成的任务，并立即进行后续处理。
• 资源敏感型任务： 有些任务完成后，可能会释放一些宝贵的资源（如数据库连接、文件句柄）。通过CompletionService，你可以更快地知道任务完成，从而更快地回收这些资源。
• 避免死锁或资源耗尽： 在某些复杂的并发场景中，如果你提交了大量任务，并且需要处理它们的结果来决定是否继续提交或释放资源，take()的阻塞特性可以很好地控制流程，避免系统过载。
• 构建响应式或事件驱动的系统： 如果你的系统需要对任务完成事件做出响应，CompletionService提供了一种优雅且高效的机制来监听这些事件。

ExecutorCompletionService有哪些潜在的陷阱或需要注意的地方？

我曾经犯过一个错误，就是忘记处理CancellationException。当时想当然地以为，任务取消了就不会有Future出来。结果发现，它还是会出来，只不过get()的时候会抛异常。这就提醒我们，任何时候，对Future的get()操作都应该包裹在try-catch块里，把InterruptedException、ExecutionException和CancellationException都考虑进去，这样代码才足够健壮。

除了异常处理，还有一些点值得我们留意：

• 任务取消的处理： 如果你通过Future.cancel(true)取消了一个提交给CompletionService的任务，那么这个任务的Future仍然会出现在take()的队列中。当你对这个被取消的Future调用get()时，它会抛出CancellationException。所以，你的结果处理逻辑需要能够优雅地处理这种情况。
• 异常处理的完整性： future.get()方法可能会抛出InterruptedException（如果等待结果时当前线程被中断）、ExecutionException（如果任务执行过程中抛出了未捕获的异常）以及前面提到的CancellationException。你必须为这些异常提供健壮的捕获和处理逻辑，否则程序可能会意外终止或行为异常。
• 底层ExecutorService的生命周期管理： ExecutorCompletionService只是一个包装器，它不负责管理底层ExecutorService的生命周期。这意味着你需要手动调用executor.shutdown()来关闭线程池，否则程序可能无法正常退出，或者线程资源得不到释放。我通常会用try-finally或者在所有任务完成后显式地关闭。
• 结果获取循环的终止条件： 在获取结果的循环中，你需要一个明确的终止条件，例如知道总共有多少个任务需要处理，或者有一个哨兵任务来指示所有任务已提交。如果循环条件设置不当，take()可能会无限期地阻塞下去，导致程序卡死。
• 性能考量（微小开销）： 对于非常简单、执行时间极短且对处理顺序没有特殊要求的任务，ExecutorCompletionService引入了一个额外的队列层，可能会带来微乎其微的性能开销。但在绝大多数实际场景中，其带来的便利性和效率提升远超这点开销。

总的来说，ExecutorCompletionService是一个非常强大的工具，但它的威力需要你正确地理解和使用。多思考异常情况和资源管理，你的代码会更加健壮和高效。

今天关于《ExecutorCompletionService详解与使用教程》的内容介绍就到此结束，如果有什么疑问或者建议，可以在golang学习网公众号下多多回复交流；文中若有不正之处，也希望回复留言以告知！