如何以非阻塞方式组合可观察对象？

Question

我有一组 Observable，它们分别检索不同的数据类型。我正在链接那些 Observable 以获取我想要的页面的所有数据。 事实是所有这些信息都是独立的，因此加载一个不应阻塞或干扰加载其他信息。这是我无法实现的。

这是我目前所做的一个示例：

  getAll(): Observable<any[]> {
    return this.getHome().pipe(
      tap((home: Home) => this.application = home),
      mergeMap((home: Home) =>
        combineLatest([
          this.getAssets(home.images).pipe(defaultIfEmpty([])),
          this.getStyling(home.styling).pipe(defaultIfEmpty({})),
          this.getCategories(home.categories).pipe(defaultIfEmpty([])),
        ])
      )
    );
  }

在这里，Observable 已成功链接，所以我得到了我想要的所有数据，但是 combineLatest() 在上一个请求完成后执行下一个请求，并且一旦一切完成，父级就会发出收集到的数据，这会造成延迟。 我一直在尝试使用 merge() 而不是 combineLatest() 以便收到的每个数据都会立即将其发送到父 Observable，但显然我不能这样做

那么，我怎样才能设法链接这些 Observable，以便子 Observable 中的每个提交都直接发送到父 Observable？

Answer 1

您需要forkJoin，它会返回一个Array，其中包含每个Observable 的结果（与forkJoin 参数的顺序相同。

  getAll(): Observable<any[]> {
    return this.getHome().pipe(
      tap((home: Home) => this.application = home),
      mergeMap((home: Home) =>
        forkJoin(
          this.getAssets(home.images).pipe(defaultIfEmpty([])),
          this.getStyling(home.styling).pipe(defaultIfEmpty({})),
          this.getCategories(home.categories).pipe(defaultIfEmpty([]))
        )
      )
    );
  }

Answer 2

如果它们是独立的，您可以独立处理它们并将它们发射到主题上。

  getAll(): Observable<any> {
    const subject = new Subject<any>();
    const handler = (res: any) => subject.next(res);

    this.getHome().pipe(
      tap((home: Home) => this.application = home),
      tap(handler),
      tap((home: Home) => {
        forkJoin(
          this.getAssets(home.images).pipe(defaultIfEmpty([]),tap(handler))
          this.getStyling(home.styling).pipe(defaultIfEmpty({}), tap(handler))                  
          this.getCategories(home.categories).pipe(defaultIfEmpty([]), tap(handler))
        ).subscribe(res => subject.complete())
      })
    ).subscribe();
    return subject.asObservable();
  }

使用这种方法，类型检查必须由消费者完成。

Subjects are documented here.

Answer 3

我一直在尝试使用 merge() 而不是 combineLatest() 以便接收到的每个数据都会立即将其发送到父 Observable，但显然我不能这样做

听起来您正在寻找准备好的部分数据。 combineLatest() 和 forkJoin() 都只会在它们可以填充长度与可观察对象数量相等的数组时发出数据。因此，您要么等待每个 observable 中的至少一个值，要么等待它们全部完成，但它们会发出一个数组，并且需要知道在其中放入什么。

我认为您希望在准备好后立即发出每条数据。

首先使用 merge() 从所有 3 个可观察对象发出值，但对于每个可观察对象，使用 map() 为给定类型（图像、样式、类别）创建对象键/值对。

最后，使用scan() 运算符将值聚合到单个对象中，该对象将在更新时发出每个属性。

流的消费者可以检查一个属性是否为null 以查看它是否已准备好。流将在完成之前发出 3 个值。

  getAll(): Observable<{home: Home, images: any[], styling: any[], categories: any[]}> {
    return this.getHome().pipe(
      tap((home: Home) => this.application = home),
      switchMap((home: Home) =>
        merge([
          this.getAssets(home.images).pipe(
             defaultIfEmpty([]),
             map(images => ({images}))
          ),
          this.getStyling(home.styling).pipe(
             defaultIfEmpty({}),
             map(styling => ({styling}))
          ),
          this.getCategories(home.categories).pipe(
             defaultIfEmpty([]),
             map(categories => ({categories})
          ),
        ]),
        scan((acc, next) => ({...acc, ...next}), {home, images: null, styling: null, categories: null})
      )
    );
  }

如果您想在所有属性都是null 的情况下发出第一个值，可以将startWith({}) 运算符添加到merge() 运算符。作为消费者的早期价值，因此他们知道阅读其他价值已经开始。