RxJava 合并去抖动和未去抖动的 observables答案

【问题标题】：RxJava merge debounced and not debounced observablesRxJava 合并去抖动和未去抖动的 observables
【发布时间】：2016-09-21 17:18:22
【问题描述】：

我有两个可观察的：

Observable O(open): 在 textview 中打开一些内容的文件

Observable E(edit)：在 textview 中编辑的文件内容

我想对 E observable 进行去抖动，并将其与 O observable 合并。

obs = Observable.merge(E.debounce(2000, TimeUnit.MILLISECONDS) , O)
                .subscribe(content->System.out.println("new content: " + content))

问题是，如果 E 发出事件 E1 并且紧接着 O 发出 O1 事件，我们有输出：

new content: O1
new content: E1 // this output is rebundant (cuz we already have newer content O1)

这是正在发生的事情的图表：

如何从 debounced observable 中去掉这个过多的旧事件？

【问题讨论】：

为什么不将debounce 运算符移出merge 工作？
@TassosBassoukos ，编辑事件（E1）非常频繁（实际上它是一个键类型事件）所以我想去抖动它。但是，当我打开文件时，我想立即收到一个 O1 事件而不反弹。
您可以使用“版本控制”，例如时间。映射您的 Observable O 和 Observable E 事件（去抖动之前）以包含内容和时间。然后，您可以比较 Observable E 事件是否比 Observable O 事件“更老”。您可能需要切换到 Observable.combineLatest 才能实现。

标签： java rx-java reactivex

【解决方案1】：

你可以试试

Observable.merge(O, O.switchMap(o -> E.debounce()))
          .subscribe()

switchMap 的行为与 flatMap 非常相似，不同之处在于每当有新项目由源 Observable 发出，它将取消订阅并停止镜像从先前发射的项目，并开始仅镜像当前项目。

【讨论】：

【解决方案2】：

我看到两个主要选项。一种是使用时间戳，这很容易，但存在理论上的竞争条件（但推测非常不可能），另一种选择是使用与每个文件打开相关联的唯一标识符，以及从编辑到该打开文件的文本发出的事件是附带文件打开的标识符。

使用时间戳：

obs = Observable.defer(() -> {
  AtomicBoolean first = new AtomicBoolean(true);
  e.timestamp()
   .debounce(2000, TimeUnit.MILLISECONDS))
   .mergeWith(o.timestamp())
   .buffer(2,1)
   .flatMap(list -> {
     Observable<Object> start;
     if (first.compareAndSet(true, false)) 
         start = Observable.just(list.get(0).getValue ());
     else 
         start = Observable.empty();
     if (list.size() == 1) 
         return start;
     else {
       Timestamped<Object> a = list.get(0);
       Timestamped<Object> b = list.get(1);
       if (a.getTimestampMillis() <= b.getTimestampMillis()) 
         return start.concatWith(Observable.just(b.getValue ()));
       else 
         return start; 
     }
   })
});

我怀疑带时间戳的版本就足够了。

【讨论】：