这篇文章主要介绍了RxJava2配置及使用的示例分析，具有一定借鉴价值，感兴趣的朋友可以参考下，希望大家阅读完这篇文章之后大有收获，下面让小编带着大家一起了解一下。
RxJava2.0是一个非常棒的流式编程,采用的观察者模式思想,事件的产生者产生事间之后发送给绑定的接受者,接受顺序与发送顺序一致.
依赖:
compile 'io.reactivex.rxjava2:rxjava:2.0.1'compile 'io.reactivex.rxjava2:rxandroid:2.0.1'简单使用:
//观察者模式,这里产生事件,事件产生后发送给接受者,但是一定要记得将事件的产生者和接收者捆绑在一起,否则会出现错误Observable.create(new ObservableOnSubscribe<String>() { @Override public void subscribe(ObservableEmitter<String> e) throws Exception { //这里调用的方法会在产生事件之后会发送给接收者,接收者对应方法会收到 e.onNext("hahaha"); e.onError(new Exception("wulala")); e.onComplete(); }/*--}).subscribe(new Observer<String>() { //接受者,根据事件产生者产生的事件调用不同方法 @Override public void onSubscribe(Disposable d) { Log.e(TAG, "onSubscribe: "); } @Override public void onNext(String value) { Log.e(TAG, "onNext: " + value); } @Override public void onError(Throwable e) { Log.e(TAG, "onError: ", e); } @Override public void onComplete() { Log.e(TAG, "onComplete: "); }});我们来用图解一下这其中发生了什么事:
上游朝下游发送数据,经过subscribe使上下游产生关系,即达成订阅.
解析1:
ObservableEmitter,这是个啥东西?Emitter:顾名思义,即Rxjava的发射器,通过这个发射器,即可发送事件-----通过调用onNext,onError,onComplete方法发送不同事件.
注意:
虽然RxJava可以进行事件发送,但这并不意味着你可以随便发送,这其中需要遵循一些规则.
onNext:你可以发送无数个onNext,发送的每个onNext接受者都会接收到.
onError:当发送了onError事件之后,发送者onError之后的事件依旧会继续发送,但是接收者当接收到onError之后就会停止接收事件了.
onComplete:当发送了onComplete事件之后,发送者的onComplete之后的事件依旧会继续发送,但是接收者接收到onComplete之后就停止接收事件了.
onError事件和onComplete事件是互斥的,但是这并不代表你配置了多个onError和onComplete一定会崩溃,多个onComplete是可以正常运行的,但是只会接收到第一个,之后的就不会再接收到了,多个onError时,只会接收到第一个,第二个会直接造成程序崩溃.
解析2:
Disposable又是个啥东西,翻译之后百度告诉我这东西叫做一次性的,是用来控制发送者和接受者之间的纽带的,默认为false,表示发送者和接受者直接的通信阀门关闭,可以正常通信,在调用dispose()方法之后,阀门开启,会阻断发送者和接收者之间的通信,从而断开连接.
重载方法:
subscribe(); //表示发送者随意发送数据,接受者什么都不管,什么都不接收. subscribe(Consumer<? super T> onNext) {} //只响应onNext()事件,其他的事件忽略. subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {} //含义同上 subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {} //含义同上 subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {} //含义同上解析3:
默认情况下,发送者和接收者都运行在主线程,但是这显然是不符合实际需求的,我们在日常使用中,通常用的最多的就是在子线程进行各种耗时操作,然后发送到主线程进行,难道我们就没有办法继续用这个优秀的库了?想多了你,一个优秀的库如果连这都想不到,怎么能被称为优秀呢,RxJava中有线程调度器,通过线程调度器,我们可以很简单的实现这种效果,下面放代码.
Observable.create(new ObservableOnSubscribe<String>() { @Override public void subscribe(ObservableEmitter<String> e) throws Exception { e.onNext("hahaha"); e.onNext("hahaha"); e.onNext("hahaha"); Log.e(TAG,"运行在什么线程" + Thread.currentThread().getName()); e.onComplete(); }}).subscribeOn(Schedulers.newThread()) //线程调度器,将发送者运行在子线程 .observeOn(AndroidSchedulers.mainThread()) //接受者运行在主线程 .subscribe(new Observer<String>() { @Override public void onSubscribe(Disposable d) { Log.e(TAG, "onSubscribe: "); Log.e(TAG,"接收在什么线程" + Thread.currentThread().getName()); } @Override public void onNext(String value) { Log.e(TAG, "onNext: " + value); } @Override public void onError(Throwable e) { Log.e(TAG, "onError: ", e); } @Override public void onComplete() { Log.e(TAG, "onComplete: "); }});最终结果:
可以看到我们只加了两行代码,就实现了效果,还有比这个更优秀,更简单的么?
注意事项:
subscribeOn(),只有在第一次调用的时候生效,之后不管调用多少次,只会以第一次为准.
observeOn(),可以被调用多次,每次调用都会更改线程.
RxJava线程池中的几个线程选项
- Schedulers.io() io操作的线程, 通常io操作,如文件读写.
- Schedulers.computation() 计算线程,适合高计算,数据量高的操作.
- Schedulers.newThread() 创建一个新线程,适合子线程操作.
- AndroidSchedulers.mainThread() Android的主线程,主线程
感谢你能够认真阅读完这篇文章，希望小编分享的“RxJava2配置及使用的示例分析”这篇文章对大家有帮助，同时也希望大家多多支持编程网，关注编程网行业资讯频道，更多相关知识等着你来学习!
