JS控制并发请求详解：Promise.all、allSettled与p-limit实现

标题

假设有100个请求需发送，请设计一个算法，利用Promise限制最大并发数为10，以处理这100个请求；

首先模拟生成100个请求：

// 请求列表
const requestList = [];

// 为了方便查看，i从1开始计数
for (let i = 1; i <= 100; i++) {
  requestList.push(
    () =>
      new Promise(resolve => {
        setTimeout(() => {
          console.log('done', i);
          resolve(i);
        }, Math.random() * 1000);
      }),
  );
}

Promise.all()方法

初看此题，多数人第一反应是 Promise.all，作为常规并发手段，具体实现如下：

const parallelRun = async max => {
  const requestSliceList = [];
  for (let i = 0; i < requestList.length; i += max) {
    requestSliceList.push(requestList.slice(i, i + max));
  }

  for (let i = 0; i < requestSliceList.length; i++) {
    const group = requestSliceList[i];
    try {
      const res = await Promise.all(group.map(fn => fn()));
      console.log('接口返回值为：', res);
    } catch (error) {
      console.error(error);
    }
  }
};

观察运行效果：

效果令人满意！！

每次并行执行10个请求，待这10个全部完成后，再启动下一批10个，完全符合需求；

然而面试官追问：若其中某个请求发生错误，结果会如何？

我：嗯.......，不太确定

面试官：回去等通知吧！

尽管回家等待通知，但这道面试题仍需深入理解，调整模拟请求代码以随机抛出错误，修改后如下：

// 请求列表
const requestList = [];

for (let i = 1; i <= 100; i++) {
  requestList.push(
    () =>
      new Promise((resolve, reject) => {
        setTimeout(() => {
          if (i === 92) {
            reject(new Error('出错了，出错请求：' + i));
          } else {
            console.log('done', i);
            resolve(i);
          }
        }, Math.random() * 1000);
      }),
  );
}

查看控制台输出结果：

一旦有一个请求失败，整个 Promise.all即告失败，且无返回值

组内任一请求失败便无法获取该组其他成员的返回值，对于无需返回值的情况尚可，但在实际业务中，返回值至关重要

我们可以容忍单个接口失败且无返回值，但不能接受因一个请求失败，导致同组其他9个请求也失去返回值

既然失败请求会中断 Promise.all，是否存在一种方法能避免被失败中断？

确实存在，那就是 Promise.allSettled！

Promise.allSettled()方法

首先参考权威的MDN文档介绍

Promise.allSettled()是promise并发处理的方法之一。当有多个不依赖彼此成功完成的异步任务，或需知晓每个promise的结果时，应使用此方法

简而言之：无论成功或失败，每个请求都会返回结果

使用 Promise.allSettled()替换原代码中的 Promise.all() ：

const parallelRun = async max => {
  const requestSliceList = [];
  for (let i = 0; i < requestList.length; i += max) {
    requestSliceList.push(requestList.slice(i, i + max));
  }

  for (let i = 0; i < requestSliceList.length; i++) {
    const group = requestSliceList[i];
    try {
      // 使用 allSettled 替换 all
      const res = await Promise.allSettled(group.map(fn => fn()));
      console.log('接口返回值为：', res);
    } catch (error) {
      console.error(error);
    }
  }
};

查看返回结果：

可见所有接口均有返回值，且返回值中准确记录了当前请求的成功或失败状态

表现不错，感觉 Promise.allSettled()似乎是最佳解决方案！

此时面试官再次提问：若有一个请求耗时极长，会发生什么？

回答：若有一个请求耗时极长，该组请求的返回就会变慢，进而阻塞后续接口的并发执行。

面试官：有没有什么方法可以解决这一问题？

我：嗯...... 不清楚......

面试官：回去等通知吧~~~

最佳解决方案

问题分析

使用 Promise.all()或 Promise.allSettled()，每次并发10个请求，虽能满足并发要求，但效率较低：若存在一个或多个慢接口，将引发以下两个问题：

包含慢接口的并发组返回缓慢，一个慢接口拖累了其他9个接口，得不偿失
原本可并发10个请求，但因一个慢接口导致该组其他9个并发槽位被浪费，致使处理这100个接口的总耗时大幅增加
慢接口组后续的并发组均被阻塞，处理速度更慢

解决方案

是否存在解决上述问题的方法？答案是肯定的：

可维护一个运行池和一个等待队列，运行池始终保持10个请求并发。当运行池中有一个请求完成时，便从等待队列中取出一个新请求放入运行池执行，从而保持运行池始终满负荷运行。即使存在慢接口，也不会阻塞后续接口入池

代码实现

// 运行池
const pool = new Set();

// 等待队列
const waitQueue = [];

/**
 * @description: 限制并发数量的请求
 * @param {*} reqFn：请求方法
 * @param {*} max：最大并发数
 */
const request = (reqFn, max) => {
  return new Promise((resolve, reject) => {
    // 判断运行吃是否已满
    const isFull = pool.size >= max;

    // 包装的新请求
    const newReqFn = () => {
      reqFn()
        .then(res => {
          resolve(res);
        })
        .catch(err => {
          reject(err);
        })
        .finally(() => {
          // 请求完成后，将该请求从运行池中删除
          pool.delete(newReqFn);
          // 从等待队列中取出一个新请求放入等待运行池执行
          const next = waitQueue.shift();
          if (next) {
            pool.add(next);
            next();
          }
        });
    };

    if (isFull) {
      // 如果运行池已满，则将新的请求放到等待队列中
      waitQueue.push(newReqFn);
    } else {
      // 如果运行池未满，则向运行池中添加一个新请求并执行该请求
      pool.add(newReqFn);
      newReqFn();
    }
  });
};

requestList.forEach(async item => {
  const res = await request(item, 10);
  console.log(res);
});

效果展示

可见，100个接口持续执行，没有任何等待或阻塞现象，完美实现！

其他优秀库

社区已有许多优秀的并发限制库，此处重点介绍 p-limit的安装方法：

npm install p-limit -S

使用方法：

import plimit from 'p-limit';


const limit = plimit(10);

requestList.forEach(async item => {
  const res = await limit(item);
  console.log(res);
});

运行效果与上述队列实现的效果一致。下面查看库源码（精简版）：

import Queue from 'yocto-queue';

export default function pLimit(concurrency) {
  const queue = new Queue();
  let activeCount = 0;

  const next = () => {
    activeCount--;

    if (queue.size > 0) {
      queue.dequeue()();
    }
  };

  const run = async (function_, resolve, arguments_) => {
    activeCount++;

    const result = (async () => function_(...arguments_))();

    resolve(result);

    try {
      await result;
    } catch {}

    next();
  };

  const enqueue = (function_, resolve, arguments_) => {
    queue.enqueue(run.bind(undefined, function_, resolve, arguments_));

    (async () => {
      // This function needs to wait until the next microtask before comparing
      // `activeCount` to `concurrency`, because `activeCount` is updated asynchronously
      // when the run function is dequeued and called. The comparison in the if-statement
      // needs to happen asynchronously as well to get an up-to-date value for `activeCount`.
      await Promise.resolve();

      if (activeCount < concurrency && queue.size > 0) {
        queue.dequeue()();
      }
    })();
  };

  const generator = (function_, ...arguments_) =>
    new Promise(resolve => {
      enqueue(function_, resolve, arguments_);
    });

  return generator;
}

短短60行代码便实现了一个功能强大的并发处理库，令人赞叹。下面分析其具体实现：

首先，p-limit库默认导出一个函数 pLimit，该函数接收一个数字，表示最大并发数
pLimit函数返回一个 generator生成器函数，该函数返回一个 Promise，并在其中调用了 enqueue函数
enqueue 函数主要将 run函数加入队列 queue中，随后判断 activeCount < concurrency && queue.size > 0，表示当前活动队列大小小于最大并发数且队列不为空，则需从队列中取出一个请求执行，即执行 run函数
run函数执行时需先将 activeCount加一，随后执行真正的请求函数 (async () => function_(...arguments_))()
随后等待请求完成 await result; 之后执行 next函数
next函数主要从队列中取出一个新请求执行，并将 activeCount 减一

总结

本文主要总结了限制100个请求并发的方法：

Promise.all() 最简单的控制并发方式，但请求出错会导致该组无返回值
Promise.allSettled() 解决了 Promise.all()的问题，却存在慢接口阻塞后续请求，且浪费其余并发槽位的问题
通过维护一个运行池，当运行池中有请求完成时便从等待队列中取一个请求入池执行，直到所有请求均入池
介绍了社区的 p-limit库的使用方法和实现原理

到此这篇关于JavaScript实现控制并发请求的方法详解的文章就介绍到这了，更多相关JavaScript控制并发请求内容请搜索本站以前的文章或继续浏览下面的相关文章希望大家以后多多支持本站！

JavaScript控制并发请求的三种方案：从Promise.all到p-limit

标题