RxJava debounce 运算符仅显示 Android 上的最后一次发射

Question

我正在尝试在 Android 上使用 RxJava 在 1 秒后发出一个数字。

但是，只有最后一项 (7) 被发出。

我做错了什么？

private Observable<Integer> getIntegerObservable(){
    return Observable.just(1,2,3,4,5,6,7);
}

getIntegerObservable()
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .debounce(1, TimeUnit.SECONDS)
                    .subscribe(new Observer<Integer>() {
                        @Override
                        public void onSubscribe(Disposable d) {

                        }

                        @Override
                        public void onNext(Integer integer) {
                            textView.setText(textView.getText().toString() + String.valueOf(integer)+"\n");
                        }

                        @Override
                        public void onError(Throwable e) {
                            System.out.println(e.getMessage());
                        }

                        @Override
                        public void onComplete() {

                        }
                    });

Answer 1

Debounce 运算符过滤掉源 Observable 发出的项目，这些项目紧随其后的是另一个发射的项目。

在您的情况下，数字 1 到 7 都在一秒钟内发出。 为去抖动指定的窗口是一秒。在一秒结束时发出的最新项目是 7，因此所有其他项目都被过滤掉，您会收到 7 作为输出。

下图显示了在一个时间范围内连续发出 6 个项目的情况。只有第六项会通过，其他项会被过滤掉。

下图显示了另一种情况，即在一个时间帧之后发出第 6 项，因此第 5 项和第 6 项都被发出，而其他项被过滤。

希望这会有所帮助，请尝试在documentation 网站上玩弹珠图。

Answer 2

正如documentation 所说

Debounce 仅在经过特定时间跨度后才从 Observable 发射一个项目，而它没有发射另一个项目。它过滤掉 源 Observable 发出的项目被快速跟随 另一个发射的项目。

因为在您的情况下，所有项目都在时间窗口内发出。只有最新的项目被发布，所有其他项目都被过滤掉。

我猜你要找的是Delay

Delay 操作符通过暂停一个 在发射每个之前的特定时间增量（您指定） 源 Observable 的项目。这具有转移 Observable 发出的整个项目序列在时间上向前 那个指定的增量。

所以它应该看起来像这样：

getIntegerObservable()
          .delay(1, TimeUnit.SECONDS)
          .subscribeOn(Schedulers.io())
          .observeOn(AndroidSchedulers.mainThread())
          .subscribe(...);

您还应该考虑在subscribeOn(..) 之前调用delay(..) 以避免阻塞主线程

查看此article，它解释了subscribeOn 和observeOn 之间的区别

更新

您可以将 CountDownTimer 包装在 Obserable 中。例如：

Observable.create(emitter -> new CountDownTimer(7000, 1000) {

          public void onTick(long millisUntilFinished) {
                 emitter.onNext(millisUntilFinished / 1000);
           }

           public void onFinish() {
                emitter.onComplete();
           }
         }.start())
.doOnNext() //you can log your timer here
.subscribe(); //or here

奖金（只是为了好玩：D）

RxJava 使用 java8 看起来更酷

Observable.create(this::startCountDownTimer)
       .doOnNext() //use long value
       .map(String::valueOf) //map to string
       .doOnNext() //use string value
       .doOnComplete() //gets called when the timer is up
       .subscribe(this::handleTick, this::handleError);

倒计时方法

private void startCountDownTimer(ObservableEmitter<Long> emitter) {
    new CountDownTimer(7000, 1000) {
        public void onTick(long millisUntilFinished) {
            emitter.onNext(millisUntilFinished / 1000);
        }

        public void onFinish() {
            emitter.onComplete();
        }
    }.start();
}

处理滴答方法

private void handleTick(String timer) {
    //Update UI
}

处理错误方法

private void handleError(Throwable t) {
    //Log and inform user
}