RxJava中对缓冲区的理解不正确

Question

我正在努力熟悉RxJava.这是我想要实现的用例:

我的屏幕上有一个按钮,我正在尝试收集水龙头的数量.因此,如果用户点击按钮,则会注册单击并生成日志.现在,如果用户单击按钮两次,则它会记录两次点击,收集它们并输出2而不是1.

基本上,我试图在一段时间内累积点击次数,然后吐出最终结果.我猜" 缓冲 "是我需要使用的方法.我在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个项目的List,每个项目的值都为"1".使用现有代码,我总是得到一个空列表.

问题2:如果通过检查List<Integer> integers大小没有收到任何事件,我基本上会做一个控制台日志.如果list不为空,我会抛出一个控制台日志,说"没有收到点击".似乎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();
}

简单地说,只需关闭该方法的OnClickListener实现call,这样您就可以使用其中的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())));
}