Angular 上的批量订阅 |同时管理多个订阅

Question

想象一下这种情况...

我有一个大约+1k 的列表要在网格中显示。

用户可以单独更改网格值，或者他可以选择一些选项来更改所有项目中的相同属性。

因为我不知道当用户选择批量更改时，请求可以返回多少，例如更改今天的数据时间。我必须将这些更改存储在某些属性中，因此当他单击“确认”按钮时，我必须发出一些“GET”请求来了解我必须处理多少个请求，在我的 Angular 项目中进行属性更改并发送更新请求到后端。

我已经做了所有这些更改并且它正在工作，唯一的问题是浏览器会冻结，因为它必须同时处理许多订阅。

为了解决这个问题，我编写了下面的函数。因此，我制作了一个准备订阅的完整操作系统请求列表，并订阅了 X 个请求（由 QUANTITY_OPEN_REQUESTS 控制），这样浏览器就不会冻结。

当我处于最大并行请求时，我只是睡觉并在 Y 秒后重试（由 LOOP_INTERVAL_MS 控制）。

代码可以工作，但我感觉自己搞得一团糟，但我尝试在 Stackoverflow、Medium、Angular Docs、RXJS 文档中找到一些文章或帖子，但不知道还能做什么。

    async loopRequests(){
        let lsWaitingReqs: Array<ISubControl> = this.lsRequests.filter((x)=> x.status === StatusSubscription.WAITING)
        let lsRunningReqs: Array<ISubControl> = this.lsRequests.filter((x)=> x.status === StatusSubscription.STARTED)
        let lsDoneReqs: Array<ISubControl> = this.getFinishedRequests()
        let count: number = 0;
        this.loopExecuting = true

        // While my Done List has less itens then my requestList we must check if I can subscribe

        while(lsDoneReqs.length !== this.lsRequests.length){
            
            // While I have less running requests then max quantity, create another subscribe
            if (lsRunningReqs.length <= QUANTITY_OPEN_REQUESTS){
                if (lsWaitingReqs.length > 0){
                    const nextReq: IAltLoteTipoSubControl = lsWaitingReqs[0]
                    this.callProgressRefresh();

                    this.updateItemReqStack(nextReq, StatusSubscription.STARTED);
                    // The execRequest subscribes the request and when done it moves the StatusSubscription to Done
                    this.lsSubscriptions.push(this.execRequest(nextReq));
                } else {
                    await sleep(LOOP_INTERVAL_MS);  
                }
            } else {
                // If its at the maximum parallel requests, sleep to try again
                await sleep(LOOP_INTERVAL_MS);
            }
            
            // update the progressbar
            this.callProgressRefresh();

            // Update control lists
            lsRunningReqs = this.lsRequests.filter((x)=> x.status === StatusSubscription.STARTED);
            lsWaitingReqs = this.lsRequests.filter((x)=> x.status === StatusSubscription.WAITING);
            lsDoneReqs    = this.getFinishedRequests()
        }
        this.loopExecuting = false
    }

你们能留下一些提示或任何可以帮助我处理项目中所有这些混乱的东西吗？我主要关心的是性能、刷新率和诸如此类的事情。

即使你们可以推荐性能评估工具，我也会尝试，因为大多数时候我的座右铭是“让我们交付一些东西，然后我们考虑性能”，并且我想预测性能问题。

编辑¹：我忘记提及的一件事是每个项目都可以发送更新请求。

Obs：我知道在后端处理所有这些请求会容易得多，但我就是做不到......这更像是一个公司问题，我必须在我的 Angular 项目中提供解决方案。

Answer 1

Rxjs 可以非常轻松地处理并发。我将展示一个通用示例，因为我不知道您提出请求的确切代码。

makeReqeusts(updatedItems: Item[]) {
   from(updatedItems).pipe(
      mergeMap(item => this.http.post('/url', item), 5) // <-- concurrency = 5
   ).subscribe();
}

通过上面的代码，我们使用

from

创建一个单独发出每个数组元素的可观察对象。我们使用

mergeMap

来订阅每个项目的 http 调用。您可以将并发数作为第二个参数传递，这将限制活动请求的数量。

---

我看到您想表明此操作的进度。我们可以使用

map

运算符来计算项目完成的百分比：

makeReqeusts(updatedItems: Item[]) {
   from(updatedItems).pipe(
      mergeMap(item => this.http.post('/url', item), 5),
      map((_, i) => (i+1) / updatedItems.length),
      startWith(0)
   ).subscribe(
      percent => this.updateProgress(percent)
   );
}

这里

map

里面的代码每当http调用完成就会被触发。

map

接受参数，第一个是发射值（在本例中我们不需要），另一个是“发射索引”，它只是一个从0开始递增的整数。

我们使用

startWith(0)

立即发出 0。