RxJava 异步去抖动

【问题标题】：RxJava Asynchronous debouncingRxJava 异步去抖动
【发布时间】：2018-04-27 01:46:26
【问题描述】：

使用 RxJava 操作，如果上游发出新项目，是否可以“忽略”已在下游处理的元素？

例如

Observable.create(...)
  .flatMap(...) // 30 seconds to process (Asynchronous)
  .flatMap(...) // 30 seconds to process (Asynchronous)
  etc...

我需要实现的是，如果上游通知新项目，它将取消当前正在运行的下面流上的任何操作，即使下游在其他线程异步执行某些操作。

【问题讨论】：

标签： rx-java

【解决方案1】：

您可以使用switchMap() 来代替flatMap() 运算符。当switchMap() 接收到一个新值时，旧的 observable 会被取消订阅，而替换的 observable 会被订阅。

Observable.create( ... )
  .switchMap( value -> getObservable1( value )
                         .switchMap( value2 -> getObservable2( value2 ) )
  ...

如果您希望下游的观察者也被取消，则必须传播switchMap()。在上面的代码中，观察者链第一阶段的发射将取消订阅getObservable1( value ) 和getObservable2( value2 )。

【讨论】：