JavaScript實現(xiàn)控制并發(fā)請求的方法詳解

更新時間：2024年03月27日 16:20:35 作者：前端探險家克魯

這篇文章主要為大家詳細介紹了如果有100個請求,那么如何使用JavaScript實現(xiàn)控制并發(fā)請求,感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一

題目

現(xiàn)有100個請求需要發(fā)送，請設(shè)計一個算法，使用Promise來控制并發(fā)（并發(fā)數(shù)量最大為10），來完成100個請求；

首先先模擬下 100 個請求：

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

// 為了方便查看，i從1開始計數(shù)
for (let i = 1; i <= 100; i++) {
  requestList.push(
    () =>
      new Promise(resolve => {
        setTimeout(() => {
          console.log('done', i);
          resolve(i);
        }, Math.random() * 1000);
      }),
  );
}

Promise.all()

初次看到這個問題，相信大部分同學(xué)第一個想到的肯定是 Promise.all，因為它是最常見的并發(fā)請求方式，下面來實現(xiàn)一下：

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);
    }
  }
};

看下效果：

效果不錯?。?/p>

每次都是并發(fā) 10 個請求，當(dāng)這 10 個請求都完成返回時，繼續(xù)下一個 10 個請求，完美實現(xiàn)需求；

可是此時面試官問：如果這里邊有一個請求失敗了會怎樣？

我：額.......，不確定

面試官：回去等通知吧！

雖然回家等通知了，但這道面試題還是得弄明白，修改下模擬請求，使其隨機產(chǎn)生一個錯誤，修改如下：

// 請求列表
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);
      }),
  );
}

控制臺看下運行結(jié)果：

有一個請求失敗了，這個 Promise.all就失敗了，沒有返回值

一組中一個請求失敗就無法獲取改組其他成員的返回值，這對于不需要判斷返回值的情況倒是可以，但是實際業(yè)務(wù)中，返回值是一個很重要的數(shù)據(jù)

我們可以接受某個接口失敗了沒有返回值，但是無法接受一個請求失敗了，跟它同組的其他 9 個請求也沒有返回值

既然，失敗的請求會打斷 Promise.all，那有沒有一種方法可以不被失敗打斷呢？

還真有，它就是 Promise.allSettled!

Promise.allSettled()

先來看下權(quán)威的 MDN 的介紹

Promise.allSettled() 方法是 promise 并發(fā)方法之一。在你有多個不依賴于彼此成功完成的異步任務(wù)時，或者你總是想知道每個 promise 的結(jié)果時，使用 Promise.allSettled()

簡單說就是：每個請求都會返回結(jié)果，不管失敗還是成功

使用 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);
    }
  }
};

看下返回結(jié)果：

可以看到，接口全部正常有返回值，返回值中會正常記錄當(dāng)前請求時成功還是失敗

不錯哦，感覺 Promise.allSettled()就是最優(yōu)解了！

此時面試官又問：那如果有一個請求非常耗時，會出現(xiàn)什么情況？

答：有一個請求非常耗時，那組的請求返回就會很慢，會阻塞了后續(xù)的接口并發(fā)。

面試官：有沒有什么方法可以解決這個問題？

我：額...... 不知道......

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

最優(yōu)解

分析問題

使用 Promise.all()或是 Promise.allSettled()，每次并發(fā) 10 個請求，確實可以滿足并發(fā)要求，但是效率較低：如果存在一個或多個慢接口，那么會出現(xiàn)以下兩個問題：

有慢接口的并發(fā)組返回會很慢，一個慢接口拖慢了其他 9 個接口，得不償失
本來我們是可以并發(fā) 10 個請求的，但是一個慢接口導(dǎo)致該組的其他 9 個并發(fā)位置都被浪費了，這會導(dǎo)致這 100 個接口的并發(fā)時間被無情拉長
慢接口組后續(xù)的并發(fā)組都被阻塞了，更慢了

解決方法

有沒有辦法解決上述問題呢，答案是肯定的：

可以維護一個運行池和一個等待隊列，運行池始終保持 10 個請求并發(fā)，當(dāng)運行池中有一個請求完成時，就從等待隊列中拿出一個新請求放到運行池中運行，這樣就可以保持運行池始終是滿負(fù)荷運行，即使有一個慢接口，也不會阻塞后續(xù)的接口入池

代碼實現(xiàn)

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

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

/**
 * @description: 限制并發(fā)數(shù)量的請求
 * @param {*} reqFn：請求方法
 * @param {*} max：最大并發(fā)數(shù)
 */
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);
          // 從等待隊列中取出一個新請求放入等待運行池執(zhí)行
          const next = waitQueue.shift();
          if (next) {
            pool.add(next);
            next();
          }
        });
    };

    if (isFull) {
      // 如果運行池已滿，則將新的請求放到等待隊列中
      waitQueue.push(newReqFn);
    } else {
      // 如果運行池未滿，則向運行池中添加一個新請求并執(zhí)行該請求
      pool.add(newReqFn);
      newReqFn();
    }
  });
};

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

效果

可以看到，100 個接口不斷執(zhí)行，并沒有任何等待或是被阻塞的現(xiàn)象，完美！

其他優(yōu)秀庫

社區(qū)已有很多優(yōu)秀的并發(fā)限制庫，這里重點介紹下 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 行代碼就實現(xiàn)了一個功能強大的并發(fā)處理庫，真是厲害，下面分析下具體實現(xiàn)：

首先 p-limit 庫默認(rèn)導(dǎo)出一個函數(shù)pLimit，該函數(shù)接收一個數(shù)字，表示最大并發(fā)數(shù)
pLimit函數(shù)函數(shù)返回一個 generator函數(shù)，該函數(shù)返回一個 Promise，并且其中調(diào)用了 enqueue函數(shù)
enqueue 函數(shù)主要是將 run函數(shù)加入隊列 queue中，之后判斷下 activeCount < concurrency && queue.size > 0，表示當(dāng)前隊列大小小于最大并發(fā)數(shù)且隊列不為空，則需要從隊列中取出一個請求執(zhí)行，即執(zhí)行run函數(shù)
run函數(shù)執(zhí)行時需要先將 activeCount加一，之后執(zhí)行真正的請求函數(shù) (async () => function_(...arguments_))()
之后等待請求完成 await result; 之后執(zhí)行 next函數(shù)
next函數(shù)主要從隊列中取出一個新請求執(zhí)行并將activeCount 減一

總結(jié)

本文主要總結(jié)了 100 個請求限制并發(fā)的方法：

Promise.all() 最簡單的控制并發(fā)，但是請求出錯會導(dǎo)致該組無返回值
Promise.allSettled() 解決了Promise.all()的問題，但是卻存在慢接口阻塞后續(xù)請求，且浪費其余并發(fā)位置的問題
通過維護一個運行池，當(dāng)運行池中有請求完成時便從等待隊列中取一個心情求入池執(zhí)行，直到所有的請求都入池
介紹了社區(qū)的 p-limit庫的使用方法和實現(xiàn)原理

到此這篇關(guān)于JavaScript實現(xiàn)控制并發(fā)請求的方法詳解的文章就介紹到這了,更多相關(guān)JavaScript控制并發(fā)請求內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: