Rx 中的立即去抖

Question

我正在寻找 debounce 一系列事件的操作员，比如说用户的点击。输入和输出应该是这样的：

interval :      ->     <-            ->     <-
in       : 1--2--3-------4--5--5--6-7-8--------
out      : 1-------------4---------------------

这个想法就像使用 immediate 选项 on http://underscorejs.org/#debounce 的下划线去抖动。运算符可以是 presented/implemented 任何支持 Reactive Extensions

的语言

编辑：澄清间隔，比如说 5 秒（两个箭头之间有 5 个空格）：-> <-

Edit2：一个更容易理解的版本：我有一个用户，他反复点击一个按钮（1、2、3）；我想抓住第一个 click (1) 而忽略其余部分。过了一会儿，他累了休息了7秒（比两个箭头间隔的5秒还长）继续再次点击按钮（4,5,6,7,8）我要抓第一个click (4) 忽略其余部分。

如果他在第四个箭头后点击，我也想抓住那个点击。

Edit3：这是一张图片 which can be found at the original article

Answer 1

因为debounce()本质上是异步的，需要显式地将结果带回当前线程。

seriesOfUnfortunateEvents
  .debounce( 14, TimeUnit.MILLISECONDS )
  .observeOn( Schedulers.immediate() )
  .subscribe( v -> yourStuff() );

Answer 2

根据 documentation，RxJS 中有两个去抖运算符。您可能对 debounceTime 特别感兴趣。

`debounceTime`

From documentation

Emits a value from the source Observable only after a particular time span has passed without another source emission.

示例：

Rx.Observable
    .fromEvent(document.querySelector('button'), 'click')
    .debounceTime(200)
    .mapTo(() => 'clicked!')
    .subscribe(v => console.log(v));

如果按钮在给定时间跨度（本例中为 200 毫秒）内被点击，它将发出 clicked！。

`debounce`

From documentation

Emits a value from the source Observable only after a particular time span determined by another Observable has passed without another source emission.

Answer 3

您想要的行为不是 debounce 运算符在 Rx 中所做的。

这称为 throttle、throttleTime 或 throttleWithTimeout（但是，它属于 debounce 运算符类别）。我不知道你使用的是什么语言，但在 RxJS 中它看起来像下图：

见http://reactivex.io/documentation/operators/debounce.html。

Answer 4

编辑： 根据说明，RxJava 没有用于此类流的运算符，但它可以由 non-trivial 组其他运算符组成:

import java.util.concurrent.TimeUnit;

import rx.Observable;

public class DebounceFirst {

    public static void main(String[] args) {
        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("T=" + v))
        .compose(debounceFirst(500, TimeUnit.MILLISECONDS))
        .toBlocking()
        .subscribe(v -> System.out.println("Debounced: " + v));
    }

    static <T> Observable.Transformer<T, T> debounceFirst(long timeout, TimeUnit unit) {
        return f -> 
            f.publish(g ->
                g.take(1)
                .concatWith(
                    g.switchMap(u -> Observable.timer(timeout, unit).map(w -> u))
                    .take(1)
                    .ignoreElements()
                )
                .repeatWhen(h -> h.takeUntil(g.ignoreElements()))
            );
    }
}

Rx 中的立即去抖

Immediate debounce in Rx

system.reactive

rxjs

rx-java

debounce

`debounceTime`

`debounce`