Android Rxjava3 使用場景詳解

2022-04-07 19:01:42

一、Rxjava使用場景

為了模擬實際場景，從wanandroid網站找了二個介面，如下：（對Wanandroid表示感謝!）

public interface ApiServer {

    /**
     * 介面一
     * 獲取文章列表
     * @return
     */
    @GET("article/list/1/json")
    Observable<BaseResponse<ArticleListResp>> getArticleList();


    /**
     * 介面二
     * 獲取熱詞
     * @return
     */
    @GET("hotkey/json")
    Observable<BaseResponse<List<HotKeyResp.DataBean>>> getHotKey();

}

1、多工巢狀回撥

場景：比如呼叫介面一有回撥後才能呼叫介面二，如果介面一呼叫失敗不再呼叫介面二。下面是二種寫法。

寫法一，程式碼如下：

//為了看清楚程式碼,沒有使用lambda簡化
//介面一
Observable<BaseResponse<ArticleListResp>> articleList = ApiManager.getInstance().getApiService().getArticleList();
//介面二
Observable<BaseResponse<List<HotKeyResp.DataBean>>> hotKey = ApiManager.getInstance().getApiService().getHotKey();
Observable.just(articleList)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .map(new Function<Observable<BaseResponse<ArticleListResp>>, Observable<BaseResponse<List<HotKeyResp.DataBean>>>>() {
            @Override
            public Observable<BaseResponse<List<HotKeyResp.DataBean>>> apply(Observable<BaseResponse<ArticleListResp>> baseResponseObservable) throws Throwable {
               //處理第一個請求返回的資料
                if(baseResponseObservable!=null) mTv.setText(baseResponseObservable.blockingSingle().toString());
                return hotKey;   //發起第二次網路請求
            }
        }).subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Consumer<Observable<BaseResponse<List<HotKeyResp.DataBean>>>>() {
            @Override
            public void accept(Observable<BaseResponse<List<HotKeyResp.DataBean>>> baseResponseObservable) throws Throwable {
                //處理第二次網路請求的結果
                if(baseResponseObservable!=null) mTvTwo.setText(baseResponseObservable.blockingSingle().toString());
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Throwable {
                //異常的處理:比如Dialog的Dismiss,預設頁展示等
                //注意:如果第一個網路請求異常,整個事件會中斷,不會執行第二個網路請求,如果多個請求同理
                //但是請求成功的還是能正常處理
                LogUtil.e(throwable.toString());
            }
        });

寫法二，程式碼如下:

        //為了看清楚程式碼,沒有使用lambda簡化
        //介面一
        Observable<BaseResponse<ArticleListResp>> articleList = ApiManager.getInstance().getApiService().getArticleList();
        //介面二
        Observable<BaseResponse<List<HotKeyResp.DataBean>>> hotKey = ApiManager.getInstance().getApiService().getHotKey();
        //請求第一個
        articleList.subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext(new Consumer<BaseResponse<ArticleListResp>>() {
                    @Override
                    public void accept(BaseResponse<ArticleListResp> articleListRespBaseResponse) throws Throwable {
                        //處理第一個網路請求的結果
                        if(articleListRespBaseResponse!=null) mTv.setText(articleListRespBaseResponse.toString());
                    }
                }).observeOn(Schedulers.io())
                .flatMap(new Function<BaseResponse<ArticleListResp>, ObservableSource<BaseResponse<List<HotKeyResp.DataBean>>>>() {
                    @Override
                    public ObservableSource<BaseResponse<List<HotKeyResp.DataBean>>> apply(BaseResponse<ArticleListResp> articleListRespBaseResponse) throws Throwable {
                        return hotKey;   //將第一個網路請求轉換為第二個網路請求
                    }
                }).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<BaseResponse<List<HotKeyResp.DataBean>>>() {
                    @Override
                    public void accept(BaseResponse<List<HotKeyResp.DataBean>> listBaseResponse) throws Throwable {
                        //處理第二次網路請求的結果
                        if(listBaseResponse!=null) mTvTwo.setText(listBaseResponse.toString());
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Throwable {
                        //注意:如果第一個網路請求異常,整個事件會中斷,不會執行第二個網路請求,多個請求同理
                        //但是在異常前面已經成功的網路請求還是能正常處理
                        //異常的處理:比如Dialog的Dismiss,預設頁展示等
                        LogUtil.e(throwable.toString());
                    }
                });

注意例外處理和執行緒切換，其他細節程式碼和註釋比較詳細。

2、多工合併處理

場景：介面一和介面二返回資料後一起處理。
程式碼如下：

private void zipRequest() {
    //為了看清楚程式碼,沒有使用lambda簡化
    //介面一
    Observable<BaseResponse<ArticleListResp>> articleList = ApiManager.getInstance().getApiService().getArticleList();
    //介面二
    Observable<BaseResponse<List<HotKeyResp.DataBean>>> hotKey = ApiManager.getInstance().getApiService().getHotKey();
    Observable.zip(articleList, hotKey, this::combiNotification)  //傳入方法定義合併規則
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<String>() {
                @Override
                public void onSubscribe(@NonNull Disposable d) {

                }

                @Override
                public void onNext(@NonNull String msg) {
                    if(!TextUtils.isEmpty(msg)){
                        mTv.setText(msg);
                    }
                }

                @Override
                public void onError(@NonNull Throwable e) {

                }

                @Override
                public void onComplete() {

                }
            });

}

//合併的規則,以及定義合併的返回值
public String combiNotification(BaseResponse<ArticleListResp> articleListRespBaseResponse, BaseResponse<List<HotKeyResp.DataBean>> hotkeyResponse) {
    //比如這裡取二個介面資料toString返回
    if (articleListRespBaseResponse != null && hotkeyResponse != null) {
        return articleListRespBaseResponse.toString() + hotkeyResponse.toString();
    }
    return null;
}

3、輪詢

場景一：輪詢固定的次數（間隔一定的時間），可以提前退出輪詢，也可以等輪詢到指定次數後自動退出，每次輪詢必須等上一次輪詢有結果後才能開始下一次輪詢。

特別注意repeatWhen操作符，只有在repeatWhen的Function方法中發射onNext事件，重複(repeat)才能觸發，發射onError或者onComplite都會結束重複(repeat)，基於這一點，通過flatMap操作符將事件轉化為延遲一定時間的onNext事件，就達到了延時輪詢的目的。至於onNext事件發射的什麼不重要。

延伸：retryWhen的Function方法發射onError事件才會重試(retry)。

takeUntil操作符可以定義一定的條件，當達到條件時自動結束整個事件的目的，事件結束時會回撥subscribe。

程式碼如下：

/**
 * 輪詢
 * @param pollingTimes 輪詢的次數
 */
private void timedPolling(int pollingTimes) {
    AtomicInteger times = new AtomicInteger();
    Observable<BaseResponse<ArticleListResp>> articleList = ApiManager.getInstance().getApiService().getArticleList();
    articleList.repeatWhen(new Function<Observable<Object>, ObservableSource<?>>() {
        @Override
        public ObservableSource<?> apply(Observable<Object> objectObservable) throws Throwable {
            return objectObservable.flatMap(new Function<Object, ObservableSource<?>>() {  //轉換事件
                @Override
                public ObservableSource<?> apply(Object o) throws Throwable {
                    //這裡發射延時的onNext事件，觸發repeat動作,發射的0不會回撥到下面的subscribe
                    return Observable.just(0).delay(2, TimeUnit.SECONDS);  
                }
            });
        }
    }).subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            //takeUntil定義了二個結束條件：前面是達到了輪詢的次數，後面是網路請求返回了成功，當然也可以寫成程式碼塊做其他的返回判斷
            .takeUntil(response -> times.incrementAndGet() >= pollingTimes || response.getErrorCode() == 0)
            .subscribe(new Observer<BaseResponse<ArticleListResp>>() {
                @Override
                public void onSubscribe(@NonNull Disposable d) {
                   
                }

                @Override
                public void onNext(@NonNull BaseResponse<ArticleListResp> articleListRespBaseResponse) {
                    
                }

                @Override
                public void onError(@NonNull Throwable e) {
                   
                }

                @Override
                public void onComplete() {
                   
                }
            });
}

如果想改成不限制次數的也比較簡單。

場景二：輪詢固定的次數（間隔一定的時間），可以提前退出輪詢，也可以等輪詢到指定次數後自動退出，這裡的輪詢不關心上次請求的結果。
程式碼如下:

/**
 * 輪詢一定的次數
 * @param pollTimes 輪詢次數
 */
private void timedPolling(int pollTimes) {
    //網路請求
    Observable<BaseResponse<ArticleListResp>> articleList = ApiManager.getInstance().getApiService().getArticleList();
    //返回值用於取消輪詢
    mSubscribe = Observable.intervalRange(0, pollTimes, 0, 2000, TimeUnit.MILLISECONDS)
            .flatMap(new Function<Long, ObservableSource<BaseResponse<ArticleListResp>>>() {
                @Override
                public ObservableSource<BaseResponse<ArticleListResp>> apply(Long aLong) throws Throwable {
                    return articleList;  //轉換事件
                }
            }).subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<BaseResponse<ArticleListResp>>() {
                @Override
                public void accept(BaseResponse<ArticleListResp> listBaseResponse) throws Throwable {
                    //如果滿足了退出輪詢的條件,可以呼叫下面的方法退出輪詢
                    //mSubscribe.dispose();
                }
            });
}

思路是定時發射事件，然後將事件轉化為網路請求。同理可以寫出不限次數的輪詢。

場景三：不限次數輪詢（間隔一定的時間），不關心上次請求的結果。

假如介面返回的code為0時需要取消輪詢，程式碼如下：

Observable<BaseResponse<ArticleListResp>> articleList = ApiManager.getInstance().getApiService().getArticleList();
//返回值用於取消輪詢
mSubscribe = Observable.interval(0, 2000, TimeUnit.MILLISECONDS)
        .flatMap(new Function<Long, ObservableSource<BaseResponse<ArticleListResp>>>() {
            @Override
            public ObservableSource<BaseResponse<ArticleListResp>> apply(Long aLong) throws Throwable {
                return articleList;
            }
        })
        .takeUntil(response -> response.getErrorCode() == 0)  //使用takeUntil自動取消發射
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Consumer<BaseResponse<ArticleListResp>>() {
            @Override
            public void accept(BaseResponse<ArticleListResp> articleListRespBaseResponse) throws Throwable {
                //處理回撥
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Throwable {
               //處理異常
            }
        });

如果是其他取消條件，也可以寫在程式碼塊裡：

.takeUntil(response -> {
    //處理介面資料，然後判斷是返回true還是false，true:停止發射，false:繼續發射
    return false;
})  //使用takeUntil自動取消發射

不管何種輪詢，注意在OnDestroy中取消。

4、其他小場景

1）倒計時

驗證碼的倒計時功能，程式碼如下：

/**
 * 倒計時
 * @param countDownSeconds 倒計時的秒數
 */
private void countDown(int countDownSeconds) {
    Observable.intervalRange(0, countDownSeconds, 0, 1000, TimeUnit.MILLISECONDS)
            .map(new Function<Long, String>() {
                @Override
                public String apply(Long aLong) throws Throwable {
                    return (countDownSeconds - aLong) + "s後重新獲取";
                }
            }).observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Observer<String>() {
                @Override
                public void onSubscribe(@NonNull Disposable d) {
                    mTv.setEnabled(false);
                }

                @Override
                public void onNext(@NonNull String s) {
                    mTv.setText(s);
                }

                @Override
                public void onError(@NonNull Throwable e) {
                    mTv.setEnabled(true);
                    mTv.setText("獲取驗證碼");
                }

                @Override
                public void onComplete() {
                    mTv.setText("獲取驗證碼");
                    mTv.setEnabled(true);
                }
            });
}

效果

2）打字機效果

幾行程式碼實現打字機效果：

@RequiresApi(api = Build.VERSION_CODES.M)  //6.0
public class DaziView extends View {
    private TextPaint mTextPaint;
    private StaticLayout mStaticLayout;

    public DaziView(Context context) {
        super(context,null);
    }

    public DaziView(Context context, @Nullable AttributeSet attrs) {
        super(context, attrs);
        initTextPaint();
    }

    /**
     * 初始化畫筆
     */
    private void initTextPaint() {
        mTextPaint = new TextPaint(Paint.ANTI_ALIAS_FLAG);
        mTextPaint.setTextSize(48);
        mTextPaint.setColor(Color.parseColor("#000000"));
    }

    /**
     * 繪製
     * @param content
     */
  public void drawText(String content){
        if(!TextUtils.isEmpty(content)){
            Observable.intervalRange(0,content.length()+1,0,150, TimeUnit.MILLISECONDS)
                    .subscribe(new Consumer<Long>() {
                @Override
                public void accept(Long aLong) throws Throwable {
                //動態改變文字長度
                    mStaticLayout = StaticLayout.Builder.obtain(content, 0, aLong.intValue(), mTextPaint, getWidth())
                            .build();
                    invalidate();
                }
            });
        }
  }

    @Override
    protected void onDraw(Canvas canvas) {
        super.onDraw(canvas);
        //繪製文字
        mStaticLayout.draw(canvas);
    }
}

效

文字

<string name="dazi_content">u3000u3000你好,這是一個打字機,這是一個打字機這是一個打字機這是一個打字機。nu3000u3000換行空格繼續列印。</string>

二、結合Rxbinding的使用場景

RxBinding 提供的繫結能夠將任何 Android View 事件轉換為 Observable。

一旦將 View 事件轉換為 Observable ，將發射資料流形式的 UI 事件，我們就可以訂閱這個資料流，這與訂閱其他 Observable 方式相同。

引入下面的庫：

implementation 'com.jakewharton.rxbinding4:rxbinding:4.0.0'

1、點選事件防抖

點選事件的寫法:

RxView.clicks(button)   //button為控制元件
        .subscribe(new Consumer<Unit>() {
            @Override
            public void accept(Unit unit) throws Throwable {
                //點選事件
            }
        });

長點選事件的寫法:

RxView.longClicks(button)
        .subscribe(new Consumer<Unit>() {
            @Override
            public void accept(Unit unit) throws Throwable {
                //長點選自動響應,不需要等放開手指
            }
        });

點選防抖事件的寫法:

RxView.clicks(button)
        .throttleFirst(1000, TimeUnit.MILLISECONDS)   //一秒以內第一次點選事件有效
        .subscribe(new Consumer<Unit>() {
            @Override
            public void accept(Unit unit) throws Throwable {
                //點選事件
            }
        });

2、輸入搜尋優化

RxTextView.textChanges(editText)  //傳入EditText控制元件
        .debounce(1000,TimeUnit.MILLISECONDS)  //一秒內沒有新的事件時,取最後一次事件發射
        .skip(1)    //跳過第一次EditText的空內容
        .subscribeOn(AndroidSchedulers.mainThread())
        .subscribe(new Consumer<CharSequence>() {
            @Override
            public void accept(CharSequence charSequence) throws Throwable {
                //EditText的內容
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Throwable {

            }
        });

3、聯合判斷

combineLatest 操作符將多個 Observable 發射的事件組裝起來，然後再發射組裝後的新事件。

Observable<CharSequence> observableEdittext = RxTextView.textChanges(editText).skip(1);
 Observable<CharSequence> observableEdittextTwo =   RxTextView.textChanges(editText_two).skip(1);

 Observable.combineLatest(observableEdittext, observableEdittextTwo, new BiFunction<CharSequence, CharSequence, Boolean>() {
     @Override
     public Boolean apply(CharSequence charSequence, CharSequence charSequence2) throws Throwable {
         if(!TextUtils.isEmpty(charSequence)&&!TextUtils.isEmpty(charSequence2)){
             return true;
         }
         return false;
     }
 }).subscribe(new Consumer<Boolean>() {
     @Override
     public void accept(Boolean aBoolean) throws Throwable {
             //TODO 其他處理
     }
 });

三、防洩漏

1、Observable.unsubscribeOn

Observable<Integer> just = Observable.just(0);
just.subscribeOn(Schedulers.io()).unsubscribeOn(Schedulers.io());  //取消事件，防止洩漏

2、disposable.dispose

這個比較常用。

3、CompositeDisposable

對訂閱事件統一管理

CompositeDisposable compositeDisposable = new CompositeDisposable();
compositeDisposable.add(disposableOne);
compositeDisposable.add(disposableTwo);
compositeDisposable.clear();