本文为博主原创文章，转载请注明出处

前言

本文总结项目中常用的 Rxjava2 操作符。

CV大神

使用RxJava2原因

• ★★★★★ 封装了线程切换，不用自己封装线程池、Handler了
• ★★★★☆ 链式调用，一气呵成

单值发射

Single：大多数场景都是单值发射，所以使用 Single 即可覆盖大部分场景。

Case1: 在非UI线程执行且不关注结果

• fromCallable

Single.fromCallable(new Callable<Integer>() {@Overridepublic Integer call() throws Exception {Logger.d(TAG, "test: fromCallable() invoked on %s", Thread.currentThread().getName());return generateRandom();}}).subscribeOn(Schedulers.io()).subscribe();

从名字 Callable 就能看出，这是个回调函数，在 io 线程运行

• defer

        Single.defer(new Callable<SingleSource<Integer>>() {@Overridepublic SingleSource<Integer> call() throws Exception {Logger.d(TAG, "test: defer() invoked on %s", Thread.currentThread().getName());return Single.just(generateRandom());}}).subscribeOn(Schedulers.io()).subscribe();

注意，不能直接使用 Single.just(generateRandom()).subscribeOn(Schedulers.io()).subscribe()，这么写将直接在当前线程调用 generateRandom() ，无法实现在 io 线程执行的效果。

Case2: 在非UI线程执行并关注结果

需要关注结果的场景，建议都订阅 Consumer<Throwable>，因为 RxJava 内部捕获了 Exception，导致外部无感知

        Single.fromCallable(new Callable<Integer>() {@Overridepublic Integer call() throws Exception {return generateRandom();}}).subscribeOn(Schedulers.io()).subscribe(new Consumer<Integer>() {@Overridepublic void accept(Integer integer) throws Exception {Logger.d(TAG, "test: accept(Integer integer) invoked on %s", Thread.currentThread().getName());}}, new Consumer<Throwable>() {@Overridepublic void accept(Throwable throwable) throws Exception {Logger.d(TAG, "test: accept(Throwable throwable) invoked on %s", Thread.currentThread().getName());}});

Case3: 在非UI线程执行但在UI线程关注结果

        Single.fromCallable(new Callable<Integer>() {@Overridepublic Integer call() throws Exception {return generateRandom();}}).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<Integer>() {@Overridepublic void accept(Integer integer) throws Exception {Logger.d(TAG, "test: accept(Integer integer) invoked on %s", Thread.currentThread().getName());}}, new Consumer<Throwable>() {@Overridepublic void accept(Throwable throwable) throws Exception {Logger.d(TAG, "test: accept(Integer integer) invoked on %s", Thread.currentThread().getName());}});

Case4: 变换返回值

        Single.fromCallable(new Callable<Integer>() {@Overridepublic Integer call() throws Exception {return generateRandom();}}).map(new Function<Integer, String>() {@Overridepublic String apply(Integer integer) throws Exception {return String.valueOf(integer + "_mapped");}}).subscribe(new Consumer<String>() {@Overridepublic void accept(String s) throws Exception {Logger.d(TAG, "test: " + s);}});

Case5: 按顺序做某事且下一件事依赖上一件事的结果[常用于网络请求接口依赖时]

        Single.defer(new Callable<SingleSource<String>>() {@Overridepublic SingleSource<String> call() throws Exception {return getUserId();}}).flatMap(new Function<String, SingleSource<UserInfo>>() {@Overridepublic SingleSource<UserInfo> apply(String userId) throws Exception {return getUserInfo(userId);}}).subscribeOn(Schedulers.io()).subscribe(new Consumer<UserInfo>() {@Overridepublic void accept(UserInfo userInfo) throws Exception {Logger.d(TAG, "test: get userInfo success: " + userInfo);}}, new Consumer<Throwable>() {@Overridepublic void accept(Throwable throwable) throws Exception {Logger.e(TAG, "test: get userInfo error.", throwable);}});

Case6: 并发读取不同数据源，转换成同类型后，合并

        Single<IBook> novel = Single.fromCallable(new Callable<Novel>() {@Overridepublic Novel call() throws Exception {return getNovel();}}).map(new Function<Novel, IBook>() {@Overridepublic IBook apply(Novel novel) throws Exception {return new NovelAdapter(novel);}}).subscribeOn(Schedulers.io());Single<IBook> rxJava2Tutorial = Single.fromCallable(new Callable<RxJava2Tutorial>() {@Overridepublic RxJava2Tutorial call() throws Exception {return getRxJava2Tutorial();}}).map(new Function<RxJava2Tutorial, IBook>() {@Overridepublic IBook apply(RxJava2Tutorial rxJava2Tutorial) throws Exception {return new RxJava2TutorialAdapter(rxJava2Tutorial);}}).subscribeOn(Schedulers.io());Single.zip(novel, rxJava2Tutorial, new BiFunction<IBook, IBook, List<IBook>>() {@Overridepublic List<IBook> apply(IBook iBook, IBook iBook2) throws Exception {List<IBook> books = new ArrayList<>(2);books.add(iBook);books.add(iBook2);return books;}}).subscribe(new Consumer<List<IBook>>() {@Overridepublic void accept(List<IBook> iBooks) throws Exception {Logger.d(TAG, "test: books are " + iBooks);}}, new Consumer<Throwable>() {@Overridepublic void accept(Throwable throwable) throws Exception {Logger.d(TAG, "test: get books error.", throwable);}});

多值发射

Observable：使用场景较少，比如搜索功能需要不断发射搜索关键字。

Case1: 搜索频率限制

        Observable.create(new ObservableOnSubscribe<String>() {@Overridepublic void subscribe(final ObservableEmitter<String> emitter) throws Exception {mEditText.addTextChangedListener(new TextWatcher() {@Overridepublic void beforeTextChanged(CharSequence s, int start, int count, int after) {}@Overridepublic void onTextChanged(CharSequence s, int start, int before, int count) {}@Overridepublic void afterTextChanged(Editable s) {if (!emitter.isDisposed()) {emitter.onNext(s.toString().trim());}}});}}).debounce(200, TimeUnit.MILLISECONDS).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<String>() {@Overridepublic void accept(String keyword) throws Exception {mTextView.setText(search(keyword));}}, new Consumer<Throwable>() {@Overridepublic void accept(Throwable throwable) throws Exception {Logger.e(TAG, "test: emitter keyword error.", throwable);}});

背压

Flowable：使用场景最少，目前仅发射下载进度时可以用上。

Demo 地址

Click Here

总结

陈康肃公善射，当世无双 ，公亦以此自矜。尝射于家圃，有卖油翁释担而立，睨之久而不去。见其发矢十中八九，但微颔之。
康肃问曰：”汝亦知射乎？吾射不亦精乎？”翁曰：”无他， 但手熟尔。”康肃忿然曰：”尔安敢轻吾射！”翁曰：”以我酌油知之。”乃取一葫芦置于地，以钱覆其口，徐以杓酌油沥之，自钱孔入，而钱不湿。因曰：”我亦无他，惟手熟尔。”康肃笑而遣之。