按完成顺序收集结果：用 CompletionService 实现有限滚动窗口

按提交顺序遍历 Future，第一项慢任务还没有返回，第二项已经完成的结果也只能等着。任务确实并发执行了，收集端却重新排出一条队伍。若目标是“谁先完成就先处理谁”，需要把结果等待顺序从提交顺序中拆出来。

ExecutorCompletionService 把完成的 Future 放进完成队列，调用方从队列取下一个结果即可。但它不会替我们限制提交数量：一口气提交十万个任务，再说“我只开了四个线程”，仍然可能积压十万个任务对象和结果。这里把完成队列与滚动窗口放在一起实现。

窗口统计的是尚未消费的任务

本例最多同时保留 window 个已提交、尚未处理的 Future。一项结果取走并交给处理器以后，才补交下一项。这个上限覆盖正在运行、执行器排队以及已经完成但还没被消费的项目，而不只是线程数。

任务数量另限一万，窗口限一至十六，整批等待预算最多三十秒。它们都是演示入口的明确策略。只限制并发而允许输入列表无限增长，或者把所有结果一直保留，也不能宣称整体常量内存；本实现为返回完整 Report 会保存最多一万条结果。

失败按项目记录，批次继续处理剩余项目。整批截止时间、调用线程中断、结果处理器抛异常，则退出批次并取消仍在窗口内的任务。被取消的底层任务能否及时停下，仍取决于它是否响应中断与自身 I/O 截止时间。

一份能观察提交次序与完成次序的程序

环境为 Java 17，保存为 CompletionWindow.java，执行 java CompletionWindow.java。没有外部网络：第一项用门闩阻塞，第二项可以立即完成，只有处理了第二项，才放开第一项。这个安排让首个消费结果稳定可验证。

import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Consumer;

public class CompletionWindow {
    record Job(String id, Callable<String> work) {}
    record Outcome(String id, String value, Throwable failure) {}
    record Report(List<Outcome> outcomes, int peakOutstanding) {}

    static Report execute(List<Job> input, int window, Duration budget,
                          Consumer<Outcome> handler) throws InterruptedException, TimeoutException {
        if (window < 1 || window > 16) throw new IllegalArgumentException("invalid window");
        if (budget.isZero() || budget.isNegative() || budget.compareTo(Duration.ofSeconds(30)) > 0)
            throw new IllegalArgumentException("invalid budget");
        List<Job> jobs = List.copyOf(input);
        if (jobs.size() > 10000) throw new IllegalArgumentException("too many jobs");
        Set<String> ids = new HashSet<>();
        for (Job job : jobs) {
            if (job.id() == null || !ids.add(job.id()) || job.work() == null)
                throw new IllegalArgumentException("invalid or duplicate job");
        }
        Objects.requireNonNull(handler);
        if (jobs.isEmpty()) return new Report(List.of(), 0);
        ExecutorService executor = Executors.newFixedThreadPool(window);
        CompletionService<String> completed = new ExecutorCompletionService<>(executor);
        Map<Future<String>, String> pending = new HashMap<>();
        List<Outcome> outcomes = new ArrayList<>();
        long deadline = System.nanoTime() + budget.toNanos();
        int next = 0, peak = 0;
        try {
            while (next < jobs.size() || !pending.isEmpty()) {
                while (next < jobs.size() && pending.size() < window) {
                    if (deadline - System.nanoTime() <= 0) throw new TimeoutException("batch deadline");
                    Job job = jobs.get(next++);
                    pending.put(completed.submit(job.work()), job.id());
                    peak = Math.max(peak, pending.size());
                }
                long remaining = deadline - System.nanoTime();
                if (remaining <= 0) throw new TimeoutException("batch deadline");
                Future<String> future = completed.poll(remaining, TimeUnit.NANOSECONDS);
                if (future == null) throw new TimeoutException("batch deadline");
                String id = pending.remove(future);
                Outcome outcome;
                try { outcome = new Outcome(id, future.get(), null); }
                catch (ExecutionException e) { outcome = new Outcome(id, null, e.getCause()); }
                catch (CancellationException e) { outcome = new Outcome(id, null, e); }
                outcomes.add(outcome);
                handler.accept(outcome);
            }
            return new Report(List.copyOf(outcomes), peak);
        } finally {
            pending.keySet().forEach(future -> future.cancel(true));
            executor.shutdownNow();
            if (!executor.awaitTermination(3, TimeUnit.SECONDS))
                System.err.println("executor has not terminated; tasks must cooperate with cancellation");
        }
    }
    public static void main(String[] args) throws Exception {
        CountDownLatch releaseSlow = new CountDownLatch(1);
        List<Job> jobs = List.of(
            new Job("slow-first", () -> { releaseSlow.await(); return "slow result"; }),
            new Job("fast-second", () -> "fast result"),
            new Job("broken", () -> { throw new IllegalStateException("probe failed"); }),
            new Job("last", () -> "last result")
        );
        List<String> handled = new ArrayList<>();
        Report report = execute(jobs, 2, Duration.ofSeconds(3), outcome -> {
            handled.add(outcome.id());
            if (outcome.id().equals("fast-second")) releaseSlow.countDown();
        });
        System.out.println("first handled=" + handled.get(0));
        System.out.println("success=" + report.outcomes().stream().filter(x -> x.failure() == null).count());
        System.out.println("failed=" + report.outcomes().stream().filter(x -> x.failure() != null).count());
        System.out.println("peak outstanding=" + report.peakOutstanding());
        if (!handled.get(0).equals("fast-second") || report.outcomes().size() != 4)
            throw new AssertionError("completion contract");
        CountDownLatch never = new CountDownLatch(1);
        try {
            execute(List.of(new Job("blocked", () -> { never.await(); return "unreachable"; })),
                    1, Duration.ofMillis(40), outcome -> {});
            throw new AssertionError("timeout expected");
        } catch (TimeoutException expected) { System.out.println("timeout=true"); }
    }
}

运行输出：

first handled=fast-second
success=3
failed=1
peak outstanding=2
timeout=true

第二项先被处理，证明收集没有等待提交在前的第一项。其余成功项和失败项之间没有承诺固定排序，因此程序只断言首项、结果数量与窗口峰值，不把一次调度打印冒充普遍完成次序。

最后一项单独批次使用不会自行打开的门闩，触发整批超时。门闩等待可以被中断，所以 finally 中的取消能够收回实验线程。四十毫秒只是触发分支的等待预算，输出没有声称真实服务能在四十毫秒内完全停止所有后台动作。

从队列取出的仍然是 Future

完成队列只说明任务已经达到完成状态，不代表业务成功。调用 get 才会获得返回值或 ExecutionException；异常原因通过 getCause 保存到 Outcome，而不是把失败结果当成 null 值混过去。一个业务上合法返回 null 的任务，也因此能与失败区分。

pending 映射让失败的 Future 仍能找到任务标识。如果只让成功返回值携带 id，任务在构造结果之前抛异常时，收集端可能不知道是哪一项失败。标识来自提交端，结果和异常来自执行端，两份信息在消费处合并。

结果处理器在调用线程串行执行，不需要额外给 handled 列表加同步；任务本身仍在执行器线程运行。如果把 handler 改成再次异步提交，方法返回时结果是否已处理、异步队列是否有界，都要重新约定，不能沿用当前 Report 的完成含义。

handler 的异常也没有被包装成某个任务失败。任务已经完成，结果处理却失败，通常属于另一项责任。这里让异常正常向外传播，并在 finally 停止本批未消费任务。真实系统若把结果写入存储，应同时设计处理幂等，避免重跑批次导致已经写入的结果重复生效。

为什么没有给完成队列套一个小容量

默认完成队列采用 LinkedBlockingQueue。容易想到的“优化”是换成一个容量很小的队列，但 ExecutorCompletionService 按队列可以无界接收的方式使用它；完成入队失败时，任务并不会自动变成一种可靠背压，结果甚至可能无法再从完成队列取到。

本例保留默认队列，在提交入口限制未消费数量。这样任意时刻能进入完成队列的 Future 总量也受窗口约束，消费慢时不会继续补交无限项目。这种约束建立在同一个批次拥有全部提交入口的前提上；其他代码绕过 execute 往相同执行器提交任务，会带来额外负载，不能算在这个窗口的保证里。

执行器虽然使用固定线程池默认工作队列，但这里只允许有限窗口进入，所以这条受控路径不会向它灌入整个输入列表。线程数上限与任务对象上限在代码里分别可见，后续审查也能直接找到真正实施约束的 pending.size 条件，而不是从构造函数名字猜测。

一个截止时间，还是每项各等一次

代码只在开头创建整批 deadline，每轮计算剩余时间。若每次都调用 poll(timeout)，十个慢结果可能各自消耗一份完整预算，整批等待就远超调用方预期。总预算与单项预算都可以合理，方法参数必须说明是哪一种。

补交新任务之前也检查 deadline，防止处理器已经耗尽预算后又启动新的工作。这个检查不能中断一个阻塞在 handler 里的调用，因此 handler 应该快速返回，或自行对外部写入设置截止时间。一个线程正在同步执行任意函数，外层的时间戳不会自动让它停止。

nanoTime 用于计算经过时间，不依赖墙上日期调整。这里最长三十秒，也避免将无限长 Duration 转成纳秒后产生不适合的区间。保存 deadline 与计算剩余值属于同一进程内的短期等待协议，不应把这个纳秒值作为数据库里的未来执行时间。

取消发出去以后，还要管理执行器寿命

finally 只取消 pending 中仍未消费的 Future；已经处理的项目不再持有。随后调用 shutdownNow 并有限等待终止，避免本地演示因为线程池存活而一直不退出。如果某个任务无视中断，程序只报告没有终止，不能打印“全部停止”来掩盖它仍在后台工作。

对网络探测、数据库查询或外部进程，应让底层操作具备自己的可取消机制与时间限制。Future.cancel(true) 是中断请求，不能保证任意业务副作用撤回；已经发送的远程请求甚至可能继续在服务端完成。因此超时后的重试仍需幂等与结果核对，完成队列不负责分布式取消。

这个方法自己创建执行器，所以自己关闭。如果业务传入共享执行器，就应把生命周期交还拥有者，不能在一次批次结束时关闭整个应用的线程池。与此同时，窗口只代表当前调用的在途数量，并不会自动变成跨多个调用的全局并发上限。

需要第一个成功结果就结束的场景，可以在处理到满足条件的 Outcome 后退出，并保留 finally 的取消逻辑；需要全部结果时就像现在这样消费到计数归零。两种策略的停止条件不同，不能为了让最快失败先返回，就把整批剩余结果默默丢弃。