立即发送第一个项目,"debounce"以下项目

Question

考虑以下用例:

• 需要尽快交付第一件物品
• 需要在1秒超时后对事件进行去抖动

我最终实现了自定义运算符,OperatorDebounceWithTime然后像这样使用它

.lift(new CustomOperatorDebounceWithTime<>(1, TimeUnit.SECONDS, Schedulers.computation()))

CustomOperatorDebounceWithTime立即发送第一个项目,然后使用OperatorDebounceWithTime操作员的逻辑去除后期项目.

是否有更简单的方法来实现所描述的行为？让我们跳过compose运算符,但它没有解决问题.我正在寻找一种方法来实现这一点,而无需实现自定义运算符.

Answer 1

更新:
来自@ lopar的评论更好的方法是:

Observable.from(items).publish(publishedItems -> publishedItems.limit(1).concatWith(publishedItems.skip(1).debounce(1, TimeUnit.SECONDS)))

会这样的工作:

String[] items = {"one", "two", "three", "four", "five", "six", "seven", "eight"};
Observable<String> myObservable = Observable.from(items);
Observable.concat(myObservable.first(), myObservable.skip(1).debounce(1, TimeUnit.SECONDS))
    .subscribe(s -> System.out.println(s));

Answer 2

@LortRaydenMK和@lopar的答案是最好的,但我想提出一些其他的建议,以防它碰巧更适合你或类似情况的人.

有一个变体debounce()需要一个函数来决定去抖这个特定项目多长时间.它通过返回在一段时间后完成的observable来指定它.您的功能可以返回empty()第一个项目和timer()其余项目.像(未经测试)的东西:

String[] items = {"one", "two", "three", "four", "five", "six"};
Observable.from(items)
    .debounce(item -> item.equals("one")
            ? Observable.empty()
            : Observable.timer(1, TimeUnit.SECONDS));

诀窍是这个函数必须知道哪个项目是第一个.你的序列可能知道.如果没有,你可能不得不zip()用range()或什么的.在这种情况下更好的是在另一个答案中使用该解决方案.

Answer 3

使用RxJava 2.0的简单解决方案,从RxJS 的相同问题的答案翻译而来,结合了throttleFirst和debounce,然后删除重复项.

private <T> ObservableTransformer<T, T> debounceImmediate() {
    return observable  -> observable.publish(p -> 
        Observable.merge(p.throttleFirst(1, TimeUnit.SECONDS), 
            p.debounce(1, TimeUnit.SECONDS)).distinctUntilChanged());
} 

@Test
public void testDebounceImmediate() {
    Observable.just(0, 100, 200, 1500, 1600, 1800, 2000, 10000)
        .flatMap(v -> Observable.timer(v, TimeUnit.MILLISECONDS).map(w -> v))
        .doOnNext(v -> System.out.println(LocalDateTime.now() + " T=" + v))
            .compose(debounceImmediate())
            .blockingSubscribe(v -> System.out.println(LocalDateTime.now() + " Debounced: " + v));
}

使用limit()或take()的方法似乎不能处理长期存在的数据流,我可能希望不断观察,但仍然会立即对一段时间内看到的第一个事件采取行动.

Answer 4

使用该debounce函数的版本并以这种方式实现该函数:

    .debounce(new Func1<String, Observable<String>>() {
        private AtomicBoolean isFirstEmission = new AtomicBoolean(true);
        @Override
        public Observable<String> call(String s) {
             // note: standard debounce causes the first item to be
             // delayed by 1 second unnecessarily, this is a workaround
             if (isFirstEmission.getAndSet(false)) {
                 return Observable.just(s);
             } else {
                 return Observable.just(s).delay(1, TimeUnit.SECONDS);
             }
        }
    })

第一个项目立即发出.后续项目延迟一秒.如果延迟观察值在下一个项目到达之前没有终止,则会被取消,因此会实现预期的去抖动行为.

Answer 5

Kotlin 扩展函数基于@lopar 的评论：

fun <T> Flowable<T>.debounceImmediate(timeout: Long, unit: TimeUnit): Flowable<T> {
    return publish {
        it.take(1).concatWith(it.debounce(timeout, unit))
    }
}

fun <T> Observable<T>.debounceImmediate(timeout: Long, unit: TimeUnit): Observable<T> {
    return publish {
        it.take(1).concatWith(it.debounce(timeout, unit))
    }
}

Answer 6

该LordRaydenMK和洛帕尔的回答有一个问题：你总是输掉第二项。我认为以前没有人会解决这个问题，因为如果您进行了防抖操作，通常会发生很多事件，而第二种方法无论如何都会与防抖操作有关。永不丢失事件的正确方法是：

observable
    .publish(published ->
        published
            .limit(1)
            .concatWith(published.debounce(1, TimeUnit.SECONDS)));

不用担心，您不会获得任何重复的事件。如果不确定，可以运行以下代码并亲自检查：

Observable.just(1, 2, 3, 4)
    .publish(published ->
        published
            .limit(1)
            .concatWith(published))
    .subscribe(System.out::println);