RxJava 2.x 入门

之前只大概了解RxJava,并没在实际的项目中实战过，但最近在研究讯飞语音的一个demo的时候发现，他们都在使用mvvm,dagger2,rxjava2.x, 姿态很优雅，很吸引人，心想，卧槽再不尝试一下就落后了，于是决定在项目中采用这些优秀的框架，与时俱进。在这里记录梳理一下Rxjava2.x 的知识。

RxJava的优点就不多说了，直接接入正题。

1.添加依赖

compile 'io.reactivex.rxjava2:rxjava:2.1.1'

2.Rxjava 原理

RxJava 以观察者模式为骨架

Rxjava2.x 中有两种观察者模式：

Observable ( 被观察者 ) / Observer ( 观察者 )
Flowable （被观察者）/ Subscriber （观察者）

3.基本使用

1）.使用步骤

一：初始化 Observable
二：初始化 Observer
三：建立订阅关系

2）.create

可用于获取一个别观察者的对象。

Observable.create(new ObservableOnSubscribe<Integer>() { // 第一步：初始化Observable@Overridepublic void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {e.onNext(1);e.onNext(2);e.onNext(3);e.onNext(4);e.onComplete();}}).subscribe(new Observer<Integer>() { // 第三步：订阅// 第二步：初始化Observer@Overridepublic void onSubscribe(@NonNull Disposable d) {      }@Overridepublic void onNext(@NonNull Integer integer) {Log.e(TAG, "onNext : value : " + integer + "\n" );}@Overridepublic void onError(@NonNull Throwable e) {Log.e(TAG, "onError : value : " + e.getMessage() + "\n" );}@Overridepublic void onComplete() {Log.e(TAG, "onComplete" + "\n" );}});

Observable 通过在subcribe方法中调用e.onNext(1), 在订阅被观察者之后，可以在订阅者的onNext（Integer integer）方法中获取对应的值。

3）.subScribeOn 与 observeOn

subscribeOn 用于指定 subscribe() 时所发生的线程

observeOn 方法用于指定下游 Observer(被观察者) 回调发生的线程。

简单地说，subscribeOn() 指定的就是发射事件的线程，observerOn 指定的就是订阅者接收事件的线程。
多次指定发射事件的线程只有第一次指定的有效，也就是说多次调用 subscribeOn() 只有第一次的有效，其余的会被忽略。
但多次指定订阅者接收线程是可以的，也就是说每调用一次 observerOn()，下游的线程就会切换一次。

Observable.create(new ObservableOnSubscribe<Integer>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {Log.e(TAG, "Observable thread is : " + Thread.currentThread().getName());e.onNext(1);e.onComplete();}}).subscribeOn(Schedulers.newThread()).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).doOnNext(new Consumer<Integer>() {@Overridepublic void accept(@NonNull Integer integer) throws Exception {Log.e(TAG, "After observeOn(mainThread)，Current thread is " + Thread.currentThread().getName());}}).observeOn(Schedulers.io()).subscribe(new Consumer<Integer>() {@Overridepublic void accept(@NonNull Integer integer) throws Exception {Log.e(TAG, "After observeOn(io)，Current thread is " + Thread.currentThread().getName());}});

实例代码中，分别用 Schedulers.newThread() 和 Schedulers.io() 对发射线程进行切换，并采用 observeOn(AndroidSchedulers.mainThread() 和 Schedulers.io() 进行了接收线程的切换。可以看到输出中发射线程仅仅响应了第一个 newThread，但每调用一次 observeOn() ，线程便会切换一次。

RxJava 中，已经内置了很多线程选项供我们选择，例如有：

Schedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作；
Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作；
Schedulers.newThread() 代表一个常规的新线程；
AndroidSchedulers.mainThread() 代表Android的主线程

4.操作符

4.1 map

map 操作符的作用是对上游大发送的每一个事件的Observables 通过一个函数，使得每一个事件都按照指定的函数去变化。

Observable.create(new ObservableOnSubscribe<Response>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<Response> e) throws Exception {Builder builder = new Builder().url("http://api.avatardata.cn/MobilePlace/LookUp?key=ec47b85086be4dc8b5d941f5abd37a4e&mobileNumber=13021671512").get();Request request = builder.build();Call call = new OkHttpClient().newCall(request);Response response = call.execute();e.onNext(response);}}).map(new Function<Response, MobileAddress>() {@Overridepublic MobileAddress apply(@NonNull Response response) throws Exception {if (response.isSuccessful()) {ResponseBody body = response.body();if (body != null) {Log.e(TAG, "map:转换前:" + response.body());return new Gson().fromJson(body.string(), MobileAddress.class);}}return null;}}).observeOn(AndroidSchedulers.mainThread()).doOnNext(new Consumer<MobileAddress>() {@Overridepublic void accept(@NonNull MobileAddress s) throws Exception {Log.e(TAG, "doOnNext: 保存成功：" + s.toString() + "\n");}}).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<MobileAddress>() {@Overridepublic void accept(@NonNull MobileAddress data) throws Exception {Log.e(TAG, "成功:" + data.toString() + "\n");}, new Consumer<Throwable>() {@Overridepublic void accept(@NonNull Throwable throwable) throws Exception {Log.e(TAG, "失败：" + throwable.getMessage() + "\n");}});

Observable 请求的数据发送后，经过map操作符，转换成Gson解析后的bean对象，然后再传到Observer中的accept 中。简而言之，map的作用就是转换数据。

4.2 concat 操作符

concat 连接操作符，其作用就是对数据的连接。可接受Observable的可变参数，或者Observable的集合。

   Observable.concat(Observable.just(1,2,3), Observable.just(4,5,6)).subscribe(new Consumer<Integer>() {@Overridepublic void accept(@NonNull Integer integer) throws Exception {mRxOperatorsText.append("concat : "+ integer + "\n");Log.e(TAG, "concat : "+ integer + "\n" );}});

输出结果是：

concat 1
concat 2
concat 3
concat 4
concat 5
concat 6

4.3flatMap

Observable 通过某种方法转换为多个 Observables，然后再把这些分散的 Observables装进一个单一的发射器 Observable.

应用场景：实现多个网络请求依次依赖。

Rx2AndroidNetworking.get("http://www.tngou.net/api/food/list").addQueryParameter("rows", 1 + "").build().getObjectObservable(FoodList.class) // 发起获取食品列表的请求，并解析到FootList.subscribeOn(Schedulers.io())        // 在io线程进行网络请求.observeOn(AndroidSchedulers.mainThread()) // 在主线程处理获取食品列表的请求结果.doOnNext(new Consumer<FoodList>() {@Overridepublic void accept(@NonNull FoodList foodList) throws Exception {// 先根据获取食品列表的响应结果做一些操作Log.e(TAG, "accept: doOnNext :" + foodList.toString());mRxOperatorsText.append("accept: doOnNext :" + foodList.toString()+"\n");}}).observeOn(Schedulers.io()) // 回到 io 线程去处理获取食品详情的请求.flatMap(new Function<FoodList, ObservableSource<FoodDetail>>() {@Overridepublic ObservableSource<FoodDetail> apply(@NonNull FoodList foodList) throws Exception {if (foodList != null && foodList.getTngou() != null && foodList.getTngou().size() > 0) {return Rx2AndroidNetworking.post("http://www.tngou.net/api/food/show").addBodyParameter("id", foodList.getTngou().get(0).getId() + "").build().getObjectObservable(FoodDetail.class);}return null;}}).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<FoodDetail>() {@Overridepublic void accept(@NonNull FoodDetail foodDetail) throws Exception {Log.e(TAG, "accept: success ：" + foodDetail.toString());mRxOperatorsText.append("accept: success ：" + foodDetail.toString()+"\n");}}, new Consumer<Throwable>() {@Overridepublic void accept(@NonNull Throwable throwable) throws Exception {Log.e(TAG, "accept: error :" + throwable.getMessage());mRxOperatorsText.append("accept: error :" + throwable.getMessage()+"\n");}});

4.4zip

zip 操作符可以将多个 Observable 的数据结合为一个数据源再发射出去。

Observable<MobileAddress> observable1 = Rx2AndroidNetworking.get("http://api.avatardata.cn/MobilePlace/LookUp?key=ec47b85086be4dc8b5d941f5abd37a4e&mobileNumber=13021671512").build().getObjectObservable(MobileAddress.class);Observable<CategoryResult> observable2 = Network.getGankApi().getCategoryData("Android",1,1);Observable.zip(observable1, observable2, new BiFunction<MobileAddress, CategoryResult, String>() {@Overridepublic String apply(@NonNull MobileAddress mobileAddress, @NonNull CategoryResult categoryResult) throws Exception {return "合并后的数据为：手机归属地："+mobileAddress.getResult().getMobilearea()+"人名："+categoryResult.results.get(0).who;}}).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<String>() {@Overridepublic void accept(@NonNull String s) throws Exception {Log.e(TAG, "accept: 成功：" + s+"\n");}}, new Consumer<Throwable>() {@Overridepublic void accept(@NonNull Throwable throwable) throws Exception {Log.e(TAG, "accept: 失败：" + throwable+"\n");}});

4.5interval

Observable.interval(3, 2, TimeUnit.SECONDS).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()) // 由于interval默认在新线程，所以我们应该切回主线程.subscribe(new Consumer<Long>() {@Overridepublic void accept(@NonNull Long aLong) throws Exception {mRxOperatorsText.append("interval :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");Log.e(TAG, "interval :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");}});

间隔执行操作，默认在新线程。

4.6 concatMap

与flatmap 一样，区别在于，其保证了事件的顺序。

  Observable.create(new ObservableOnSubscribe<Integer>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {e.onNext(1);e.onNext(2);e.onNext(3);}}).concatMap(new Function<Integer, ObservableSource<String>>() {@Overridepublic ObservableSource<String> apply(@NonNull Integer integer) throws Exception {List<String> list = new ArrayList<>();for (int i = 0; i < 3; i++) {list.add("I am value " + integer);}int delayTime = (int) (1 + Math.random() * 10);return Observable.fromIterable(list).delay(delayTime, TimeUnit.MILLISECONDS);}}).subscribeOn(Schedulers.newThread()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<String>() {@Overridepublic void accept(@NonNull String s) throws Exception {Log.e(TAG, "concatMap : accept : " + s + "\n");mRxOperatorsText.append("concatMap : accept : " + s + "\n");}});

实例中的输出按照，发射的顺序来执行。

4.7 doOnNext

让订阅者在接收到数据前干点事情？

 Observable.just(1, 2, 3, 4).doOnNext(new Consumer<Integer>() {@Overridepublic void accept(@NonNull Integer integer) throws Exception {mRxOperatorsText.append("doOnNext 保存 " + integer + "成功" + "\n");Log.e(TAG, "doOnNext 保存 " + integer + "成功" + "\n");}}).subscribe(new Consumer<Integer>() {@Overridepublic void accept(@NonNull Integer integer) throws Exception {mRxOperatorsText.append("doOnNext :" + integer + "\n");Log.e(TAG, "doOnNext :" + integer + "\n");}});

4.8 filter

过滤操作符，取正确的值。

 Observable.just(1, 20, 65, -5, 7, 19).filter(new Predicate<Integer>() {@Overridepublic boolean test(@NonNull Integer integer) throws Exception {return integer >= 10;}}).subscribe(new Consumer<Integer>() {@Overridepublic void accept(@NonNull Integer integer) throws Exception {mRxOperatorsText.append("filter : " + integer + "\n");Log.e(TAG, "filter : " + integer + "\n");}});

4.9 skip

接收一个long型的参数，表示跳过多少个数目的事件再开始接收。

    Observable.just(1,2,3,4,5).skip(2).subscribe(new Consumer<Integer>() {@Overridepublic void accept(@NonNull Integer integer) throws Exception {mRxOperatorsText.append("skip : "+integer + "\n");Log.e(TAG, "skip : "+integer + "\n");}});

示例中输出的结果为：345。

4.10 take

用于指定订阅者最多接收到多少数据。

  Flowable.fromArray(1,2,3,4,5).take(2).subscribe(new Consumer<Integer>() {@Overridepublic void accept(@NonNull Integer integer) throws Exception {mRxOperatorsText.append("take : "+integer + "\n");Log.e(TAG, "accept: take : "+integer + "\n" );}});

示例中输出的结果为：12。

4.11 timer

可以延迟执行一段逻辑，也可以间隔执行一段逻辑。

 Observable.timer(2, TimeUnit.SECONDS).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<Long>() {@Overridepublic void accept(@NonNull Long aLong) throws Exception {mRxOperatorsText.append("timer :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");Log.e(TAG, "timer :" + aLong + " at " + TimeUtil.getNowStrTime() + "\n");}});

4.12 just

接收一个可变参数，一次发送。

 Observable.just("1", "2", "3").subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<String>() {@Overridepublic void accept(@NonNull String s) throws Exception {mRxOperatorsText.append("accept : onNext : " + s + "\n");Log.e(TAG,"accept : onNext : " + s + "\n" );}});

4.13 single

single只会接受一个参数而snigleObserver 只会调用onError或者onSuccess

  Single.just("haha").subscribe(new SingleObserver<String>() {@Overridepublic void onSubscribe(Disposable d) {}@Overridepublic void onSuccess(String s) {Log.e("成功："+s);}@Overridepublic void onError(Throwable e) {}});

4.14 distinct

去重操作符

   Observable.just(1, 1, 1, 2, 2, 3, 4, 5).distinct().subscribe(new Consumer<Integer>() {@Overridepublic void accept(@NonNull Integer integer) throws Exception {mRxOperatorsText.append("distinct : " + integer + "\n");Log.e(TAG, "distinct : " + integer + "\n");}});

4.15 buffer

将observable中的数据按skip（步长）分成最长不超过count的buffer，然后生成一个observable。

  Observable.just(1, 2, 3, 4, 5).buffer(3, 2).subscribe(new Consumer<List<Integer>>() {@Overridepublic void accept(@NonNull List<Integer> integers) throws Exception {mRxOperatorsText.append("buffer size : " + integers.size() + "\n");Log.e(TAG, "buffer size : " + integers.size() + "\n");mRxOperatorsText.append("buffer value : ");Log.e(TAG, "buffer value : " );for (Integer i : integers) {mRxOperatorsText.append(i + "");Log.e(TAG, i + "");}mRxOperatorsText.append("\n");Log.e(TAG, "\n");}});

先取3个,每2个再取3个

4.16 debounce

过滤掉发射频率过快的数据项。

 Observable.create(new ObservableOnSubscribe<Integer>() {@Overridepublic void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Exception {// send events with simulated time waitemitter.onNext(1); // skipThread.sleep(400);emitter.onNext(2); // deliverThread.sleep(505);emitter.onNext(3); // skipThread.sleep(100);emitter.onNext(4); // deliverThread.sleep(605);emitter.onNext(5); // deliverThread.sleep(510);emitter.onComplete();}}).debounce(500, TimeUnit.MILLISECONDS).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<Integer>() {@Overridepublic void accept(@NonNull Integer integer) throws Exception {mRxOperatorsText.append("debounce :" + integer + "\n");Log.e(TAG,"debounce :" + integer + "\n");}});

4.17 defer

就是在每次订阅的时候就会创建一个新的Observable

   Observable<Integer> observable = Observable.defer(new Callable<ObservableSource<Integer>>() {@Overridepublic ObservableSource<Integer> call() throws Exception {return Observable.just(1, 2, 3);}});observable.subscribe(new Observer<Integer>() {@Overridepublic void onSubscribe(@NonNull Disposable d) {}@Overridepublic void onNext(@NonNull Integer integer) {mRxOperatorsText.append("defer : " + integer + "\n");Log.e(TAG, "defer : " + integer + "\n");}@Overridepublic void onError(@NonNull Throwable e) {mRxOperatorsText.append("defer : onError : " + e.getMessage() + "\n");Log.e(TAG, "defer : onError : " + e.getMessage() + "\n");}@Overridepublic void onComplete() {mRxOperatorsText.append("defer : onComplete\n");Log.e(TAG, "defer : onComplete\n");}});

4.18 last

取出最后一个值，参数没有值的时候的默认值。

 Observable.just(1, 2, 3).last(4).subscribe(new Consumer<Integer>() {@Overridepublic void accept(@NonNull Integer integer) throws Exception {mRxOperatorsText.append("last : " + integer + "\n");Log.e(TAG, "last : " + integer + "\n");}});

4.19 merge

将多个Observable合起来，接收可变参数，也支持使用迭代器集合。

    Observable.merge(Observable.just(1, 2), Observable.just(3, 4, 5)).subscribe(new Consumer<Integer>() {@Overridepublic void accept(@NonNull Integer integer) throws Exception {mRxOperatorsText.append("merge :" + integer + "\n");Log.e(TAG, "accept: merge :" + integer + "\n" );}});

4.20 reduce

就是一次用一个方法处理一个值，可以有一个seed作为初始值。

 Observable.just(1, 2, 3).reduce(new BiFunction<Integer, Integer, Integer>() {@Overridepublic Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception {return integer + integer2;}}).subscribe(new Consumer<Integer>() {@Overridepublic void accept(@NonNull Integer integer) throws Exception {mRxOperatorsText.append("reduce : " + integer + "\n");Log.e(TAG, "accept: reduce : " + integer + "\n");}});

示例输出结果为： 6；

4.21 scan

和reduce差不多，scan会将过程中每一个结果输出。

     Observable.just(1, 2, 3).scan(new BiFunction<Integer, Integer, Integer>() {@Overridepublic Integer apply(@NonNull Integer integer, @NonNull Integer integer2) throws Exception {return integer + integer2;}}).subscribe(new Consumer<Integer>() {@Overridepublic void accept(@NonNull Integer integer) throws Exception {mRxOperatorsText.append("scan " + integer + "\n");Log.e(TAG, "accept: scan " + integer + "\n");}});

示例输出：1 3 6

4.22 window

按照时间划分窗口，将数据发送给不同的Observable。window操作符会在时间间隔内缓存结果

   Observable.interval(1, TimeUnit.SECONDS) // 间隔一秒发一次.take(15) // 最多接收15个.window(3, TimeUnit.SECONDS).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<Observable<Long>>() {@Overridepublic void accept(@NonNull Observable<Long> longObservable) throws Exception {mRxOperatorsText.append("Sub Divide begin...\n");Log.e(TAG, "Sub Divide begin...\n");longObservable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).subscribe(new Consumer<Long>() {@Overridepublic void accept(@NonNull Long aLong) throws Exception {mRxOperatorsText.append("Next:" + aLong + "\n");Log.e(TAG, "Next:" + aLong + "\n");}});}});

4.23 PublishSubject

onNext（）会通知每个观察者

 PublishSubject<Integer> publishSubject = PublishSubject.create();publishSubject.subscribe(new Observer<Integer>() {@Overridepublic void onSubscribe(@NonNull Disposable d) {mRxOperatorsText.append("First onSubscribe :"+d.isDisposed()+"\n");Log.e(TAG, "First onSubscribe :"+d.isDisposed()+"\n");}@Overridepublic void onNext(@NonNull Integer integer) {mRxOperatorsText.append("First onNext value :"+integer + "\n");Log.e(TAG, "First onNext value :"+integer + "\n");}@Overridepublic void onError(@NonNull Throwable e) {mRxOperatorsText.append("First onError:"+e.getMessage()+"\n");Log.e(TAG, "First onError:"+e.getMessage()+"\n" );}@Overridepublic void onComplete() {mRxOperatorsText.append("First onComplete!\n");Log.e(TAG, "First onComplete!\n");}});publishSubject.onNext(1);publishSubject.onNext(2);publishSubject.onNext(3);publishSubject.subscribe(new Observer<Integer>() {@Overridepublic void onSubscribe(@NonNull Disposable d) {mRxOperatorsText.append("Second onSubscribe :"+d.isDisposed()+"\n");Log.e(TAG, "Second onSubscribe :"+d.isDisposed()+"\n");}@Overridepublic void onNext(@NonNull Integer integer) {mRxOperatorsText.append("Second onNext value :"+integer + "\n");Log.e(TAG, "Second onNext value :"+integer + "\n");}@Overridepublic void onError(@NonNull Throwable e) {mRxOperatorsText.append("Second onError:"+e.getMessage()+"\n");Log.e(TAG, "Second onError:"+e.getMessage()+"\n" );}@Overridepublic void onComplete() {mRxOperatorsText.append("Second onComplete!\n");Log.e(TAG, "Second onComplete!\n");}});publishSubject.onNext(4);publishSubject.onNext(5);publishSubject.onComplete();

RxJava 2.x 入门相关推荐

这可能是最好的RxJava 2.x 入门教程（二）
这可能是最好的 RxJava 2.x 入门教程系列专栏文章链接: 这可能是最好的 RxJava 2.x 入门教程(完结版)[推荐直接看这个] 这可能是最好的RxJava 2.x 入门教程(一) 这可 ...
【知识整理】这可能是最好的RxJava 2.x 入门教程（四）
这可能是最好的RxJava 2.x入门教程系列专栏文章链接: 这可能是最好的RxJava 2.x 入门教程(完结版)[强力推荐] 这可能是最好的RxJava 2.x 入门教程(一) 这可能是最好的R ...
RxJava 2.x入门教程
前言首先来说一下rxjava1和rxjava2的区别吧,附带一些RxJava 1升级到RxJava 2过程中踩过的一些"坑",RxJava 对大家而言肯定不陌生,其受欢迎程度不言 ...
RxJava 2.0 入门教程
RxJava 2.0 入门教程 RxJava 2.0 是来自NetFlix的开源java异步编程框架.和java 8 lambda表达式很接近,响应式编程的基本构建快是被观察对象(Observable ...
Android响应式编程（一）RxJava前篇[入门基础]
1.RxJava概述 ReactiveX与RxJava 在讲到RxJava之前我们首先要了解什么是ReactiveX,因为RxJava是ReactiveX的一种java实现. ReactiveX是Re ...
Android RxJava+Retrofit+MVP 入门总结
前言 RxJava+Retrofit+MVP的使用已经推出一段时间了,也一直想找个时间学习一下并且应用到接下来的项目中.趁放假这段时间仔细研究了一下,确实相比于其他框架的学习成本要高一些,不过功能实现 ...
这可能是最好的RxJava 2.x 入门教程学习系列
前言在网上看到一个讲rxjava2系列的文章,然后跟着学了一遍,下面是我跟着学习的代码,后续还会附上一张rxjav2学习的思维导图. github官方链接 https://github.com/Re ...
Rxjava：基础入门
定义 Rxjava基于事件流.实现异步操作. 使用引入依赖 implementation "io.reactivex.rxjava3:rxjava:3.1.2" implemen ...
深入浅出RxJava（三：响应式的好处）
在第一篇中,我介绍了RxJava的基础知识.第二篇中,我向你展示了操作符的强大.但是你可能仍然没被说服.这篇里面,我讲向你展示RxJava的其他的一些好处,相信这篇足够让你去使用Rxjava. 错误处 ...

RxJava 2.x 入门

RxJava 2.x 入门相关推荐

最新文章

热门文章