/**
 * 并發(fā)限制執(zhí)行器
 * <p>
 * 核心功能：控制任務并發(fā)度，采用滑動窗口策略
 * <p>
 * 知識點：
 * 1. ExecutorCompletionService - 按完成順序獲取任務結果
 * 2. ListenableFuture - Guava 可監(jiān)聽 Future
 * 3. SettableFuture - 可手動設置結果的 Future
 * 4. 滑動窗口并發(fā)控制 - 初始提交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 待執(zhí)行的任務列表
     * @return Future列表，順序與tasks一致
     */
    @SuppressWarnings("unchecked")
    public List<ListenableFuture<V>> submitAll(List<Callable<V>> tasks) {
        if (tasks.isEmpty()) {
            return ImmutableList.of();
        }
        // 創(chuàng)建結果占位符，數量等于任務數量
        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;
    }
    /**
     * 提交剩余任務（在獨立線程中執(zhí)行）
     * <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) {
            // 阻塞等待任意任務完成（釋放一個并發(fā)槽位）
            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個車次）
初始狀態(tài)：提交前3個車次到線程池
┌───┬───┬───┐ ┌───┬───┬───┬───┬───┬───┬───┐
│ G1│ G2│ G3│ │ G4│ G5│ G6│ G7│ G8│ G9│G10│
└───┴───┴───┘ └───┴───┴───┴───┴───┴───┴───┘
  執(zhí)行中          等待中
G1 完成 → 立即提交 G4
┌───┬───┬───┐ ┌───┬───┬───┬───┬───┬───┐
│   │ G2│ G3│ │ G4│ G5│ G6│ G7│ G8│ G9│G10│
└───┴───┴───┘ └───┴───┴───┴───┴───┴───┴───┘
      執(zhí)行中        等待中
G2 完成 → 立即提交 G5
...
維持窗口大小 = parallelism，任務完成后立即補充新任務