rx-java - 使用一个 Observable 为另一个 Observable 计时

Question

我们需要根据另一个热可观察值“计时”，例如:

 main:            --(1)-----(2)-----(3)-----(4)-------(5)----(6)------------|
 clock:           -------(X)------------(X)------(X)--------------(X)-------|
______________________________________________________________________________

 expected result: -------(1)(2)-----(3)-(3)-(4)--(4)--(5)----(6)--(6)-------|

更具体地说，每当“时钟”发出一个项目时，另一个可观察的最后一个项目就会再次发出。项目发射会被延迟，直到两个可观察对象都发射了至少一项。 现在，我们通过以下方式实现这一目标:

  <T, U> Observable<T> clock(Observable<T> main, Observable<U> clock) {
    return Observable.combineLatest(main, clock, (mainItem, clockItem) -> mainItem);
  }

这看起来有些愚蠢，因为我们正在应用转换，然后丢弃其中一个输入。此外，我们偶尔会遇到这种解决方案的背压问题。 似乎现有的 Rx 操作符可以执行此操作，但到目前为止我还没有找到正确的 API 方法来执行此操作。 有没有更好、更惯用的方法来使用 RxJava 解决这个问题？

Answer 1

我觉得不错。这并不是说时钟输入尚未使用，因为虽然该值尚未使用，但计时已使用。

为了处理热源上的背压，您可能需要应用 onBackPressureXXX 运算符之一。例如，如果您不想错过输出，请使用 .onBackPressureBuffer()。另一种策略是使用 .throttle 或 .sample。