6

考虑我有两个数据流，有没有办法合并它们并对这两个流之间的数据应用条件？例如

Stream A : A, B, C, D....
Stream B : -, A, -, -....
Composed : (A,-),(B,A),(C,-),(D,-)....

如何使用 rxjs 获得上面的组合流？我想对组合流应用条件以引发一些通知。也可以使用最后一个已知的非空数据，例如参见下面的组合流。

Stream A : A, B, C, D....
Stream B : 1, null, 2, null....
Composed : (A,1),(B,1),(C,2),(D,2)....

我刚刚开始使用反应流的想法，所以如果我误解了反应流的想法，请纠正我。

4

1 回答 1

12

有两个运营商可以为您的提议服务。

Zip：
处方拉链
RxJs 参考： https ://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/zip.md

CombineLatest：
Rx Combine最新
参考： https ://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/combinelatest.md

这些图像解释了两者之间的差异。现在您已经合并了 observable，您只需要使用where进行过滤，如果其中一个值为空，它将过滤。

不幸的是，两个运营商都无法获得您描述的这种行为：

Stream A : A, B, C, D, E....
Stream B : 1, null, 2, null, 3....
Composed : (A,1),(B,1),(C,2),(D,2)....

如果您使用 Zip 和 Where（之后过滤空值），结果将是：

Composed: (A,1),(C,2),(E,3)

如果您使用 Where（之前过滤空值）和 Zip，结果将是：

Composed: (A,1),(B,2),(C,3)

如果您使用 CombineLatest 将取决于事件在 Streams 中发生的顺序，当然，您将放置where运算符的位置，结果可能与您显示的不同，例如：

Stream A : A, B, C, D....
Stream B : 1, null, 2, null....
Composed : (A,1),(B,1),(C,1),(C,2),(D,2).... // OR
Composed : (A,1),(B,1),(B,2),(C,2),(D,2)....

除非您有更具体的要求，否则我认为我提到的选项之一就是您正在寻找的选项，请随时添加信息。

有几种方法可以组成 observable，其他没有提到的操作符是：

  • distinctUntilChanged，可以添加到组合的最后，使用键选择器功能来限制仅部分 zip 或最新值。
  • switch，用于将一个可观察对象组合到另一个对象中。
于 2015-02-03T04:59:21.447 回答