Promise.all 并发限制

Promise.all 并发限制

1. 为什么要限制并发？

Promise.all 可以保证 promises 数组中所有 promise 都达到 resolve 状态，才执行 then 回调。

但如果 promises 数组每个对象都是 http 请求或复杂调用，数量达到几十万个：

• 瞬间发出几十万 http 请求（TCP 连接数不足可能造成等待）。
• 堆积无数调用栈导致内存溢出。

这时就需要对 Promise.all 做并发限制——每个时刻并发执行的 promise 数量固定，最终执行结果与 Promise.all 保持一致。

Promise 并非因为调用 Promise.all 才执行，而是在实例化时就开始执行。要做并发限制，只能从实例化上下手，把生成 promises 数组的控制权交给并发控制逻辑。

2. 实现思路

1. 通过第三方插件控制异步函数的并发执行。
2. 自行实现：定义 PromisePool 对象，先把并发池塞满，再通过 Promise.race 获取池中任务完成的信号，从任务队列里取新任务塞入。

3. 简单实现

class PromisePool {
  constructor(max, fn) {
    this.max = max;     // 最大并发量
    this.fn = fn;       // 自定义请求函数
    this.pool = [];     // 并发池
    this.urls = [];     // 剩余请求
  }

  start(urls) {
    this.urls = urls;
    while (this.pool.length < this.max) {
      const url = this.urls.shift();
      this.setTask(url);
    }
    return this.run(Promise.race(this.pool));
  }

  run(race) {
    race.then(() => {
      const url = this.urls.shift();
      this.setTask(url);
      return this.run(Promise.race(this.pool));
    });
  }

  setTask(url) {
    if (!url) return;
    const task = this.fn(url);
    this.pool.push(task);
    console.log(`${url} 开始，当前并发数：${this.pool.length}`);
    task.then(() => {
      this.pool.splice(this.pool.indexOf(task), 1);
      console.log(`${url} 结束，当前并发数：${this.pool.length}`);
    });
  }
}

const URLS = [
  'bytedance.com', 'tencent.com', 'alibaba.com',
  'microsoft.com', 'apple.com', 'hulu.com', 'amazon.com'
];

const requestFn = (url) => new Promise((resolve) => {
  setTimeout(() => resolve(`任务 ${url} 完成`), 1000);
}).then((res) => console.log('外部逻辑', res));

const pool = new PromisePool(5, requestFn);
pool.start(URLS);

4. 现代方案

• p-limit：Node 端小巧实用的并发限制库。
• p-queue：基于 Promise 的并发队列，支持优先级。
• async-pool：轻量实现。

const limit = require('p-limit');
const pool = limit(5);

await Promise.all(URLS.map((url) => pool(() => requestFn(url))));

来源整理自：我的有道云笔记