Java 完成队列如何实现最快结果优先：take、超时与剩余任务取消

做“多个后端同时查，谁先有可用结果就先返回”的接口时，按提交顺序调用每个 Future.get() 很容易被最慢任务卡住。Java 的 ExecutorCompletionService 把已经完成的任务放进完成队列，调用方可以用 take() 按完成顺序取结果；如果到了时间预算还没有结果，再用 poll() 判断超时并取消剩余任务。

ExecutorCompletionService 适合“结果按完成顺序消费”的场景：submit 负责投递，take 等待已完成项，poll 负责带时间预算地取项；超时后还要显式取消未消费的 Future。

先看场景：提交顺序不等于结果顺序

假设一个聚合接口并发查询三个来源：商品库存、门店库存和仓库库存。门店接口通常很快，但仓库接口偶尔需要重试。如果先把三个任务提交到线程池，再依次对三个 Future 调用 get()，第一个 Future 变慢时，后面已经完成的结果也无法先被业务使用。

ExecutorCompletionService 的关键不是让任务更快，而是改变结果的消费顺序。任务仍由 ExecutorService 执行；每个任务完成后，对应的 Future 进入完成队列（completion queue），消费方从队列取出“已经完成的那一个”。

用 submit 和 take 消费第一个完成结果

最小示例保留任务与结果的映射。submit 返回的 Future 放入列表，take 返回的是完成队列中的 Future；真正读取值仍要调用 get()。

ExecutorService pool = Executors.newFixedThreadPool(3);
CompletionService service = new ExecutorCompletionService(pool);
List> submitted = new ArrayList();

try {
    submitted.add(service.submit(() -> query("门店库存")));
    submitted.add(service.submit(() -> query("仓库库存")));
    submitted.add(service.submit(() -> query("商品库存")));

    for (int i = 0; i  completed = service.take();
        Stock stock = completed.get();
        if (stock.isUsable()) {
            return stock;
        }
    }
    return Stock.empty();
} finally {
    pool.shutdown();
}

这里的 take 会在完成队列为空时等待；它适合没有严格截止时间、但必须等到某个任务完成的场景。get() 仍可能抛出 ExecutionException，所以真实代码要把单个任务失败当作一个可记录、可跳过的结果，而不是让整个消费循环失去清理机会。

有时间预算时，用 poll 区分“没结果”和“任务失败”

接口只给聚合查询 300 毫秒时，不能直接对 take() 无限等待。用截止时间计算每一轮剩余预算，再调用带超时的 poll。返回 null 只表示在这段等待时间内没有完成项，并不等于所有任务都失败。

long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(300);
List> submitted = new ArrayList();

try {
    for (String source : List.of("门店库存", "仓库库存", "商品库存")) {
        submitted.add(service.submit(() -> query(source)));
    }

    for (int received = 0; received  completed = service.poll(remaining, TimeUnit.NANOSECONDS);
        if (completed == null) {
            break;
        }
        try {
            Stock stock = completed.get();
            if (stock.isUsable()) {
                return stock;
            }
        } catch (ExecutionException failed) {
            log.warn("库存来源失败", failed.getCause());
        }
    }
    return Stock.empty();
} finally {
    for (Future future : submitted) {
        if (!future.isDone()) {
            future.cancel(true);
        }
    }
    pool.shutdown();
}

poll 的超时时间必须是剩余预算，而不是每轮重新给 300 毫秒；否则三个任务可能把总等待时间放大。这个循环还区分了两种状态：completed == null 是等待窗口耗尽，ExecutionException 是某个已完成任务自身失败。

ExecutorService、CompletionService 和直接 Future 怎么选

如果任务内部调用了不响应中断的阻塞客户端，cancel(true) 只能设置中断标志，不能保证底层调用立即停止。因此取消之后仍要观察线程池和下游连接的实际状态，不要把 Future 变成“已取消”误认为资源已经释放。

常见问题

take() 返回的 Future 为什么还要调用 get()？

完成队列只负责告诉你哪个任务已完成，并把对应 Future 交出来；结果值或异常仍由 get() 读取。

poll 返回 null 是不是任务都失败了？

不是。它表示本次等待窗口没有完成项，可能是任务仍在运行，也可能是总时间预算已经耗尽。

调用 cancel(true) 后还需要关闭线程池吗？

需要。取消是任务级动作，shutdown() 是执行器级动作；两者分别处理未完成任务和不再接收新任务。

总结

需要最快结果优先时，ExecutorCompletionService 能把结果消费从提交顺序中解耦。没有硬截止时间就用 take 等待，有时间预算就用 poll 配合单调时钟计算剩余时间；预算耗尽后取消未完成 Future，再关闭 ExecutorService。最后别忘了验证任务本身是否真正响应中断，这决定了超时方案能否在生产环境收干净。