线程协作与中断：用明确协议等待结果并结束任务

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.function.IntUnaryOperator;

public class Main {
    record Indexed(int index,int value) {}
    static int[] map(int[] input,int k,IntUnaryOperator function)
            throws InterruptedException,ExecutionException {
        Objects.requireNonNull(input); Objects.requireNonNull(function);
        if(k<=0) throw new IllegalArgumentException();
        if(Thread.interrupted()) throw new InterruptedException();
        int[] data=input.clone(), result=new int[data.length];
        if(data.length==0) return result;
        ExecutorService executor=Executors.newFixedThreadPool(Math.min(k,data.length));
        CompletionService<Indexed> completed=new ExecutorCompletionService<>(executor);
        Set<Future<Indexed>> active=new HashSet<>();
        int next=0, done=0;
        try {
            while(next<data.length && active.size()<k) {
                int index=next++;
                active.add(completed.submit(() -> new Indexed(index,function.applyAsInt(data[index]))));
            }
            while(done<data.length) {
                Future<Indexed> future=completed.take(); // 按完成顺序回收，及时腾出窗口。
                active.remove(future);
                Indexed value=future.get(); result[value.index()]=value.value(); done++;
                if(next<data.length) {
                    int index=next++;
                    active.add(completed.submit(() -> new Indexed(index,function.applyAsInt(data[index]))));
                    // 每回收一项最多补一项，保持活跃任务不超过k。
                }
            }
            return result;
        } finally {
            for(Future<Indexed> future:active) future.cancel(true); // 失败或中断时发出协作取消。
            executor.shutdownNow();
        }
    }
    static void check(boolean ok) { if(!ok) throw new AssertionError(); }
    public static void main(String[] args) throws Exception {
        check(Arrays.equals(map(new int[]{3,1,2},2,x->x*x),new int[]{9,1,4}));
        check(map(new int[0],3,x->x).length==0);
        check(Arrays.equals(map(new int[]{0,-2},1,x->x+1),new int[]{1,-1}));
        AtomicInteger active=new AtomicInteger(),peak=new AtomicInteger();
        int[] many=new int[100]; Arrays.fill(many,1);
        int[] output=map(many,3,x->{
            int now=active.incrementAndGet(); peak.accumulateAndGet(now,Math::max);
            try { return x*2; } finally { active.decrementAndGet(); }
        });
        check(Arrays.stream(output).sum()==200 && peak.get()<=3);
        try { map(new int[]{1,2,3},2,x->{if(x==2)throw new IllegalStateException();return x;}); throw new AssertionError(); }
        catch(ExecutionException expected) { check(expected.getCause() instanceof IllegalStateException); }
        Thread.currentThread().interrupt();
        try { map(new int[]{1},1,x->x); throw new AssertionError(); }
        catch(InterruptedException expected) { check(!Thread.currentThread().isInterrupted()); }
        finally { Thread.interrupted(); }
        System.out.println("PASS 6");
    }
}