Angular 9订阅内部订阅

Question

我正在使用Angular 9和rxjs 6，并尝试获取节点列表：

NodeService.ts

public getList(): Observable<Nodes[]> { return this.http.get<Nodes[]>(URL); }

节点实体：

export class Nodes {
     public id: number;
     public name: string;
     public site_id: number;
     public site: Site;
}

nodes.component.ts：

public getAllNodes() {
  this.dataSource.data = [];
  this.nodeService.getList()
    .subscribe(
      nodes => {
        nodes.map(
          (node) => {
            this.siteService.getSiteById(node.site_id).subscribe(site => {
              node.site = site;
            })
          });
        this.dataSource.data = nodes;
        console.log(nodes);
      },
      error => {
        console.log(error);
      },
      () => {
      }
    );
}

默认情况下，node.site = =>，所以我使用node.site_id在subscribe方法中获取完整的"Site“对象。但是我有将近6000次"siteService.getSiteById()“调用，并且我遇到了类似这样的错误:资源不足。

你有没有更好的办法，

谢谢

Answer 1

如果您需要进行6000个单独的HTTP调用，您可能需要修改后端以在单个请求中返回所有信息。同时，您可以尝试使用带有from和of函数的concatMap的RxJS switchMap operator来按顺序执行请求。

import { of, from } from 'rxjs';
import { switchMap, concatMap, finalize } from 'rxjs/operators';

public getAllNodes() {
  this.dataSource.data = [];
  let sites = [];
  this.nodeService.getList().pipe(
    switchMap(nodes => {
      let requests = [];
      this.dataSource.data = nodes;
      nodes.forEach(node => requests.push(this.siteService.getSiteById(node.site_id)));
      return from(requests);
    }),
    concatMap(request => of(request)),
    finalize(() => {
      this.dataSource.data.forEach((node, index) => node.site = sites[index]);
    })
  ).subscribe(
    site => {
      sites.push(site);
    },
    error => {
      // handle error
    }
  );
}

现在6000个请求将按顺序完成，所以它太慢了。

更新:使用mergeMap的并发请求

正如Will Alexander在评论中所建议的，您可以使用RxJS mergeMap operator来实现并发请求。但我建议您将其保持在6以下，因为大多数浏览器对对唯一域的并发请求数量有硬性限制。

这样你就可以替换

concatMap(request => of(request))

使用

mergeMap(request => of(request), null, 5)   // <-- 5 concurrent requests

Answer 2

大致是这样的：

public getAllNodes() {
  return this.nodeService.getList().pipe(

    // emit all elements of the array separately
    map(nodes => from(nodes)),
    
    // subscribe to each Observable in turn
    // allow 5 concurrent requests
    mergeMap(node$ => node$, null, 5).pipe(

      // for each node, get the corresponding site, and emit the full object
      concatMap(
        node => this.siteService.getSiteById(node.site_id),
        (node, site) => { ...node, site })
    ),

    // when Observable completes, emit all emissions as an Array
    toArray()
  );
}

更新:考虑到getList()返回可观测对象的事实