可观察的 RxJS 动态源答案

【问题标题】：RxJS dynamic source of observables可观察的 RxJS 动态源
【发布时间】：2020-01-26 08:38:56
【问题描述】：

我想创建一个包含许多 observable 的 Observable（合并它们）。这可以通过merge(...arrayOfObservables) 来实现。问题是这个数组有时会被改变，并且 observable 也应该订阅新的 observables。

【问题讨论】：

标签： rxjs rxjs6

【解决方案1】：

您可以将新的 Observable 推送到一个数组，然后发出该数组并使用 switchMap 订阅它们。

import { of, merge, BehaviorSubject } from 'rxjs'; 
import { switchMap } from 'rxjs/operators';

const s = new BehaviorSubject([of(1), of(2), of(3)]);

s.pipe(
  switchMap(array => merge(...array)),
).subscribe(x => console.log(x));

s.next([...s.getValue(), of(4)]);

现场演示：https://stackblitz.com/edit/rxjs-vmcqs9

【讨论】：

这就像一个魅力！唯一我不确定当我添加/删除 observable 时，Subject 是否不会错过其他 observable 的任何数据？
switchMap 将取消订阅，因此您可以改用 concatMap 或 mergeMap。
我忘了提到的是 observables 很热。刚刚用switchMap 对其进行了测试，效果很好。似乎没有跳过任何数据。但是在这种情况下是否有可能丢失数据？