4

我正在尝试熟悉 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()'sonNext向我发送一份包含累计点击次数的列表。因此，如果按钮被点击 5 次，那么我期待一个包含 5 个项目的列表，每个项目的值都为“1”。使用现有代码，我总是得到一个空列表。

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

问题 3：发射的数量似乎几乎呈指数增长。这几乎就像它收集所有以前的按钮空水龙头一样。

4

3 回答 3

8

这是一个显示我将如何做的代码（假设您的按钮的 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对象。

使用onResumelambda 会更好看（看看 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())));
}
于 2014-07-24T21:35:44.997 回答
1

我认为最好使用很棒的rx-binding库。我认为该解决方案更清洁，尤其是使用 lambdas。我使用了过滤器运算符，所以我只在 onNext() 中得到结果，其中用户在给定的 2 秒内按下了 5 次以上。

我基于 ( RxJava-Android-Samples ) 中的一个示例，该示例使用旧的 rx 绑定依赖项。

@Override
public void onResume() {
    super.onResume();
    mClickSubscription = getBufferedSubscription();
}

@Override 
public void onPause() {
    super.onPause();
    mClickSubscription.unsubscribe();
} 

private Subscription getBufferedSubscription() {
    return RxView.clicks(rx_button)
            .map(aVoid -> 1)
            .buffer(2, TimeUnit.SECONDS)
            .filter(integers -> integers.size() > 5)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<List<Integer>>() {

                @Override
                public void onCompleted() {
                    // fyi: you'll never reach here
                }

                @Override
                public void onError(Throwable e) {
                    SLog.i("Dang error! check your logs");
                }

                @Override
                public void onNext(List<Integer> integers) {
                    SLog.i(String.format("%d taps", integers.size()));

                }
            });
}
于 2015-12-01T13:22:31.590 回答
0

试试下面的代码：

public class BufferExampleActivity extends AppCompatActivity {

    private static final String TAG = BufferExampleActivity.class.getSimpleName();
    Button btn;
    TextView textView;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_example);
        btn = (Button) findViewById(R.id.btn);
        textView = (TextView) findViewById(R.id.textView);

        btn.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View view) {
                doSomeWork();
            }
        });
    }

    /*
     * simple example using buffer operator - bundles all emitted values into a list
     */
    private void doSomeWork() {

        Observable<List<String>> buffered = getObservable().buffer(3, 1);

        // 3 means,  it takes max of three from its start index and create list
        // 1 means, it jumps one step every time
        // so the it gives the following list
        // 1 - one, two, three
        // 2 - two, three, four
        // 3 - three, four, five
        // 4 - four, five
        // 5 - five

        buffered.subscribe(getObserver());
    }

    private Observable<String> getObservable() {
        return Observable.just("one", "two", "three", "four", "five");
    }

    private Observer<List<String>> getObserver() {
        return new Observer<List<String>>() {

            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, " onSubscribe : " + d.isDisposed());
            }

            @Override
            public void onNext(List<String> stringList) {
                textView.append(" onNext size : " + stringList.size());
                textView.append(AppConstant.LINE_SEPARATOR);
                Log.d(TAG, " onNext : size :" + stringList.size());
                for (String value : stringList) {
                    textView.append(" value : " + value);
                    textView.append(AppConstant.LINE_SEPARATOR);
                    Log.d(TAG, " : value :" + value);
                }

            }

            @Override
            public void onError(Throwable e) {
                textView.append(" onError : " + e.getMessage());
                textView.append(AppConstant.LINE_SEPARATOR);
                Log.d(TAG, " onError : " + e.getMessage());
            }

            @Override
            public void onComplete() {
                textView.append(" onComplete");
                textView.append(AppConstant.LINE_SEPARATOR);
                Log.d(TAG, " onComplete");
            }
        };
    }
}
于 2017-04-15T05:11:26.833 回答