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/