RxJava & RxAndroid 调研
RxJava:a library for composing asynchronous and event-based programs using observable sequences for the Java VM.
异步处理库
1. 原理
2. 能做什么
3. 源码数据流分析 (同步,异步)
4. 切换线程
观察者模式
它定义对象间的一种一对多的依赖关系,当一个对象的状态发生改变时,所有依赖于它的对象都得到通知并被自动更新。
Observer Observable (Observe 守)
打比方:
A-新闻 B-读者 B订阅A,当A更新时,会通知B,B做相应反应。
还有几个词:
dispose disposable disposed
emit emitter
Schedule Scheduler
Subscribe
注册—通知—撤销注册
Subscribe - Emit - dispose
可以做的事情:
It extends the observer pattern to support sequences of data/events and adds operators that allow you to compose sequences together declaratively while abstracting away concerns about things like low-level threading, synchronization, thread-safety and concurrent data structures.
它扩展了观察者模式以支持数据/事件序列,并添加了允许您以声明方式组合序列的运算符,同时抽象出对低级线程,同步,线程安全和并发数据结构等问题的关注。
1. RxAndroid,结合sqare公司的retrofit可以很轻松的完成网络的访问。
2. 所有用到异步的地方。而且随着程序逻辑变得越来越复杂,RxJava依然能够保持简洁。
3. 加上链式调用,可以使程序更简洁。
4. RxJava有大量丰富强大的operator,可以满足你的大部分数据处理需求。
5. Java 8 lambda-friendly API 可以用lambda表达式书写
6. virtual time and schedulers for parameterized concurrency
用一个demo解释这几个名词:
{我定了一份报纸}
Because RxAndroid releases are few and far between, it is recommended you also explicitly depend on RxJava's latest version for bug fixes and new features.因为RxAndroid版本很少,所以建议您明确依赖RxJava的最新版本来修复错误和新功能。
implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'
implementation "io.reactivex.rxjava2:rxjava:2.2.6"
基本subscribe数据流
定义如下
observable
final;ObservableCreate类以及source在运行时有用,与被观察者(Observable)同时创建的。之后细讲
subscribe()方法里什么参数也不放是空实现,也就是说报纸无论出什么新闻,读者都不关心,推送过来了也不读。只关心回调。
Observer
一个接口,四个成员函数,其中三个回调方法,分别用于 响应 对应于 Observable被观察者 发送的不同事件(一一对应)
onNext:接受到一个事件
onCompleted:接受完事件后调用,只会调用一次
onError :发生错误时调用,并停止接受事件,调用一次
注:onCompleted和onError不会同时调用,只会调用其中之一,原因后述。Disposable也是待会讲。
第三步:通过订阅(subscribe)连接观察者和被观察者
subscribe数据流分析:
subscribeActual 实际订阅,属于抽象方法,由子类实现;此处的子类 = 创建被观察者(Observable)时创建的ObservableCreate类。作用:订阅时,通过接口回调 调用被观察者(Observerable) 与 观察者(Observer)的方法。原理如下:
CreateEmitter 继承了 ObservableEmitter 和 Disposable(Observable和observer的参数)。作用:发射事件。
被首先调用,是Observer复写的onSubscribe()的实现:
接着,到Observable复写的subscribe()。如下:
按顺序执行,此时流入CreateEmitter 中 重写的 Emitter的成员函数 OnNext,OnComplete等(以OnNext举例)
信息不能为空,保证dispose之后不会有输出
接着再到 observer 的 OnNext
如果走到了onComplete的情况:
DisposableHelper类里的 dispose
如果是 onError()、onComplete(),最终都会自动调用dispose(),即断开观察者 & 被观察者的连接
dispose后,观察者onNext和onComplete将不会再执行。
总结:
- 在步骤1(创建被观察者(Observable))、步骤2(创建观察者(Observer))时,仅仅只是定义了发送的事件 & 响应事件的行为;
- 只有在步骤3(订阅时),才开始发送事件 & 响应事件,真正连接了被观察者 & 观察者;即刻发送。
ObservableEmitter emitter 的成员函数
Observer 的成员函数
Disposable 提供同步和异步的取消订阅方法(断开observer 和 observable的连接)
Disposable 的成员函数
以上数据发射和观察者通知都在同一个线程,串行的,未发生线程切换。但RxJava主要功能是提供异步:
异步与链式编程
链式编程:可以通过"点"语法,将需要执行的代码块连续的书写下去,使得代码简单易读,书写方便。
RxJava用 Scheduler 实现异步。Scheduler(Schedule),控制线程。
没有设置的时候,RxJava遵循哪个线程产生就在哪个线程消费的原则,也就是说线程不会产生变化,始终在同一个。然后我们一般使用RxJava都是后台执行,前台调用,本着这个原则,我们需要调用observeOn(AndroidSchedulers.mainThread())。
首先在主线程声明你的 Observable
随即会在新线程上执行Observable,并通过主线程上的onNext发出结果。
Schedulers.io()是子线程,这里也可以用Schedulers.newThread(),只不过io线程可以重用空闲的线程,因此多数情况下 io() 比 newThread() 更有效率。
数据流分析:
创建被观察者ObservableCreate
和上面的一样
observeOn()
verifyPositive里只是判断缓冲区是否没有,创建ObservableObserveOn对象:
接着看subscribeOn,被观察者执行以及信息回传的地方
开辟/切换线程(此时是io)
被观察者ObservableSubscribeOn, 多了对线程的控制,和Scheduler。
订阅过程,subscribe方法:
我们发现,其实先走的是ObservableSubscribeOn,在哪里订阅的类
看一下Scheduler里面有什么:
找到了我们定义的onSubscribe:
Scheduler 本质上就是用来调度 Runnable 的,支持立即、延时和周期形式的调用,而 Worker 是任务的最小单元的载体。
一个或者多个 Worker 对应一个ScheduledThreadPoolExecutor对象。通过 Worker 来调度任务。
- 同一个 Worker 创建的 Task 都会确保串行,且立即执行的任务符合先进先出原则。
- Worker 绑定了调用了他的方法的 Runnable,当该 Worker 取消时,基于他的 Task 均被取消
小结:因此当有操作符需要使用 Scheduler 时,可以通过 Worker 来将一系列的 Runnable 统一的调度和取消
我们其实发现其他两个subscribeActual 不在主线程执行;
最后,回到了ObservableCreate 的 subscribeActual:
subscribe的方法重载(如果读者只关心onNext方法里的内容, 减少代码)
在RxJava中, 已经内置了很多线程选项供我们选择:
Schedulers.io() 代表io操作的线程, 通常用于网络信息交互,读写文件等io密集型的操作(可以重用空闲线程,效率更高)
Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作
Schedulers.newThread() 代表一个常规的新线程
AndroidSchedulers.mainThread() 代表Android的主线程
在RxJava内部使用的是线程池来维护这些线程,所有效率也比较高
References:
http://reactivex.io/RxJava/2.x/javadoc/overview-summary.html
https://www.jianshu.com/p/b742526c7dec
https://www.jianshu.com/p/bb9d7a0fb0a7
https://www.jianshu.com/p/cd3557b1a474
线程间的*切换
最能体现RxJava优势的地方。
from:数组
flatmap:嵌套的 Observable ,铺平初始的对象,统一路径分发给回调方法
看一个 map() 的例子:
它不仅可以针对事件对象,还可以针对整个事件队列
References:
http://gank.io/post/560e15be2dca930e00da1083
————未完待续————
Retrofit+RxJava的上网模式
Retrofit是目前主流的网络请求框架,功能强大,操作便捷。
RxJava是实现异步操作的库。可在线程间快速切换,同时提供许多操作符,使一些复杂的操作代码变得清晰有条理。
两者结合使用后,使得网络请求更加简洁,尤其在嵌套请求等特殊场景大有作为。
RxJava容易内存泄漏
熟悉RxJava的同学,当我们开启一个异步任务时,通常需要在Activity/Fragment销毁时,及时关闭异步任务,否则就会有内存泄漏的。
一般的做法是订阅成功后,拿到Disposable对象,在Activity/Fragment销毁时,调用Disposable对象的dispose方法,将异步任务中断,也就是中断RxJava的管道,代码如下:
Disposable disposable = Observable .interval(0, 1, TimeUnit.SECONDS) //开启一个定时器 .subscribe(aLong -> { }); //Activity/Fragment销毁时,中断RxJava管道 if(disposable != null&& !disposable.isDisposed) { disposable.dispose; }