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

Posted 2020-12-29 想不起来的角落

一、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的方法创建的