Android 常用开源框架源码解析系列Rxjava 异步框架 _源码解析

【Android 常用开源框架源码解析系列Rxjava 异步框架】相逢意气为君饮，系马高楼垂柳边。这篇文章主要讲述Android 常用开源框架源码解析系列Rxjava 异步框架相关的知识，希望能为你提供帮助。
一、Rxjava的产生背景一、进行耗时任务传统解决办法：传统手动开启子线程，听过接口回调的方式获取结果传统解决办法的缺陷：随着项目的深入、扩展。代码量的增大会产生回调之中套回调的，耦合度高度增加的不利场景。对代码维护和扩展是很严重的问题。 RxJava本质上是一个异步操作库优点：使用简单的逻辑，处理复杂，困难的异步操作事件库; 在一定程度上替代handler、AsyncTask等等二、传统的观察者模式使用场景 1、一个方面的操作依赖于另一个方面的状态变化 2、如果在更改一个对象的时候，需要同时连带改变其他的对象（不确定有多少对象需要改变） 3、当一个对象必须通知其他的对象，但是又希望这个对象和其他被通知的对象是松散耦合度的关系在App开发过程中，有一个对象的状态数据需要时常关注，很多个页面的Ui都跟这个对象又有绑定关系。当这个对象发生改变的时候就需要通知所有跟他有关系的Ui都进行相应的改变。这种情况下就是一种观察者模式的使用场景。简单来说： A对象对B对象的数据高度敏感，当B对象变化的一瞬间，A对象要做出反应。这时候A对象就是观察者，B对象就是被观察者观察者模式说白了就是众多的观察者对被观察者的数据高度敏感变化的自身的一种反应。其反应的是一种多对一的关系。组成：（一）< interface> Observerable 被观察者接口 a、registerObserver() ：将观察者注册到被观察者当中，是一个订阅方法 b、removeObserver()：将观察者从被观察者中移除，取消订阅 c、notifyObservers()：当被观察者状态改变的时候，该方法就会被调用 ***** ps：内部会调用观察者的 update() 函数，来通知观察者做出相应的数据改变，依次循环遍历整个观察者数量并获取到观察者并调用update()方法进行相应的更新操作（二）< class> ConcreteObserverable 被观察者具体的实现实现了被观察者接口中的abc方法，并且定义了一个List< Observer> observers 用来保存注册好的观察者对象的 ps: 由于集合的范型参数是接口类型所以不能是具体的Observer 实现类，只能是Observer的接口接口的定义和设计要为了以后的拓展而考虑解析：这样做的原因让一个被观察者可能会有多个实现类的观察者都有可能实现了Observerable 这个接口。这样就能把观察者和被观察者通过List 这个集合进行解耦（三）< interface> Observer 观察者接口 Update() :接口与被观察者的notifyObservers（）关联进行相应的数据变化（四）< class > ConcreteObserver 具体的观察者实现 update() 或是其他方法三、Rxjava观察者模式和基本用法 Rxjava 四要素 1、被观察者 2、观察者 3、订阅 4、事件事件响应式编程是基于异步数据流概念的编程模式。响应式编程的一个核心概念事件步骤 1、创建被观察者：create rxjava中：决定什么时候触发事件，以及决定触发怎样的事件 //第一步、创建被观察者：Create //Observable 也会转换成subscriber进行响应的处理 Observable observable = Observable.create(new Observable.OnSubscribe< String> () { //按代码当中的顺序进行响应式调用 //这里的Subscriber就是观察者，在被观察者的回调中调用了观察者的方法实际上就是一种事件的传递 @Override public void call(Subscriber< ? super String> subscriber) { subscriber.onNext("1"); subscriber.onNext("2"); subscriber.onCompleted(); } }); //第二种创建被观察者对象的方法： //通过just 方法来创建被观察者对象 Observable observableJust = Observable.just("1","2”); //最多有10个String参数 //第三种创建，通过from方法,把参数作为字符数组，然后添加到参数里 String[] parameters = {“1”,”2"}; Observable observableFrom = Observable.from(parameters); //第二步、创建观察者：observer //第二步、创建观察者Observer,决定事件触发会有怎样的行为 Observer< Object> observer = new Observer< Object> () { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } //实际就是传统观察者模式中的update()方法 @Override public void onNext(Object o) { } }; //第三步，订阅，通过被观察者.subscribe(观察者) public void doRxjava() { //第三部，订阅 observable.subscribe(observer); } ps：注意是被观察者订阅观察者！为了通过流式api 进行不同的操作符操作、线程控制都能通过链式调用来完善。四、Rxjava创建Observable & observer 1、Observable(被观察者) 2、OnSubscribe 对象：被观察者用来通知观察者的notifyObservers() 函数 3、Subscriber 观察者 4、subscribe() ；通过该方法完成观察者与被观察之间的订阅 A、创建被观察者、Observable：同时创建Observable 内部的一个onSubscribe对象作为参数传递到create()方法当中 Observable observable = Observable.create(new Observable.OnSubscribe< String> () //内部传入一个OnSubscribe参数，最终会赋值给Observable的成员变量 onSubscribe public static < T> Observable< T> create(OnSubscribe< T> f) { ps：hook 可以被看作是一抽象的代理类，该代理类默认情况下不会对onSubscribe做任何处理。 static final RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook(); //create（）构造了一个新的被观察者Observable对象，同时将参数赋值给Observable的成员变量 onSubscribe，生成被观察者对象 return new Observable< T> (hook.onCreate(f)); } B、创建观察者、Subscriber or Observer ：实现了Observer接口，和Subscription接口； Subscription接口的实现： void unsubscribe(); //进行接触绑定——不再有订阅事件了，订阅事件列表为空了调用发方法 boolean isUnsubscribed(); //判断是否接触绑定，判断是否已经取消了订阅事件 public abstract class Subscriber< T> implements Observer< T> , Subscription private final SubscriptionList subscriptions; //订阅事件的集合，在这个集合List当中保存了所有这个观察者的订阅事件，当取消订阅的时候，该List会有事件被删除 C、订阅关系：调用observable 内部的subscribe() 完成订阅 public final Subscription subscribe(final Observer< ? super T> observer) { //传入一个observer ，调用subscribe，最终转型成subscribe if (observer instanceof Subscriber) { return subscribe((Subscriber< ? super T> )observer); } private static < T> Subscription subscribe(Subscriber< ? super T> subscriber, Observable< T> observable) { … subscriber.onStart(); //空方法实现，需要的时候自己调用并实现 if (!(subscriber instanceof SafeSubscriber)) { //将subscriber 包装成SafeSubscriber ps：在SafeSubscriber()中执行了onCompled()、和onError()方法，就不会再执行onNext()方法 subscriber = new SafeSubscriber< T> (subscriber); } … //调用完call方法，意味着完成了一次订阅 hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber); return hook.onSubscribeReturn(subscriber); ps：完成订阅后默认会触发 Observable.Onsubscribe< String> 中的call 函数五、Rxjava 的操作符变换：就是将事件序列中的对象或整个序列进行加工再处理，转换成不同的事件或是事件序列 map操作符：就是用来把一个事件转换为另一个事件 /** * 通过被观察者Observabled 调用just方法创建被观察者并传入图片的路径地址，调用map操作符，对原来的观察者进行数据流的变化操作。将String类型的图片路径转换成bitmap，来完成map操作符的调用，map操作符会创建一个新的Observable 对象然后再链式调用subscribe完成订阅关系 */ private void map() { Observable.just("map/image/map.png") //通过map的链式调用，将String转换成bitmap对象 .map(new Func1< String, Bitmap> () { //Func1 是Rxjava中的接口，用于包装含有参数的方法， // func1中第一个参数的类型就代表Observable发射当前的类型；第二个参数是String类型将要转换的类型 @Override public Bitmap call(String filepath) { return getBitmap(filepath); } }) .subscribe(new Action1< Bitmap> () { @Override public void call(Bitmap bitmap) { //... } }); } map（）函数接受一个Func1类型的参数，然后把这个Func1应用到每一个由Observable发射的值上，将发射的值转换为我们期望的值。将参数重String类型参数转换成Bitmap 并返回六、Rxjava 的map操作符的原理 lift()方法是Rxjava 所有操作符的核心方法 public final < R> Observable< R> map(Func1< ? super T, ? extends R> func) { return lift(new OperatorMap< T, R> (func)); } ps：OperatorMap 是实现了Operator操作符的一个接口 public final class OperatorMap< T, R> implements Operator< R, T> 在该类中核心方法call方法内，接受外部传递给它的subscriber观察者 public Subscriber< ? super T> call(final Subscriber< ? super R> o) { return new Subscriber< T> (o) { ... @Override public void onNext(T t) { try { //通过调用onNext()完成观察数据流的转化 o.onNext(transformer.call(t)); } catch (Throwable e) { Exceptions.throwOrReport(e, this, t); } } … } 而transformer 是定义在func1 这个接口下的,通过Func1类中 call()方法的调用完成T —> 转换成 R的操作 final Func1< ? super T, ? extends R> transformer; //transformer的作用就是将范型< T , > 转换成范型 < ,R> public interface Func1< T, R> extends Function { R call(T t); } //基本所有操作符内部都会用到lift ()函数的内部相关原理； lift ：本质上是针对事件序列的处理和再发送 lift（）： public final < R> Observable< R> lift(final Operator< ? extends R, ? super T> operator) { //1、首先内部生成了一个新的Observable 被观察者对象并返回 ps：new Observable——代理主要负责接收原始的Observable 发出的事件，当创建好了 new Observable 会将其发送给下面的 Subscriber< ? super T> st 让其进行处理 returnnew Observable< R> (new OnSubscribe< R> () { @Override public void call(Subscriber< ? super R> o) { try { //在新对象的OnSubscribe对象当中，通过call方法调用拿到之前的生成的Observable，生成一个新的Subscriber对象 Subscriber< ? super T> st = hook.onLift(operator).call(o); try { // new Subscriber created and being subscribed with so ‘onStart‘ it st.onStart(); //将新建的Subscriber作为参数传递到call 方法中，在这给call方法中完成了订阅工作 onSubscribe.call(st); } catch (Throwable e) { Exceptions.throwIfFatal(e); st.onError(e); } } catch (Throwable e) { Exceptions.throwIfFatal(e); o.onError(e); } } }); } 七、Rxjava 的flatmap操作符 flatmap 和map有一个共同点：均是用来进行事件转化的 map 是将 String 类型 ——转化——> bitmap ，一一对应，映射成一个新的事件 flatmap 是将 String 类型——转化——> Observable————将所有的事件转化成一个Observable然后由这个Observable进行统一的事件分发 /** * flatMap -输入URi地址返回Uri 列表 */ private Subscription processNetAddress() { return Observable.just( "http://www.xxx", "http://www.yyy", "http://www.zzz" ) //1、将传入的String类型事件对象，转换成Observable()类型对象 .flatMap(new Func1< String, Observable< String> > () { //2、不会直接发送这个Observable ，而是将这个Observable激活让他自己开始发送事件 @Override public Observable< String> call(String s) { return null; } }) //3、每一个创建出来的Observable发送的事件，都被汇入同一个Observable //接收到上面一连串的字符串完成输出 .subscribe(new Action1< String> () { @Override public void call(String s) { } }); } 八、Rxjava 的线程控制在默认不指定线程的情况下，Rxjava 遵循的是线程不变的原则。也就算说在哪个线程调用的subscribe ()订阅方法，就会在哪个线程生产事件；所以在哪个线程生产了事件，就在哪个线程消费事件 Schedulers——线程控制符 Rxjava 通过该类进行线程的调度 Schedulers.immediate() 默认情况下，在当前线程运行，不切换任何线程默认操作 Schedulers.newThread() 总是启用新线程，在新线程中执行相应的操作 Schedulers -io() 执行IO 流操作，读写文件之类的 ps：区别于newThread（），io()内部实现是一个无数量的线程池，可以更好的利用线程效率,优于newThread() Schedulers-computation() cpu密集计算使用的Schedulers androidSchedulers.mainThread() 将指定的操作放在Android 的主线程中执行线程控制 1、subscribeOn（）指定subscribe 订阅观察者时候所发生的线程，也就是Observerable内部的OnSubscribe()被激活时候所处的线程，通知各个观察者开始执行相应的操作 2、observeOn（）指定subscribeOn（）所运行在的线程，事件消费所在的线程 public void doThreadWithScheduler() { Observable.just("x", "y", "z”) //前面通过just()方法，所创建的3个事件内容的发出，会在io线程发出，后台线程进行数据读取 .subscribeOn(Schedulers.io()) //指定了subscribeOn订阅的时候call方法里函数将发生在主线程中，主线程显示数据 .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Action1< String> () { @Override public void call(String s) { System.out.println("thread" + s); } }); } public void doWeatherCompute() { //create 方法创建Observable对象，内部传入OnSubscribe对象，该方法可以被理解成为notifyObservers()方法，由它通知观察者去进行相应的操作，也就是调用观察者的 onNext()操作 Observable.create(new Observable.OnSubscribe< String> () { @Override public void call(Subscriber< ? super String> subscriber) { subscriber.onNext(""); // ... } }).subscribe(new Action1< String> () { @Override public void call(String s) { } }); 通过线程Schedulers控制符+subscribeOn（）和observeOn（）完成主线程和子线程间的工作切换，替代复杂，易发生错误的newThreard（）和handler（） public void doWeatherCompute() { Observable.create(new Observable.OnSubscribe< String> () { @Override public void call(Subscriber< ? super String> subscriber) { subscriber.onNext(""); // ... } }) //指定subscribe()，发生在io线程 .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Subscriber< String> () { //订阅上面的call（）方法发生在io线程，可以进行耗时操作 @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(String s) { //ui 显示工作的逻辑在onNext() } }); } } 思考：subscribeOn 和 observeOn 各能调用几次呢？ 1/无限答： observeOn（）指定的是之后的工作所在的线程，因此有多次切换线程的需求，只要在每一个切换线程的位置调用一次observeOn（）就能满足需求。所以说observeOn（）支持多次调用！ subscribeOn（）位置可以放在observeOn（）前后都可以。但是！！ subscribeOn（）只能调用一次。 ————subscribeOn（）源码 public final Observable< T> subscribeOn(Scheduler scheduler) {//返回值是一个Observable对象，返回创建一个新的被观察者，通过新的被观察者进行下面的事件 if (this instanceof ScalarSynchronousObservable) { return ((ScalarSynchronousObservable< T> )this).scalarScheduleOn(scheduler); } return create(new OperatorSubscribeOn< T> (this, scheduler)); } ps：OperatorSubscribeOn implements OnSubscribe< T> 实现了该接口 public OperatorSubscribeOn(Observable< T> source, Scheduler scheduler) { this.scheduler = scheduler; this.source = source; } public void call(final Subscriber< ? super T> subscriber) { //用于线程控制实现了Subscription接口用于看是否取消订阅的操作，取消后就不会接受观察者发送的各类事件了 final Workerinner = scheduler.createWorker(); subscriber.add(inner); inner.schedule(new Action0() { @Override public void call() { final Thread t = Thread.currentThread(); //根据传递的subscriber 创建一个新的Subscriber Subscriber< T> s = new Subscriber< T> (subscriber) { @Override public void onNext(T t) { //根据新的subscriber 通知到目标subscriber并调用其onNext()方法 subscriber.onNext(t); } … //内部对线程进行一些判断 @Override public void setProducer(final Producer p) { subscriber.setProducer(new Producer() { @Override public void request(final long n) { if (t == Thread.currentThread()) { p.request(n); } else { //在该方法里进行线程的控制操作 inner.schedule(new Action0() { @Override public void call() { p.request(n); } }); } } }); } source.unsafeSubscribe(s); //source实际上是一个Obsrvable对象 … } 同理,观察者 Subscriber 也是现实了Subscription、和Observer接口的原因是当我们的观察者Subscriber取消订阅的时候，将持有事件列表中的所有Subscription 订阅全部取消。也就不会再接受订阅事件。 createWorker（）： public Worker createWorker() { return new NewThreadWorker(THREAD_FACTORY); //通过newThreadWorker()完成创建 } NewThreadWorker（）： public NewThreadWorker(ThreadFactory threadFactory) { //通过线程池创建并操作线程 ScheduledExecutorService exec = Executors.newScheduledThreadPool(1, threadFactory); // Java 7+: cancelled future tasks can be removed from the executor thus avoiding memory leak boolean cancelSupported = tryEnableCancelPolicy(exec); if (!cancelSupported & & exec instanceof ScheduledThreadPoolExecutor) { registerExecutor((ScheduledThreadPoolExecutor)exec); } schedulersHook = RxJavaPlugins.getInstance().getSchedulersHook(); executor = exec; } inner.schedule（）：是通过一个抽象类实现的abstract Subscription schedule，在其具体的实现类之一NewThreadWoker中通过schedule()方法实现具体逻辑 public ScheduledAction scheduleActual(final Action0 action, long delayTime, TimeUnit unit) { Action0 decoratedAction = schedulersHook.onSchedule(action); ScheduledAction run = new ScheduledAction(decoratedAction); Future< ?> f; if (delayTime < = 0) {//通过executor 的方法可以判断其还是根据线程池完成操作 f = executor.submit(run); } else { f = executor.schedule(run, delayTime, unit); } run.add(f); return run; } source.unsafeSubscribe（）： public final Subscription unsafeSubscribe(Subscriber< ? super T> subscriber) { try { // new Subscriber so onStart it subscriber.onStart(); // allow the hook to intercept and/or decorate //在这里调用call()函数表明整个subscribeOn（）操作已经完成 hook.onSubscribeStart(this, onSubscribe).call(subscriber); return hook.onSubscribeReturn(subscriber); } catch (Throwable e) { // special handling for certain Throwable/Error/Exception types Exceptions.throwIfFatal(e); // if an unhandled error occurs executing the onSubscribe we will propagate it try { subscriber.onError(hook.onSubscribeError(e)); } catch (Throwable e2) { Exceptions.throwIfFatal(e2); // if this happens it means the onError itself failed (perhaps an invalid function implementation) // so we are unable to propagate the error correctly and will just throw RuntimeException r = new RuntimeException("Error occurred attempting to subscribe [" + e.getMessage() + "] and then again while trying to pass to onError.", e2); hook.onSubscribeError(r); throw r; } return Subscriptions.unsubscribed(); } } SubscribeOn方法小结 1、会新生成一个Observable 2、在其内部，onSubscribe对象会在目标Subscriber订阅时候使用传入的Scheduler的worker作为线程调度执行者 3、在对应的线程中通知原始Observable发送消息给这个过程中临时生成的Subscriber 4、这个Subscriber又会通知到目标Subscriber，从而完成我们的subscribeOn的过程 ————observeOn（）源码 public final Observable< T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) { if (this instanceof ScalarSynchronousObservable) { return ((ScalarSynchronousObservable< T> )this).scalarScheduleOn(scheduler); } //最终通过lift() 函数操作 return lift(new OperatorObserveOn< T> (scheduler, delayError, bufferSize)); } class OperatorObserveOn< T> implements Operator< T, T> 实现了操作符的接口 … @Override public Subscriber< ? super T> call(Subscriber< ? super T> child) { if (scheduler instanceof ImmediateScheduler) { // avoid overhead, execute directly return child; } else if (scheduler instanceof TrampolineScheduler) { // avoid overhead, execute directly return child; } else { //完成线程切换实际的对象ObserveOnSubscriber ObserveOnSubscriber< T> parent = new ObserveOnSubscriber< T> (scheduler, child, delayError, bufferSize); parent.init(); return parent; } } private static final class ObserveOnSubscriber< T> extends Subscriber< T> implements Action0{ … @Override public void onNext(final T t) { if (isUnsubscribed() || finished) { return; } //将结果缓存到队列当中 if (!queue.offer(on.next(t))) { onError(new MissingBackpressureException()); return; } schedule(); //开启真正的线程切换 } protected void schedule() { if (counter.getAndIncrement() == 0) { recursiveScheduler.schedule(this); } } } public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit) { if (isUnsubscribed) { return Subscriptions.unsubscribed(); } return scheduleActual(action, delayTime, unit); } //线程调度方法，同上面的方法一样也是使用executor并发库来进行线程调度 public ScheduledAction scheduleActual(final Action0 action, long delayTime, TimeUnit unit) { ... if (delayTime < = 0) { f = executor.submit(run); } else { f = executor.schedule(run, delayTime, unit); ... } SubscribeOn ()和 ObserveOn() subscribeOn ()是通过新建Observable的方式，使用OnSubscribe()类的方式去做到线程切换的。不断的调用subscribeOn()实际上最终只会使用第一个subscribeOn()的方法。 observeOn()是通过opeartor 操作符的形式去完成线程切换的，所以它的作用域和其他操作符一样，是调用observeOn(0之后的链路 1、observeOn()指定的是它之后的操作所在的线程，通过observeOn()的多次调用，程序实现了线程的多次切换； 2、subscribeOn()的位置放在哪里都可以，但它是只能调用一次的，原因就是subscribeOn是通过新建Observable的方法创建的

Android 常用开源框架源码解析系列Rxjava 异步框架

推荐阅读

智慧保险的快速发展,未来的保险公司将是什么模样呢？

获取oss的pdf文件，返回文件输出流

2023重庆涪陵3月赏花时间+地点涪陵的花

wnda是什么轮胎 wnda是什么轮胎里面有一层胶

时光与岁月优美短句关于时光岁月的唯美句子

mysql怎么设置字段默认值为0 mysql设置字段为小时

魔兽世界桃乐豆详细介绍

食色,性也究竟是什么意思食色性也是什么意思

什么材质的行李箱好一点什么材质的行李箱好

华为网盘会被和谐吗,华为网盘有什么用？

成都徒步旅行～懒鱼第四天作业

关系胜于一切，关系就是温暖

g三星 Galaxy S22 / S22 +图赏：老设计焕发新感觉

北京医保提到4000北京医保提到4000的原因（北京医保提到4000北京医保

文字代码生成器，谁有日志彩字生成代码器

在网店转让平台上如何让自己的店铺快速出售？有哪些小技巧？

十七岁，我曾有一个朋友

潇潇雨歇是什么意思潇潇雨歇解释

四轴数控车床怎么编程，请大师教教我这个用四轴怎么编程

怎么判断水果是否有毒

Android 常用开源框架源码解析 系列Rxjava 异步框架

推荐阅读

Android 常用开源框架源码解析系列Rxjava 异步框架