前端并发控制

本文讲解Promise，callback，RxJS多种方式实现并发限制

1.Promise

目前来说，Promise是最通用的方案，一般我们最先想到Promise.all，当然最好是使用新出的Promise.allsettled。

下面简单介绍下二者的区别，假如存在某个请求失败时，all会整体失败，而allsettled只会让单个请求失败，对于大部分情况来说，allsettled的是更好的选择，因为allsettled更为灵活，一般来说面对这种情况，总共有三种处理方式，如下所示，all只能支持第一种，而allsettled三种都支持：

整体失败
最终结果，过滤失败的选项
将单个失败的保留，并渲染到UI中

方法1 全部并发

直接使用Promise.all是最简单的，代码如下，然后all并没有并发控制能力，一瞬间会将全部请求发出，从而造成前面提到的浏览器卡顿问题。

这里get函数我们使用setTimeout+随机时间来模拟请求，其返回promise实例。

function gets(ids, max) {return Promise.all(ids.map(id => get(id)))
}function get(id) {return new Promise((resolve) => {setTimeout(() => { resolve({ id }) }, Math.ceil(Math.random() * 5))});
}

方法2 分批并发

你可能会想到一种分批发送的办法，将请求按max数量分成N个组，每组并行发送，这需要结合递归和Promise.all，示例代码如下：

function gets(ids, max) {let index = 0;const result = [];function nextBatch() {const batch = ids.slice(index, index + max);index += max;return Promise.all(batch.map(get)).then((res) => {result.push(...res);if (index < ids.length) {return nextBatch();}return result;});}return nextBatch();
}

这种方法的优势在于实现相对简单，容易理解。但是它的缺点是，每一批请求中的最慢的请求会决定整个批次的完成时间，这可能会导致一些批次的其他请求早早完成后需要等待，从而降低整体的并发效率。

这种方法在业务中是不太能接受的，面试中的话，也只能勉强及格。

方法3 限制并发

一个更高效的思路是使用异步并发控制，而不是简单的批处理。这种方法可以在任何时刻都保持最大数量的并发请求，而不需要等待整个批次完成。这需要我们维护一个请求池，在每个请求完成时，将下一个请求添加到请求池中，示例代码如下：

gets函数返回一个promise，在请求全部完成后，promise变为fulfilled状态；内部采用递归，每个请求成功和失败后，发送下一个请求；在最下面先发送max个请求到请求池中。

function gets(ids, max) {return new Promise((resolve) => {const res = [];let loadcount = 0;let curIndex = 0;function load(id, index) {return get(id).then((data) => {loadcount++;if (loadcount === ids.length) {res[index] = data;resolve(res);} else {curIndex++;load(ids[curIndex]);}},(err) => {res[index] = err;loadcount++;curIndex++;load(ids[curIndex]);});}for (let i = 0; i < max && i < ids.length; i++) {curIndex = i;load(ids[i], i);}});
}

2.callback

在Promise之前，js中的异步都是基于回调函数的，比如 jQuery 的 ajax，Node.js 中的 http 模块等。

茴字有多种写法，下面我们挑战一下使用callback来解决这个问题。下面我们先把get函数改造一下，基于回调函数的get如下所示：

function get(id, success, error) {setTimeout(() => success({ id }), Math.ceil(Math.random() * 5))
}

gets函数的接口也要改成回调函数，如下所示：

function gets(ids, max, success, error) {}

回调函数也是基于上面的思路，把上面的代码稍加改动即可，将其中的Promise换成callback，示例如下：

还记得前面让你想其他思路吗，还有一种结合递归和异步函数的方法，在Promise下会比这种方法更简单，但其实还是这个思路更好，Promise和callback都可以使用。

function gets(ids, max, success, error) {const res = [];let loadcount = 0;let curIndex = 0;function load(id, index) {return get(id,(data) => {loadcount++;if (loadcount === ids.length) {res[index] = data;success(res);} else {curIndex++;load(ids[curIndex]);}},(err) => {res[index] = err;loadcount++;curIndex++;load(ids[curIndex]);});}for (let i = 0; i < max && i < ids.length; i++) {curIndex = i;load(ids[i], i);}
}

3.RxJS

看看RxJS，绝大部分人都不太了解RxJS，RxJS号称异步编程的lodash，对于这个问题，其代码实现会非常简单。

RxJS 是一个用于处理异步数据流的 JavaScript 库，它通过「可观察对象」（Observable）来代表随时间推移发出值的数据流。你可以使用一系列操作符（如 map、filter、merge 等）来处理这些数据流，并通过「订阅」（subscribe）来观察并执行相关操作。RxJS 使得处理复杂的异步逻辑变得简单而优雅，特别适合于实现并发控制等场景。

上面是RxJS的简介，相信看完了还是不理解，RxJS其实是比较难学的，建议大家阅读其他扩展资料。

下面先用RxJS改造我们的get函数，改造完如下所示，这需要用到Observable和observer，这些都是RxJS的概念，即便不知道其含义，看代码和Promise是比较相似的。

import { Observable } from 'RxJS';function get(id) {return new Observable((observer) => {setTimeout(() => {observer.next({ id });observer.complete();}, Math.ceil(Math.random() * 5));});
}

下面我们参考Promise中的思路，依次看看在RxJS中如何实现。

方法1 全部并发

在RxJS中和Promise.all类似的功能是forkJoin，这种方法最简单，代码如下所示，和Promise.all类似，这并不满足我们的需求。

import { forkJoin } from 'RxJS';function gets(ids) {const observables = ids.map(get);return forkJoin(observables);
}

方法2 分批并发

下面来看下如何实现分批并发，在Promise中我们使用递归+Promise.all来实现的。

在RxJS中，我们使用concatMap操作符来确保这些组是依次处理的，而不是同时处理。在处理每个组时，我们使用forkJoin来并行处理组内的所有请求。最后，我们使用reduce操作符来将所有组的结果合并成一个一维数组。

如果不理解RxJS，我们单纯看代码，可以看到RxJS代码的表现性更强，通过语义化的操作符串联，就完成了Promise中很多命令式的代码。

import { from, forkJoin } from 'RxJS';
import { concatMap, reduce } from 'RxJS/operators';function gets(ids, max) {// 将ids按max分组const groups = [];for (let i = 0; i < ids.length; i += max) {groups.push(ids.slice(i, i + max));}// 使用concatMap控制组之间的串行执行，并在每一组内使用forkJoin实现并行请求// 使用reduce来收集和合并所有组的结果return from(groups).pipe(concatMap((group) => forkJoin(group.map(get))),reduce((acc, results) => acc.concat(results), []));
}

方法3 限制并发

最后我们来看看RxJS如何实现限制并发，在这个实现中，我们使用mergeMap来控制并发，并使用一个Map对象来存储每个请求的结果，其中键是ID，值是请求结果。这样，我们可以在所有请求完成后，按照原始ID数组的顺序从Map中提取结果。

示例代码如下，控制并发是RxJS支持的功能，实现就是一个参数，非常简单

function gets(ids, max) {return from(ids).pipe(mergeMap((id) => get(id).pipe(map(result => ({ id, result }))), max),reduce((acc, { id, result }) => acc.set(id, result), new Map()),map(resMap => ids.map(id => resMap.get(id))));
}