RxJS：如何将两个流合并为一个，以一个流的速率触发事件，但使用来自其他流的最新值？

Question

我对 RX 很陌生。这是我要解决的问题的简单模型。这看起来很容易，但我很难找到合适的运算符（或以其他方式操作流）来解决它。

假设我们有两个流。一是频繁发出值；另一个远没有这样。我们希望每次第二个 observable 发出一个值时，获取另一个 observable 在该点发出的最新值，并对其进行处理。

无效示例：

let stream1 = Rx.Observable
    .interval(100);

let stream2 = Rx.Observable
    .interval(2000)
  .combineLatest(stream1, (stream2Value, stream1Value) => stream1Value)
  .do((stream1Value) => console.log('value:', stream1Value));

stream2.subscribe();

上面的 sn-p 的问题是它会等到第一个从 stream2 发出的值，然后会开始以 stream1 的频率发出一个事件流。我想要的是获得一个流，它将以 stream2 的速率触发事件，但会在 stream2 触发时发出 stream1 发出的最新值。听起来好像我需要 stream1 成为一个行为主体，以便我可以在 stream2 触发时访问它的最后一个值......但也许有一个更简单的解决方案？

Answer 1

您可以使用withLatestFrom 来做到这一点：

let stream1 = Rx.Observable
  .interval(100);

let stream2 = Rx.Observable
  .interval(2000)
  .withLatestFrom(stream1, (stream2Value, stream1Value) => stream1Value)
  .do((stream1Value) => console.log("value:", stream1Value));

stream2.subscribe();

<script src="https://npmcdn.com/@reactivex/rxjs@5.0.3/dist/global/Rx.min.js"></script>

此外，根据您的用例，您可能会发现 auditTime 运算符很有用。