如何在不丢失物品的情况下暂停Observable？

Question

我Observable每秒都会发出一个滴答声:

Observable.interval(0, 1, TimeUnit.SECONDS)
    .take(durationInSeconds + 1));

我想暂停这个Observable,以便它停止发出数字,并按需恢复它.

有一些陷阱:

• 根据ObservableJavadoc,interval操作员不支持背压
• 关于背压的RxJava wiki 有一个关于Callstack阻塞的部分作为背压的流量控制替代方案:

处理过度生产的Observable的另一种方法是阻止callstack(停止管理过度生产的Observable的线程).这具有违背Rx的"反应性"和非阻塞模型的缺点.但是,如果有问题的Observable位于可以安全阻止的线程上,这可能是一个可行的选择.目前,RxJava不会公开任何运营商来促进这一点.

有没有办法暂停一个intervalObservable？或者我应该用一些背压支持实现我自己的'滴答'Observable？

Answer 1

有很多方法可以做到这一点.例如,您仍然可以使用interval()和维护另外两种状态:例如布尔标志"暂停"和计数器.

public static final Observable<Long> pausableInterval(
  final AtomicBoolean paused, long initial, long interval, TimeUnit unit, Scheduler scheduler) {

  final AtomicLong counter = new AtomicLong();
  return Observable.interval(initial, interval, unit, scheduler)
      .filter(tick -> !paused.get())
      .map(tick -> counter.getAndIncrement()); 
}

然后你只需要在某处调用paused.set(true/false)来暂停/恢复

编辑2016-06-04

上面的解决方案存在一些问题.如果我们多次重用observable实例,它将从最后一次取消订阅的值开始.例如:

Observable<Long> o = pausableInterval(...)
List<Long> list1 = o.take(5).toList().toBlocking().single();
List<Long> list2 = o.take(5).toList().toBlocking().single();

虽然list1将是[0,1,2,3,4],但list2实际上是[5,6,7,8,9].如果不希望上述行为,则必须使观察者成为无状态.这可以通过scan()运算符来实现.修订版本可能如下所示:

  public static final Observable<Long> pausableInterval(final AtomicBoolean pause, final long initialDelay, 
      final long period, TimeUnit unit, Scheduler scheduler) {

    return Observable.interval(initialDelay, period, unit, scheduler)
        .filter(tick->!pause.get())
        .scan((acc,tick)->acc + 1);
  }

或者,如果您不希望依赖Java 8和lambdas,您可以使用Java 6+兼容代码执行此类操作:

https://github.com/ybayk/rxjava-recipes/blob/v0.0.2/src/main/java/yurgis/rxjava/recipes/RxRecipes.java#L343-L361