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()方法，那么：
- 第一个
future.get()
调用会阻塞，直到第一个任务完成。 - 第二个
future.get()
调用会阻塞，直到第二个任务完成（即使它可能比第一个任务早完成）。 这种方式的问题在于，你必须按照任务提交的顺序来获取结果，即使后面的任务已经早早地完成了。
而
ExecutorCompletionService则不同。它内部维护了一个队列，专门存放已经完成的任务的
Future对象。当你调用
completionService.take()时，它会阻塞，直到队列中有可用的
Future（即有任务完成了），然后立即返回那个已完成任务的
Future。这意味着你可以以任务完成的实际顺序来处理结果，大大提高了程序的响应性和吞吐量，尤其是在任务执行时间差异很大的场景下。
什么时候应该考虑使用ExecutorCompletionService
？
我个人在做一些爬虫项目或者数据处理管道时，特别喜欢用它。想象一下，你发出了几百个HTTP请求，有些服务器响应快，有些慢。如果我用传统的
future.get()，我可能要等最慢的那个请求才能开始处理第一个结果。但有了
CompletionService，第一个请求一回来，我就可以立即解析数据，甚至继续发出新的请求，整个流程就显得非常流畅和高效。
具体来说，以下几种情况，你真的应该考虑它：
-
任务执行时间不确定且差异大： 这是最典型的场景。如果你的任务有的几毫秒完成，有的需要几秒甚至更久，并且你希望尽快处理那些已经完成的任务结果，
ExecutorCompletionService
能让你避免被慢任务拖累。 -
需要实时反馈或渐进式处理结果： 例如，一个批处理系统，你希望每完成一个子任务就更新进度条，或者进行下一步操作（比如将数据写入数据库）。
ta
方法能让你实时获取到完成的任务，并立即进行后续处理。
ke() -
资源敏感型任务： 有些任务完成后，可能会释放一些宝贵的资源（如数据库连接、文件句柄）。通过
CompletionService
，你可以更快地知道任务完成，从而更快地回收这些资源。 -
避免死锁或资源耗尽： 在某些复杂的并发场景中，如果你提交了大量任务，并且需要处理它们的结果来决定是否继续提交或释放资源，
take()
的阻塞特性可以很好地控制流程，避免系统过载。 -
构建响应式或事件驱动的系统： 如果你的系统需要对任务完成事件做出响应，
CompletionService
提供了一种优雅且高效的机制来监听这些事件。
ke()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是一个非常强大的工具，但它的威力需要你正确地理解和使用。多思考异常情况和资源管理，你的代码会更加健壮和高效。
