JavaScript 异步并发控制实战¶

// utils/AsyncConcurrencyPool.ts

export interface TaskOptions {
  priority?: number; // 优先级：数字越大越优先执行
  retries?: number; // 失败重试次数
  retryDelayMs?: number; // 重试基础延迟时间
  signal?: AbortSignal; // 支持外部取消
}

type TaskFunction<T> = (signal?: AbortSignal) => Promise<T>;

interface QueueItem<T> {
  fn: TaskFunction<T>;
  options: TaskOptions;
  resolve: (value: T | PromiseLike<T>) => void;
  reject: (reason?: any) => void;
}

export class AsyncConcurrencyPool {
  private concurrency: number;
  private runningCount = 0;
  private queue: QueueItem<any>[] = [];

  constructor(concurrency = 5) {
    if (concurrency < 1) throw new Error('Concurrency must be at least 1');
    this.concurrency = concurrency;
  }

  /**
   * 提交任务到并发池
   */
  public add<T>(fn: TaskFunction<T>, options: TaskOptions = {}): Promise<T> {
    return new Promise<T>((resolve, reject) => {
      const item: QueueItem<T> = {
        fn,
        options: {
          priority: 0,
          retries: 0,
          retryDelayMs: 1000,
          ...options,
        },
        resolve,
        reject,
      };

      // 快速检查外部信号是否已提前取消
      if (item.options.signal?.aborted) {
        return reject(new DOMException('Task was aborted before execution', 'AbortError'));
      }

      // 按优先级插入等待队列 (降序排队)
      const insertIndex = this.queue.findIndex(
        (queued) => (queued.options.priority ?? 0) < (item.options.priority ?? 0)
      );

      if (insertIndex === -1) {
        this.queue.push(item);
      } else {
        this.queue.splice(insertIndex, 0, item);
      }

      this.runNext();
    });
  }

  private async runNext(): Promise<void> {
    if (this.runningCount >= this.concurrency || this.queue.length === 0) {
      return;
    }

    const item = this.queue.shift();
    if (!item) return;

    this.runningCount++;

    const executeWithRetry = async (attempt = 0): Promise<any> => {
      // 检查执行期间的取消信号
      if (item.options.signal?.aborted) {
        throw new DOMException('Task aborted by caller', 'AbortError');
      }

      try {
        return await item.fn(item.options.signal);
      } catch (err) {
        const maxRetries = item.options.retries ?? 0;
        if (attempt < maxRetries && !item.options.signal?.aborted) {
          // 指数退避算法等待重试
          const delay = (item.options.retryDelayMs ?? 1000) * Math.pow(2, attempt);
          await new Promise((r) => setTimeout(r, delay));
          return executeWithRetry(attempt + 1);
        }
        throw err;
      }
    };

    try {
      const result = await executeWithRetry();
      item.resolve(result);
    } catch (error) {
      item.reject(error);
    } finally {
      this.runningCount--;
      // 递归触发下一个排队任务
      this.runNext();
    }
  }

  public get activeCount(): number {
    return this.runningCount;
  }

  public get pendingCount(): number {
    return this.queue.length;
  }
}

// 模拟 100 个大文件切片批量上传，控制最大并发 4，重要元数据切片优先
async function uploadFileChunks(chunks: Blob[]) {
  const pool = new AsyncConcurrencyPool(4);
  const controller = new AbortController();

  const uploadPromises = chunks.map((chunk, index) => {
    // 假设索引 0 的分片包含文件 Meta 信息，赋予最高优先级 10
    const isMetaChunk = index === 0;

    return pool.add(
      async (signal) => {
        const formData = new FormData();
        formData.append('chunkIndex', String(index));
        formData.append('data', chunk);

        const response = await fetch('/api/upload/chunk', {
          method: 'POST',
          body: formData,
          signal,
        });

        if (!response.ok) throw new Error(`Chunk ${index} failed with ${response.status}`);
        return response.json();
      },
      {
        priority: isMetaChunk ? 10 : 0,
        retries: 3, // 单片失败最多重试 3 次
        retryDelayMs: 500,
        signal: controller.signal,
      }
    );
  });

  try {
    const results = await Promise.all(uploadPromises);
    console.log('所有分片上传完毕，准备触发服务端合片:', results);
  } catch (err) {
    if ((err as Error).name === 'AbortError') {
      console.warn('用户主动取消了上传任务');
    } else {
      console.error('分片上传遇到不可逆故障:', err);
      // 发生严重错误时，立即中断后续所有尚未执行的网络请求
      controller.abort();
    }
  }
}