前端并发控制

本文讲解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  {    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  {        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  {      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  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)))  );}

   总结

  我们探讨了使用Promise，callback和RxJS的方式实现并发限制，每种方式中又介绍了三种代码思路，包括全部并发、分批并发以及限制并发。每种方法都有其适用场景和优缺点：

  • 「全部并发」适用于需要将请求分批次处理的场景，简单易懂，但可能不是最高效的方法。

  • 「分批并发」在保持一定并发度的同时，避免同时发出过多的请求，适用于需要控制资源消耗的场景。

  • 「限制并发」则结合了并发的高效性和结果顺序的一致性，适用于对结果顺序有要求的并发请求处理。

    通过选择合适的方法，我们可以在保证性能的同时，满足不同场景下对并发控制的需求。

    再次给大家安利RxJS，RxJS作为一个强大的响应式编程库，为我们提供了灵活而强大的工具来处理这些复杂的异步逻辑。

    文章的最后，我想引申下请求层的概念，在实际项目中，请求层的设计和实现对整个应用的性能和稳定性至关重要。一个健壮的请求层不仅能够处理基本的数据请求和响应，还能够应对各种复杂的网络环境和业务需求。以下是请求层可以处理的一些常见问题：

    • 「失败和错误处理」：优雅地处理请求失败和服务器返回的错误，提升用户体验。

    • 「失败重试」：在请求失败时自动重试，增加请求的成功率。

    • 「接口降级」：在服务不可用时，提供备选方案，保证应用的基本功能。

    • 「模拟接口」：在后端服务尚未开发完成时，模拟接口响应，加速前端开发。

    • 「模拟列表接口」：模拟分页、排序等列表操作，方便前端调试和测试。

    • 「接口聚合和竞态」：合并多个接口请求，减少网络开销；处理接口请求的竞态问题，确保数据的一致性。

    • 「逻辑聚合」：将多个资源的创建和更新等操作聚合成一个请求，简化前端逻辑。

    • 「控制并发数量」：限制同时进行的请求数量，避免过度消耗资源。

    • 「前端分页」：在前端进行数据分页，减轻后端压力。

    • 「超时设置」：为每个请求设置超时时间，防止长时间等待。

      通过在请求层中实现这些功能，我们可以使得前端应用更加稳定和可靠，同时也提升了用户的体验。因此，「加强请求层的建设」是每个前端项目都应该重视的一个方面。