java - 如何在背压期间仅缓冲来自 rx.Observable 的最新发射

Question

我有一个rx.Observable将任务进度发送到onNext(). onNext()排放有时会发生得如此之快，以至于无法Observer跟上，从而导致背压。我想通过仅缓冲来自Observable.

例如：

• Observable发出1并Observer接收1。
• 虽然Observer仍在处理1，但Observable发出2、3和4。
• Observer完成处理1并开始处理4（排放2和3被丢弃）。

这似乎是在 Rx Observable 中处理进度的常见情况，因为您通常只关心使用最新进度信息更新 UI。但是我一直无法弄清楚如何做到这一点。

任何人都知道如何使用 RxJava 实现这一点？

Answer 1

onBackPressureLatest是你的朋友吗？:) http://reactivex.io/RxJava/javadoc/rx/Observable.html#onBackpressureLatest()

Answer 2

Observable.debounce听起来像你需要的。在下面的示例中，仅在每 200 毫秒窗口中来自可观察对象的最新发射将被发送给观察者。

observable
    .debounce(200, TimeUnit.MILLISECONDS)
    .subscribe(observer);