使用 Node 「多線程」處理高併發任務

摩爾定律

摩爾定律是由英特爾聯合創始人戈登·摩爾(Gordon Moore)在 1965 年提出的，即集成電路上可容納的元器件的數量每隔 18 至 24 個月就會增長一倍，性能也將提高一倍。也就是說，處理器（CPU）的性能每隔大約兩年就會翻一倍。

距離摩爾定律被提出到如今，已通過去了 50 多年。現在，隨着芯片組件的規模愈來愈接近單個原子的規模，要跟上摩爾定律的步伐變得愈來愈困難。

在 2019 年，英偉達 CEO 黃仁勳在 ECS 展會上說：「摩爾定律過去是每 5 年增加 10 倍，每 10 年增加 100 倍。而現在，摩爾定律每一年只能增加幾個百分點，每 10 年可能只有 2 倍。所以，摩爾定律結束了。」

單個處理器（CPU）的性能愈來愈接近瓶頸，想要突破這個瓶頸，則須要充分利用 多線程技術，讓單個或多個 CPU 能夠同時執行多個線程，更快的完成計算機任務。

Node 的多線程

咱們都知道，Javascript 是單線程語言，Nodejs 利用 Javascript 的特性，使用事件驅動模型，實現了異步 I/O，而異步 I/O 的背後就是多線程調度。

Node 異步 I/O 的實現能夠參考樸靈的 《深刻淺出 Node.js》

在 Go 語言中，能夠經過建立 Goroutine 來顯式調用一條新線程，而且經過環境變量 GOMAXPROCS 來控制最大併發數。

在 Node 中，沒有 API 能夠顯式建立新線程的 ，Node 實現了一些異步 I/O 的 API，例如 fs.readFile、http.request。這些異步 I/O 底層是調用了新線程執行異步任務，再利用事件驅動的模式來獲取執行結果。

服務端開發、工具開發可能都會須要使用到多線程開發。好比使用多線程處理複雜的爬蟲任務，用多線程來處理併發請求，使用多線程進行文件處理等等...

在咱們使用多線程時，必定要控制最大同時併發數。由於不控制最大併發數，可能會致使 文件描述符 耗盡引起的錯誤，帶寬不足引起的網絡錯誤、端口限制引起的錯誤等等。

在 Node 中並無用於控制最大併發數的 API 或者環境變量，因此接下來，咱們就用幾行簡單的代碼來實現。

代碼實現

咱們先假設下面的一個需求場景，我有一個爬蟲，須要天天爬取 100 篇掘金的文章，若是一篇一篇爬取的話太慢，一次爬取 100 篇會由於網絡鏈接數太多，致使不少請求直接失敗。

那咱們能夠來實現一下，每次請求 10 篇，分 10 次完成。這樣不只能夠把效率提高 10 倍，而且能夠穩定運行。

下面來看看單個請求任務，代碼實現以下：

const axios = require("axios");

async function singleRequest(article_id) {
  // 這裏咱們直接使用 axios 庫進行請求
  const reply = await axios.post(
    "https://api.juejin.cn/content_api/v1/article/detail",
    {
      article_id,
    }
  );

  return reply.data;
}

爲了方便演示，這裏咱們 100 次請求的都是同一個地址，咱們來建立 100 個請求任務，代碼實現以下：

// 請求任務列表
const requestFnList = new Array(100)
  .fill("6909002738705629198")
  .map((id) => () => singleRequest(id));

接下來，咱們來實現併發請求的方法。這個方法支持同時執行多個異步任務，而且能夠限制最大併發數。在任務池的一個任務執行完成後，新的異步任務會被推入繼續執行，以保證任務池的高利用率。代碼實現以下：

const chalk = require("chalk");
const { log } = require("console");

/**
 * 執行多個異步任務
 * @param {*} fnList 任務列表
 * @param {*} max 最大併發數限制
 * @param {*} taskName 任務名稱
 */
async function concurrentRun(fnList = [], max = 5, taskName = "未命名") {
  if (!fnList.length) return;

  log(chalk.blue(`開始執行多個異步任務，最大併發數： ${max}`));
  const replyList = []; // 收集任務執行結果
  const count = fnList.length; // 總任務數量
  const startTime = new Date().getTime(); // 記錄任務執行開始時間

  let current = 0;
  // 任務執行程序
  const schedule = async (index) => {
    return new Promise(async (resolve) => {
      const fn = fnList[index];
      if (!fn) return resolve();

      // 執行當前異步任務
      const reply = await fn();
      replyList[index] = reply;
      log(`${taskName} 事務進度 ${((++current / count) * 100).toFixed(2)}% `);

      // 執行完當前任務後，繼續執行任務池的剩餘任務
      await schedule(index + max);
      resolve();
    });
  };

  // 任務池執行程序
  const scheduleList = new Array(max)
    .fill(0)
    .map((_, index) => schedule(index));
  // 使用 Promise.all 批量執行
  const r = await Promise.all(scheduleList);

  const cost = (new Date().getTime() - startTime) / 1000;
  log(chalk.green(`執行完成，最大併發數： ${max}，耗時：${cost}s`));
  return replyList;
}

從上面的代碼能夠看出，使用 Node 進行併發請求的關鍵就是 Promise.all，Promise.all 能夠同時執行多個異步任務。

在上面的代碼中，建立了一個長度爲 max 最大併發數長度的數組，數組裏放了對應數量的異步任務。而後使用 Promise.all 同時執行這些異步任務，當單個異步任務執行完成時，會在任務池取出一個新的異步任務繼續執行，完成了效率最大化。

接下來，咱們用下面這段代碼進行執行測試（代碼實現以下）

(async () => {
  const requestFnList = new Array(100)
    .fill("6909002738705629198")
    .map((id) => () => singleRequest(id));

  const reply = await concurrentRun(requestFnList, 10, "請求掘金文章");
})();

最終執行結果以下圖所示：

到這裏，咱們的併發請求就完成啦！接下來咱們分別來測試一下不一樣併發的速度吧~ 首先是 1 個併發，也就是沒有併發（以下圖）

耗時 11.462 秒！當不使用併發時，任務耗時很是長，接下來咱們看看在其餘併發數的狀況下耗時（以下圖）

從上圖能夠看出，隨着咱們併發數的提升，任務執行速度愈來愈快！這就是高併發的優點，能夠在某些狀況下提高數倍乃至數十倍的效率！

咱們仔細看看上面的耗時會發現，隨着併發數的增長，耗時仍是會有一個閾值，不能徹底呈倍數增長。這是由於 Node 實際上並無爲每個任務開一個線程進行處理，而只是爲異步 I/O 任務開啓了新的線程。因此，Node 比較適合處理 I/O 密集型任務，並不適合 CPU（計算）密集型任務。

到這裏，咱們的使用 Node 「多線程」處理高併發任務就介紹完了。若是想要程序完善一點的話，還須要考慮到任務超時時間、容錯機制，你們感興趣的能夠本身實現一下。

參考資料

• 《深刻淺出 Nodejs》
• MBA 智庫百科
• 百度百科

最後一件事

若是您已經看到這裏了，但願您仍是點個贊再走吧~

您的點贊是對做者的最大鼓勵，也可讓更多人看到本篇文章！

若是以爲本文對您有幫助，請幫忙在 github 上點亮 star 鼓勵一下吧！