提高nodejs中promise的性能

我们先来看一个常见问题，假设我们有 N 条记录需要处理，或者例如，为每条记录发出 API 请求以获取数据。

通常情况下我们都是使用promise.all方法来实现这一需求：

// 记录
const data = [{}, {}, {}];// 处理所有的数据（获取所有记录的数据）
const results = await Promise.all(data.map(processRecord));

所以这里发生的是我们正在“并行”执行异步代码，所以它在时间线上可能看起来像这样。

在这里插入图片描述

可能我们都知道promise.all方法就是当最后一个promise也解决完才会有结果。正如上图所看到的，Promise.all()的整体运行时间与批次中最慢的一样长。所以我们的主线程基本上是“什么都不做”，正在等待最慢的请求完成。

为了解决这个问题，本文将讲解如何使用更好的优化这个方法。

promise pool

Promise Pool 的想法是充分利用 Node.js 的主线程的潜力。为了实现更好的利用率，我们需要密集地打包 API 调用（或任何其他异步任务），这样我们就不会等待最长时间的调用完成，而是在第一个调用完成后立即安排下一个调用。

在这里插入图片描述
通过这种方式，我们可以实现多种目标：

通过设置Promise Pool的并发数来控制我们服务的吞吐量；
通过设置Promise Pool的并发度来管理下游服务的负载；
提高我们应用程序的性能；减少 CPU 的空闲时间。

实现

让我们从需求开始。我们希望能够为 Promise Pool 提供一些记录和并发“请求”的最大数量以及可以进行处理的回调。处理后，我们希望收到每个单独记录的处理结果数组。

类似下面的方式：

const results = await PromisePool.for([]).withConcurrency(50).process(async (data) => {return result;});

我们需要跟踪当前正在处理的记录数、总共处理了多少条记录以及其他参数。

const { EventEmitter } = require("events");
class PromisePool {constructor(data,concurrency,processor,) {this.data = data;this.results = [];this.concurrency = concurrency;this.inFlightTasks = 0;this.processedEntries = 0;this.processor = processor;this.eventsEmitter = new EventEmitter();this.executionPromise = null;}
}

让我们添加一个结构方法以及公共函数来配置并发并指定处理器回调。

static for(data) {return new this(data, 10, () => {})}withConcurrency(concurrency) {this.concurrency = concurrency;return this;}process(processor) {this.processor = processor;return this._processRecords();}

我们已经建立了基础，现在我们需要实现 Promise Pool 的关键。这个想法很简单：

我们迭代数组中的每个记录并等待可用的坐位（基本上是我们拥有的并发坐位数量中的空闲坐位），如果我们有空闲坐位，我们将继续安排以下作业，除非所有坐位都被占用。如果所有座位都被占用，我们会等待至少一项工作完成。

 _waitAvailableSit() {if (this.inFlightTasks >= this.concurrency) {return new Promise((res) => {this.eventsEmitter.once(PromisePool.TASK_COMPLETED, res);});} else {return Promise.resolve();}}async _processRecord(data) {try {this.inFlightTasks++;const result = await this.processor(data, this);this.results.push(result);} catch (e) {} finally {this.inFlightTasks--;this.processedEntries++;this.eventsEmitter.emit(PromisePool.TASK_COMPLETED);if (this.inFlightTasks === 0 && this.processedEntries === this.data.length) {this.eventsEmitter.emit(PromisePool.DRAIN)}}}_processRecords() {if (this.executionPromise !== null) {return this.executionPromise;}this.executionPromise = new Promise(async (res, rej) => {try {for (const element of this.data) {await this._waitAvailableSit();this._processRecord(element);}this.eventsEmitter.once(PromisePool.DRAIN, () => res(this.results));} catch (e) {rej(e)}});return this.executionPromise;}

那么，一个可以处理并发“请求”的promise pool就写好了。

性能测试

我们将把处理器模拟为需要 150 到 1150 毫秒执行的函数。

const processor = (i) => {return new Promise((res, rej) => {console.log(`[${Date.now()}] 开始处理 item. i=${i}`);setTimeout(() => {console.log(`[${Date.now()}] 开始处理 item. i=${i}`);res(i);}, 150 + Math.random() * 1000);});
}

然后我们将同时处理 1000 条记录的数组，分成 50 条记录的块。首先，我们使用 Promise.all 方法对其进行测试。

async function main() {const timeStart = Date.now();const data = Array.from({ length: 1000 }).map((d,i) => i);while (data.length > 0) {const arr = data.splice(0, 50);await Promise.all(arr.map(processor));}const timeEnd = Date.now();console.log(`Promise.all=${timeEnd-timeStart}`);
}

然后是使用刚刚写的PromisePool

async function main() {const timeStart = Date.now();const data = Array.from({ length: 1000 }).map((d,i) => i);const promisePool = PromisePool.for(data).withConcurrency(50);await promisePool.process(processor);const timeEnd = Date.now();console.log(`PromisePool=${timeEnd-timeStart}`);
}