Java滑动窗口实现线程池并发度控制的方法详解

/**
 * 并发限制执行器
 * <p>
 * 核心功能：控制任务并发度，采用滑动窗口策略
 * <p>
 * 知识点：
 * 1. ExecutorCompletionService - 按完成顺序获取任务结果
 * 2. ListenableFuture - Guava 可监听 Future
 * 3. SettableFuture - 可手动设置结果的 Future
 * 4. 滑动窗口并发控制 - 初始提交N个，每完成一个立即补充下一个
 *
 * @author 桦说编程
 */
@Value
public class ConcurrentLimitExecutor<V> {
    ListeningExecutorService pool;
    int parallelism;
    BlockingQueue<Future<V>> q;
    ExecutorCompletionService<V> cs = new ExecutorCompletionService<>(pool, q);
    ListeningExecutorService submitter = MoreExecutors.listeningDecorator(
            Executors.newSingleThreadExecutor());
    /**
     * 提交所有任务，立即返回Future列表
     *
     * @param tasks 待执行的任务列表
     * @return Future列表，顺序与tasks一致
     */
    @SuppressWarnings("unchecked")
    public List<ListenableFuture<V>> submitAll(List<Callable<V>> tasks) {
        if (tasks.isEmpty()) {
            return ImmutableList.of();
        }
        // 创建结果占位符，数量等于任务数量
        List<SettableFuture<V>> result = IntStream.range(0, tasks.size())
                .mapToObj(__ -> SettableFuture.<V>create())
                .collect(toImmutableList());
        // 首批提交数量
        int start = Math.min(tasks.size(), parallelism);
        // 提交首批任务
        for (int i = 0; i < start; i++) {
            ListenableFuture<V> f = (ListenableFuture<V>) cs.submit(tasks.get(i));
            linkFuture(f, result.get(i));
        }
        // 异步提交剩余任务
        submitter.submit(() -> submitRemaining(tasks, result, start));
        return (List<ListenableFuture<V>>) (List<?>) result;
    }
    /**
     * 提交剩余任务（在独立线程中执行）
     * <p>
     * 流程：
     * 1. 等待任意任务完成（cs.take()）
     * 2. 提交下一个任务
     * 3. 将新任务的Future链接到result对应位置
     * 4. 重复直到所有任务提交完
     */
    @SneakyThrows
    private void submitRemaining(List<Callable<V>> tasks,
                                 List<SettableFuture<V>> result,
                                 int start) {
        int index = start;
        int size = tasks.size();
        while (index < size) {
            // 阻塞等待任意任务完成（释放一个并发槽位）
            cs.take();
            // 提交下一个任务
            ListenableFuture<V> f = (ListenableFuture<V>) cs.submit(tasks.get(index));
            // 链接结果到对应位置
            linkFuture(f, result.get(index));
            index++;
        }
    }
    private void linkFuture(ListenableFuture<V> from, SettableFuture<V> to) {
        to.setFuture(from);
    }
}

ExecutorService pool = Executors.newFixedThreadPool(10);
ExecutorCompletionService<Train> cs = new ExecutorCompletionService<>(pool);
// 提交任务
cs.submit(() -> queryTrain("G1"));
cs.submit(() -> queryTrain("G2"));
// 阻塞获取最先完成的结果（而不是提交顺序）
Future<Train> first = cs.take();  // 谁先完成就拿到谁
Future<Train> second = cs.take();

┌─────────────────────────────────────────────┐
│      ExecutorCompletionService              │
├─────────────────────────────────────────────┤
│  - executor: ExecutorService                │
│  - completionQueue: BlockingQueue<Future>   │
├─────────────────────────────────────────────┤
│  + submit(Callable): Future                 │
│  + take(): Future  // 阻塞获取已完成的      │
│  + poll(): Future  // 非阻塞获取            │
└─────────────────────────────────────────────┘
         │
         │ 任务完成时自动放入队列
         ▼
┌─────────────────────────────────────────────┐
│     BlockingQueue<Future<V>>                │
│  ┌──────┐  ┌──────┐  ┌──────┐              │
│  │Future│  │Future│  │Future│  ...         │
│  └──────┘  └──────┘  └──────┘              │
│   已完成    已完成    已完成                 │
└─────────────────────────────────────────────┘

// 标准Future：只能轮询或阻塞
Future<Train> future = executor.submit(() -> queryTrain("G1"));
Train result = future.get(); // 阻塞
// ListenableFuture：注册回调
ListeningExecutorService executor = MoreExecutors.listeningDecorator(pool);
ListenableFuture<Train> future = executor.submit(() -> queryTrain("G1"));
// 异步回调（不阻塞）
future.addListener(() -> {
    System.out.println("G1查询完成！");
}, directExecutor());
// 或使用Futures工具类
Futures.addCallback(future, new FutureCallback<Train>() {
    public void onSuccess(Train result) { ... }
    public void onFailure(Throwable t) { ... }
}, executor);

SettableFuture<Train> future = SettableFuture.create();
// 在其他线程设置结果
future.set(train);                       // 正常完成
future.setException(new TimeoutException());  // 异常完成
future.setFuture(otherFuture);           // 链接另一个Future
// 调用方正常使用
Train result = future.get();

假设 parallelism=3, tasks=10（查询10个车次）
初始状态：提交前3个车次到线程池
┌───┬───┬───┐ ┌───┬───┬───┬───┬───┬───┬───┐
│ G1│ G2│ G3│ │ G4│ G5│ G6│ G7│ G8│ G9│G10│
└───┴───┴───┘ └───┴───┴───┴───┴───┴───┴───┘
  执行中          等待中
G1 完成 → 立即提交 G4
┌───┬───┬───┐ ┌───┬───┬───┬───┬───┬───┐
│   │ G2│ G3│ │ G4│ G5│ G6│ G7│ G8│ G9│G10│
└───┴───┴───┘ └───┴───┴───┴───┴───┴───┴───┘
      执行中        等待中
G2 完成 → 立即提交 G5
...
维持窗口大小 = parallelism，任务完成后立即补充新任务