RxJava2 实战知识梳理(5) 简单及进阶的轮询操做

1、示例

1.1 应用场景

今天，咱们介绍一种新的场景，轮询操做。也就是说，咱们会尝试间隔一段时间就向服务器发起一次请求，在使用RxJava以前，该需求的实现通常有两种方式：

• 经过Handler发送延时消息，在handleMessage中请求服务器以后，再次发送一个延时消息，直到达到循环次数为止。
• 使用Java提供的定时器Timer。

咱们尝试使用RxJava2提供的操做符来实现这一需求，这里演示两种方式的轮询，并将单次访问的次数限制在5次：

• 固定时延：使用intervalRange操做符，每间隔3s执行一次任务。
• 变长时延：使用repeatWhen操做符实现，第一次执行完任务后，等待4s再执行第二次任务，在第二次任务执行完成后，等待5s，依次递增。

2.2 示例

public class PollingActivity extends AppCompatActivity {

    private static final String TAG = PollingActivity.class.getSimpleName();

    private TextView mTvSimple;
    private TextView mTvAdvance;
    private CompositeDisposable mCompositeDisposable;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_polling);
        mTvSimple = (TextView) findViewById(R.id.tv_simple);
        mTvSimple.setOnClickListener(new View.OnClickListener() {

            @Override
            public void onClick(View v) {
                startSimplePolling();
            }

        });
        mTvAdvance = (TextView) findViewById(R.id.tv_advance);
        mTvAdvance.setOnClickListener(new View.OnClickListener() {

            @Override
            public void onClick(View v) {
                startAdvancePolling();
            }

        });
        mCompositeDisposable = new CompositeDisposable();
    }

    private void startSimplePolling() {
        Log.d(TAG, "startSimplePolling");
        Observable<Long> observable = Observable.intervalRange(0, 5, 0, 3000, TimeUnit.MILLISECONDS).take(5).doOnNext(new Consumer<Long>() {

            @Override
            public void accept(Long aLong) throws Exception {
                doWork(); //这里使用了doOnNext，所以DisposableObserver的onNext要等到该方法执行完才会回调。
            }

        });
        DisposableObserver<Long> disposableObserver = getDisposableObserver();
        observable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(disposableObserver);
        mCompositeDisposable.add(disposableObserver);
    }

    private void startAdvancePolling() {
        Log.d(TAG, "startAdvancePolling click");
        Observable<Long> observable = Observable.just(0L).doOnComplete(new Action() {

            @Override
            public void run() throws Exception {
                doWork();
            }

        }).repeatWhen(new Function<Observable<Object>, ObservableSource<Long>>() {

            private long mRepeatCount;

            @Override
            public ObservableSource<Long> apply(Observable<Object> objectObservable) throws Exception {
                //必须做出反应，这里是经过flatMap操做符。
                return objectObservable.flatMap(new Function<Object, ObservableSource<Long>>() {

                    @Override
                    public ObservableSource<Long> apply(Object o) throws Exception {
                        if (++mRepeatCount > 4) {
                            //return Observable.empty(); //发送onComplete消息，没法触发下游的onComplete回调。
                            return Observable.error(new Throwable("Polling work finished")); //发送onError消息，能够触发下游的onError回调。
                        }
                        Log.d(TAG, "startAdvancePolling apply");
                        return Observable.timer(3000 + mRepeatCount * 1000, TimeUnit.MILLISECONDS);
                    }

                });
            }

        });
        DisposableObserver<Long> disposableObserver = getDisposableObserver();
        observable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(disposableObserver);
        mCompositeDisposable.add(disposableObserver);
    }

    private DisposableObserver<Long> getDisposableObserver() {

        return new DisposableObserver<Long>() {

            @Override
            public void onNext(Long aLong) {}

            @Override
            public void onError(Throwable throwable) {
                Log.d(TAG, "DisposableObserver onError, threadId=" + Thread.currentThread().getId() + ",reason=" + throwable.getMessage());
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "DisposableObserver onComplete, threadId=" + Thread.currentThread().getId());
            }
        };
    }

    private void doWork() {
        long workTime = (long) (Math.random() * 500) + 500;
        try {
            Log.d(TAG, "doWork start, threadId=" + Thread.currentThread().getId());
            Thread.sleep(workTime);
            Log.d(TAG, "doWork finished");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
        mCompositeDisposable.clear();
    }
}
复制代码

startSimplePolling对应于固定时延轮询：

startAdvancePolling对应于变长时延轮询：

3、示例解析

下面，就让咱们一块儿来分析一下上面这两个例子中涉及到的知识点。

3.1 intervalRange & doOnNext 实现固定时延轮询

对于固定时延轮询的需求，采用的是intervalRange的方式来实现，它是一个建立型操做符，该Observable第一次先发射一个特定的数据，以后间隔一段时间再发送一次，它是interval和range的结合体，这两个操做符的原理图为：

该操做符的优点在于：

• 与interval相比，它能够指定第一个发送数据项的时延、指定发送数据项的个数。
• 与range相比，它能够指定两项数据之间发送的时延。

intervalRange的接收参数的含义为：

• start：发送数据的起始值，为Long型。
• count：总共发送多少项数据。
• initialDelay：发送第一个数据项时的起始时延。
• period：两项数据之间的间隔时间。
• TimeUnit：时间单位。

在轮询操做中通常会进行一些耗时的网络请求，所以咱们选择在doOnNext进行处理，它会在下游的onNext方法被回调以前调用，可是它的运行线程能够经过subscribeOn指定，下游的运行线程再经过observerOn切换会主线程，经过打印对应的线程ID能够验证结果。

当要求的数据项都发送完毕以后，最后会回调onComplete方法。

3.2 repeatWhen 实现变长时延轮询

3.2.1 使用 repeatWhen 实现重订阅

之因此能够经过repeatWhen来实现轮询，是由于它为咱们提供了重订阅的功能，而重订阅有两点要素：

• 上游告诉咱们一次订阅已经完成，这就须要上游回调onComplete函数。
• 咱们告诉上游是否须要重订阅，经过repeatWhen的Function函数所返回的Observable肯定，若是该Observable发送了onComplete或者onError则表示不须要重订阅，结束整个流程；不然触发重订阅的操做。

其原理图以下所示：

repeatWhen的难点在于如何定义它的 Function参数：

• Function的输入是一个Observable<Object>，输出是一个泛型ObservableSource<?>。
• 若是输出的Observable发送了onComplete或者onError则表示不须要重订阅，结束整个流程；不然触发重订阅的操做。也就是说，它 仅仅是做为一个是否要触发重订阅的通知，onNext发送的是什么数据并不重要。
• 对于每一次订阅的数据流 Function 函数只会回调一次，而且是在onComplete的时候触发，它不会收到任何的onNext事件。
• 在Function函数中，必须对输入的 Observable进行处理，这里咱们使用的是flatMap操做符接收上游的数据，对于flatMap的解释，你们能够参考 RxJava2 实战知识梳理(4) - 结合 Retrofit 请求新闻资讯 。

  而当咱们不须要重订阅时，有两种方式：

  • 返回Observable.empty()，发送onComplete消息，可是DisposableObserver并不会回调onComplete。
  • 返回Observable.error(new Throwable("Polling work finished"))，DisposableObserver的onError会被回调，并接受传过去的错误信息。

  3.2.2 使用 Timer 实现两次订阅之间的时延

  以上就是对于repeatWhen的解释，与repeatWhen相相似的还有retryWhen操做符，这个咱们在下一篇文章中再介绍，接下来，咱们看一下如何实现两次事件的时延。

  前面咱们分析过，重订阅触发的时间是在返回的ObservableSource发送了onNext事件以后，那么咱们经过该ObservableSource延迟发送一个事件就能够实现相应的需求，这里使用的是time操做符，它的原理图以下所示，也就是，在订阅完成后，等待指定的时间它才会发送消息。

  

  3.2.3 使用 doOnComplete 完成轮询的耗时操做

  因为在订阅完成时会发送onComplete消息，那么咱们就能够在doOnComplete中进行轮询所要进行的具体操做，它所运行的线程经过subscribeOn指定。

  ---

  更多文章，欢迎访问个人 Android 知识梳理系列：

  • Android 知识梳理目录：www.jianshu.com/p/fd82d1899…
  • 我的主页：lizejun.cn
  • 我的知识总结目录：lizejun.cn/categories/