Android开源框架如何说-RxJava2

概念

• 一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库
• 响应式编程框架（扩展）
• 扩展的观察者模式

响应类型

• Observable/Observer ：常用e.onNext/onSubscribe、onNext、onError、onComplete
• Flowable/Subscriber：支持背压（背压是流速控制的一种策略）（被观察者发送速度远快于观察者处理速度的情况，告诉上游的被观察者降低发送速度的策略）
• Single/SingleObserver：只能接收一次
• Completable/CompletableObserver ：只能接收onComplete、onError
• Maybe/MaybeObserver ：Single和Completable的结合
• Subject/Processors ：即可以做Observer也可以做Observable(RxBus类似EventBus)

线程切换

• subscribeOn() 指定的就是发射事件的线程（多次指定发射事件的线程只有第一次指定的有效）
• observerOn() 指定的就是订阅者接收事件的线程（每调用一次 observerOn()，下游的线程就会切换一次）
• 默认线程选项
  • Schedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作
  • Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作
  • Schedulers.newThread() 代表一个常规的新线程
  • AndroidSchedulers.mainThread()代表Android的主线程

常用操作符

• 创建操作：Create、From、Just、Defer、Interval
• 变换操作：Map、FlatMap、ConcatMap
• 过滤操作：Debounce、Distinct、Filter、Skip、Take
• 组合操作：Merge、Zip

遇到的坑

• 越来越多的订阅，内存开销也会变得很大，最终导致OOM，onDestroy取消订阅
• 线程调度，没有用到指定线程

流程源码

• 从入口类Observable分析，Observable是一个抽象类，实现了ObservableSource接口，并留了subscribeActual这个抽象方法。ObservableSource接口只定义了subscribe()，这个方法做了一些基础判断之后直接跳转到子类的subscribeActual()，eg：ObservableCreate、ObservableXxx
• 1)创建任务链：调用操作符时候，其实只是简单的封装了一下上一级的输入source，并返回了一个Observable对象
• 2)逆向逐级订阅：subscribe(observer)->subscribeActual(observer)，在subscribeActual(observer)中 ，对应source继续向上订阅source.subscribe()，并且包装了对应操作符的observer，最后没有上游会subscribe我们创建的Observable，下一级的Observable会生成的对应的Observer订阅上一级的source
• 3)执行任务链：从ObservableEmitter.onNext开始，不断调用observer.onNext()，observer就是对应操作符在逆向逐级订阅时候包装的，最后一层的Observer就是我们调用subscribe时传入的Observer