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 = null，所以我使用 node.site_id 在 subscribe 方法中获取完整的“Site”对象。 但我有近 6000 次调用“siteService.getSiteById()”，并且出现如下错误：资源不足。

你有更好的方法吗，

谢谢

Answer 1

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

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 在 cmets 中所建议的，您可以使用 RxJS mergeMap 运算符来处理并发请求。但我建议您将其保持在 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() 返回 Observables 的事实