我是 RxJS 的新手，我正在尝试实现与 MS-Excel 完全一样的 observable 链。概念：假设 excel 有 5 列“姓名”、“年龄”、“性别”、“国家”、“邮政编码”。我们可以独立地对每一列应用过滤器，这也会影响其他列中显示的记录。
这里数据源接收数据后端服务，数据源将只有“addRecord”和“removeRecord”两个函数。
我在这里尝试实现的方式可以说我将创建 Observable 并附加到数据源，将其称为 OBS-1，这将从数据源接收数据。 OBS-1 可以有自己的过滤器。假设我将创建另一个 Observable OBS-2，它将接收数据 OBS-1（如果 OBS-1 中有任何过滤器，则过滤数据）。另一个 Observable 说 OBS-3 再次从 OBS-2 接收数据（如果在 OBS-2 中有过滤），依此类推。
如果 OBS-2 被销毁（取消订阅），OBS-3 将接收来自 OBS-1 的数据。
我们如何在 RxJs 中实现这一点？
【问题讨论】：
我认为您对 Rx 有一些误解。 Observables 没有过滤器，你不能“实时”添加和删除过滤器。 observables 也不会根据订阅者转发数据。
相反，您建立了一个调用链。你从一个可观察的源开始，比如一个来自addRecord 和一个来自removeRecord 事件。然后，您将这些 observable 链接起来，通过 Rx 中的各种 operators 形成新的 observable，并最终订阅最终的 observable。订阅将激活整个链，当源事件触发时，所有运算符都会触发，最终事件（如果未过滤）将到达subscribe。
你实际上可以做你在 Rx 中描述的事情。例如，使用switchMap 可以相对容易地更改可观察对象上的过滤器，该运算符可让您将序列投影到另一个序列上并每次切换到新序列。例如filterSource.switchMap(filterFunction => Obs-1.filter(filterFunction))。甚至比这更简单，您可以取消订阅第一个订阅并再次设置 Rx 链。然而，使用内置函数会留下很多杂耍状态。
但是，我强烈怀疑您实际上并不需要如此复杂的行为。您想要的内容可以像这样简单地归档：
var Src-1 = fromEvent(dataSource, 'addRecord') // create the first source
var Src-2 = fromEvent(dataSource, 'removeRecord') // and the other source
var Obs-1 = Src-1.combineLatest(Src-2) // combine both sources
.filter(e => someCondition(e)) // filter the source
var Obs-2 = Obs-1.mergeMap(e => someOtherCondition(e) ? Change(e) : Rx.Observable.of(e)) // on someOtherCondition, either transform the source with the `Change(e)` function. Or keep it unchanged with `of(e)`
var Obs-3 = Obs-2.filter(e => anotherCondition(e)) // Filter again
var sub = Obs-3.subscribe() // activate the sequence.
【讨论】：