java - RxJava中对buffer的错误理解

Question

我正在尝试熟悉 RxJava。这是我要实现的用例:

我的屏幕上有一个按钮，我正在尝试收集点击次数。因此，如果用户点击按钮，则会记录一次点击并生成日志。现在，如果用户点击按钮两次，那么它会记录两次点击，收集它们并输出 2 而不是 1。

本质上，我试图在一段时间内累积点击次数，然后吐出最终结果。我猜“buffer”是我需要使用的方法。我在 Android 中快速创建了一个示例(代码如下)，但缓冲方法似乎并不像收集所有事件输入并吐出一个集合那么简单。

public class DemoFragment
    extends Fragment {

    private int _tapCount = 0;
    private Observable<List<Integer>> _bufferedObservable;
    private Observer<List<Integer>> _observer;

    @Override
    public void onActivityCreated(@Nullable Bundle savedInstanceState) {
        super.onActivityCreated(savedInstanceState);
        _setupLogger();

        _bufferedObservable = _getBufferedObservable();
        _observer = _getObserver();
    }


    // the library butterknife allows this
    @OnClick(R.id.btn_start_operation)
    public void onButtonTapped() {
        _log("GOT A TAP");
        _bufferedObservable.subscribeOn(Schedulers.io())
                         .observeOn(AndroidSchedulers.mainThread())
                         .subscribe(_observer);
    }

    private Observable<List<Integer>> _getBufferedObservable() {
        return Observable.create(new Observable.OnSubscribe<Integer>() {


            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                subscriber.onNext(1);   // send one tap here
            }

        }).buffer(2, TimeUnit.SECONDS); // collect all taps in the last 2s
    }

    private Observer<List<Integer>> _getObserver() {
        return new Observer<List<Integer>>() {


            @Override
            public void onCompleted() {
                _log(String.format("%d taps", _tapCount));
                _tapCount = 0; // reset tap count
            }

            @Override
            public void onError(Throwable e) {}

            @Override
            public void onNext(List<Integer> integers) {
                if (integers.size() > 0) {
                    for (int i : integers) {
                        _tapCount += i;
                    }
                    onCompleted();
                } else {
                    _log("No taps received");
                }
            }
        };
    }

    // ... other method that help wiring up the example (irrelevant to RxJava)
}

任何人都可以帮助我理解我理解中的误解吗？

问题 1:我期待 _getObserver()的 onNext向我发送包含累计点击次数的列表。因此，如果按钮被击中 5 次，那么我期待一个包含 5 个项目的列表，每个项目的值为“1”。使用现有代码，我总是得到一个空列表。

问题 2:如果通过检查 List<Integer> integers 没有收到任何事件，我基本上会做一个控制台日志尺寸。如果列表不为空，我会输入控制台日志，提示“未收到点击”。看来 Observable 永远不会停止。它几乎就像一个计时器，它会不断地持续运行，即使没有注册按钮点击也是如此。如果在过去 10 秒内没有注册任何事件，是否有办法停止 Observable？

问题 3:发射的数量似乎几乎呈指数增长。这几乎就像它收集了以前所有时间的按钮空点击。

Answer 1

下面是一段代码，展示了我将如何做到这一点(假设您的按钮 ID 是 R.id.rx_button):

private Subscription mSubscription;

@Override
protected void onResume() {
    super.onResume();
    mSubscription = Observable.create(new OnSubscribe<Integer>() {
        @Override
        public void call(Subscriber<? super Integer> subscriber) {
            findViewById(R.id.rx_button).setOnClickListener(new OnClickListener() {
                @Override
                public void onClick(View v) {
                    subscriber.onNext(1);
                }
            });
        }
    }).buffer(2, TimeUnit.SECONDS)
            .subscribe(new Action1<List<Integer>>() {
                @Override
                public void call(List<Integer> integers) {
                    Log.i("TAG", String.valueOf(integers.size()));
                }
            });
}

@Override
protected void onPause() {
    super.onPause();
    mSubscription.unsubscribe();
}

简单地说，只需关闭 call 方法上的 OnClickListener 实现，这样您就可以使用其中的 subscriber 对象。

onResume 使用 lambda 看起来会更好(看看 Retrolambda 项目):

@Override
protected void onResume() {
    super.onResume();
    mSubscription = Observable.create((Subscriber<? super Integer> subscriber) ->
            findViewById(R.id.rx_button).setOnClickListener(view ->
                    subscriber.onNext(1))).buffer(2, TimeUnit.SECONDS)
            .subscribe(integers -> Log.i("TAG", String.valueOf(integers.size())));
}