Android Rxjava3 使用場景詳解

更新時(shí)間：2022年04月07日 15:06:17 作者：香草可樂

本文主要介紹了Android Rxjava3 使用場景詳解，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

一、Rxjava使用場景

為了模擬實(shí)際場景，從wanandroid網(wǎng)站找了二個接口，如下：（對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、多任務(wù)嵌套回調(diào)

場景：比如調(diào)用接口一有回調(diào)后才能調(diào)用接口二，如果接口一調(diào)用失敗不再調(diào)用接口二。下面是二種寫法。

寫法一，代碼如下：

//為了看清楚代碼,沒有使用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 {
               //處理第一個請求返回的數(shù)據(jù)
                if(baseResponseObservable!=null) mTv.setText(baseResponseObservable.blockingSingle().toString());
                return hotKey;   //發(fā)起第二次網(wǎng)絡(luò)請求
            }
        }).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 {
                //處理第二次網(wǎng)絡(luò)請求的結(jié)果
                if(baseResponseObservable!=null) mTvTwo.setText(baseResponseObservable.blockingSingle().toString());
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Throwable {
                //異常的處理:比如Dialog的Dismiss,缺省頁展示等
                //注意:如果第一個網(wǎng)絡(luò)請求異常,整個事件會中斷,不會執(zhí)行第二個網(wǎng)絡(luò)請求,如果多個請求同理
                //但是請求成功的還是能正常處理
                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 {
                        //處理第一個網(wǎng)絡(luò)請求的結(jié)果
                        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;   //將第一個網(wǎng)絡(luò)請求轉(zhuǎn)換為第二個網(wǎng)絡(luò)請求
                    }
                }).observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<BaseResponse<List<HotKeyResp.DataBean>>>() {
                    @Override
                    public void accept(BaseResponse<List<HotKeyResp.DataBean>> listBaseResponse) throws Throwable {
                        //處理第二次網(wǎng)絡(luò)請求的結(jié)果
                        if(listBaseResponse!=null) mTvTwo.setText(listBaseResponse.toString());
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Throwable {
                        //注意:如果第一個網(wǎng)絡(luò)請求異常,整個事件會中斷,不會執(zhí)行第二個網(wǎng)絡(luò)請求,多個請求同理
                        //但是在異常前面已經(jīng)成功的網(wǎng)絡(luò)請求還是能正常處理
                        //異常的處理:比如Dialog的Dismiss,缺省頁展示等
                        LogUtil.e(throwable.toString());
                    }
                });

注意異常處理和線程切換，其他細(xì)節(jié)代碼和注釋比較詳細(xì)。

2、多任務(wù)合并處理

場景：接口一和接口二返回?cái)?shù)據(jù)后一起處理。
代碼如下：

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)  //傳入方法定義合并規(guī)則
            .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() {

                }
            });

}

//合并的規(guī)則,以及定義合并的返回值
public String combiNotification(BaseResponse<ArticleListResp> articleListRespBaseResponse, BaseResponse<List<HotKeyResp.DataBean>> hotkeyResponse) {
    //比如這里取二個接口數(shù)據(jù)toString返回
    if (articleListRespBaseResponse != null && hotkeyResponse != null) {
        return articleListRespBaseResponse.toString() + hotkeyResponse.toString();
    }
    return null;
}

3、輪詢

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

特別注意repeatWhen操作符，只有在repeatWhen的Function方法中發(fā)射onNext事件，重復(fù)(repeat)才能觸發(fā)，發(fā)射onError或者onComplite都會結(jié)束重復(fù)(repeat)，基于這一點(diǎn)，通過flatMap操作符將事件轉(zhuǎn)化為延遲一定時(shí)間的onNext事件，就達(dá)到了延時(shí)輪詢的目的。至于onNext事件發(fā)射的什么不重要。

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

takeUntil操作符可以定義一定的條件，當(dāng)達(dá)到條件時(shí)自動結(jié)束整個事件的目的，事件結(jié)束時(shí)會回調(diào)subscribe。

代碼如下：

/**
 * 輪詢
 * @param pollingTimes 輪詢的次數(shù)
 */
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<?>>() {  //轉(zhuǎn)換事件
                @Override
                public ObservableSource<?> apply(Object o) throws Throwable {
                    //這里發(fā)射延時(shí)的onNext事件，觸發(fā)repeat動作,發(fā)射的0不會回調(diào)到下面的subscribe
                    return Observable.just(0).delay(2, TimeUnit.SECONDS);  
                }
            });
        }
    }).subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            //takeUntil定義了二個結(jié)束條件：前面是達(dá)到了輪詢的次數(shù)，后面是網(wǎng)絡(luò)請求返回了成功，當(dāng)然也可以寫成代碼塊做其他的返回判斷
            .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() {
                   
                }
            });
}

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

場景二：輪詢固定的次數(shù)（間隔一定的時(shí)間），可以提前退出輪詢，也可以等輪詢到指定次數(shù)后自動退出，這里的輪詢不關(guān)心上次請求的結(jié)果。
代碼如下:

/**
 * 輪詢一定的次數(shù)
 * @param pollTimes 輪詢次數(shù)
 */
private void timedPolling(int pollTimes) {
    //網(wǎng)絡(luò)請求
    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;  //轉(zhuǎn)換事件
                }
            }).subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(new Consumer<BaseResponse<ArticleListResp>>() {
                @Override
                public void accept(BaseResponse<ArticleListResp> listBaseResponse) throws Throwable {
                    //如果滿足了退出輪詢的條件,可以調(diào)用下面的方法退出輪詢
                    //mSubscribe.dispose();
                }
            });
}

思路是定時(shí)發(fā)射事件，然后將事件轉(zhuǎn)化為網(wǎng)絡(luò)請求。同理可以寫出不限次數(shù)的輪詢。

場景三：不限次數(shù)輪詢（間隔一定的時(shí)間），不關(guān)心上次請求的結(jié)果。

假如接口返回的code為0時(shí)需要取消輪詢，代碼如下：

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自動取消發(fā)射
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Consumer<BaseResponse<ArticleListResp>>() {
            @Override
            public void accept(BaseResponse<ArticleListResp> articleListRespBaseResponse) throws Throwable {
                //處理回調(diào)
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Throwable {
               //處理異常
            }
        });

如果是其他取消條件，也可以寫在代碼塊里：

.takeUntil(response -> {
    //處理接口數(shù)據(jù)，然后判斷是返回true還是false，true:停止發(fā)射，false:繼續(xù)發(fā)射
    return false;
})  //使用takeUntil自動取消發(fā)射

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

4、其他小場景

1）倒計(jì)時(shí)

驗(yàn)證碼的倒計(jì)時(shí)功能，代碼如下：

/**
 * 倒計(jì)時(shí)
 * @param countDownSeconds 倒計(jì)時(shí)的秒數(shù)
 */
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("獲取驗(yàn)證碼");
                }

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

效果

2）打字機(jī)效果

幾行代碼實(shí)現(xiàn)打字機(jī)效果：

@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 {
                //動態(tài)改變文本長度
                    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">\u3000\u3000你好,這是一個打字機(jī),這是一個打字機(jī)這是一個打字機(jī)這是一個打字機(jī)。\n\u3000\u3000換行空格繼續(xù)打印。</string>

二、結(jié)合Rxbinding的使用場景

RxBinding 提供的綁定能夠?qū)⑷魏?Android View 事件轉(zhuǎn)換為 Observable。

一旦將 View 事件轉(zhuǎn)換為 Observable ，將發(fā)射數(shù)據(jù)流形式的 UI 事件，我們就可以訂閱這個數(shù)據(jù)流，這與訂閱其他 Observable 方式相同。

引入下面的庫：

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

1、點(diǎn)擊事件防抖

點(diǎn)擊事件的寫法:

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

長點(diǎn)擊事件的寫法:

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

點(diǎn)擊防抖事件的寫法:

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

2、輸入搜索優(yōu)化

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

            }
        });

3、聯(lián)合判斷

combineLatest 操作符將多個 Observable 發(fā)射的事件組裝起來，然后再發(fā)射組裝后的新事件。

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

對訂閱事件統(tǒng)一管理

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