All files / modules/book-generator/utils async-pool.ts

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
/**
 * 并发控制池 - 限制最大并发数的异步任务调度器
 *
 * 用于内容并行生成：控制同时进行的 LLM 调用数量，
 * 避免打爆 API 限流同时最大化吞吐量。
 */
 
export class AsyncPool {
  private concurrency: number;
 
  /**
   * @param concurrency 最大并发数，默认 8
   */
  constructor(concurrency: number = 8) {
    this.concurrency = Math.max(1, concurrency);
  }
 
  /**
   * 并发执行一组任务，结果按原始顺序返回。
   * 任意一个任务失败会导致整体 reject。
   */
  async runAll<T>(tasks: Array<{ fn: () => Promise<T> }>): Promise<T[]> {
    const results: T[] = new Array(tasks.length);
    let nextIndex = 0;
 
    const worker = async (): Promise<void> => {
      while (nextIndex < tasks.length) {
        const idx = nextIndex++;
        results[idx] = await tasks[idx].fn();
      }
    };
 
    const workers = Array.from({ length: Math.min(this.concurrency, tasks.length) }, () => worker());
    await Promise.all(workers);
    return results;
  }
 
  /**
   * 顺序执行一组任务（按给定顺序逐个执行，无并发）。
   * 适用于需要严格顺序的场景。
   */
  async runSequential<T>(tasks: Array<{ fn: () => Promise<T> }>): Promise<T[]> {
    const results: T[] = [];
    for (const task of tasks) {
      results.push(await task.fn());
    }
    return results;
  }
}