Android开发Rxjava源码解析

释放双眼,带上耳机,听听看~!

1.Rxjava本质上是一个异步操作库
2.看下Observable.create方法

  public static <T> Observable<T> create(OnSubscribe<T> f) {//OnSubscribe参数最终会被赋值到        OnSubscribe<T> onSubscribe成员变量
return new Observable<T>(hook.onCreate(f)); //返回Observable(被观察者)实例
}

看下hook,可以理解成抽象的代理类
static final RxJavaObservableExecutionHook hook = RxJavaPlugins.getInstance().getObservableExecutionHook();

看下subscribe方法

   public final Subscription subscribe(final Observer<? super T> observer) {
if (observer instanceof Subscriber) {
return subscribe((Subscriber<? super T>)observer); //观察者有两种,Observer和Subscriber,Observer最终还是会转型成Subscriber
}
return subscribe(new Subscriber<T>() {
@Override
public void onCompleted() {
observer.onCompleted();
}
@Override
public void onError(Throwable e) {
observer.onError(e);
}
@Override
public void onNext(T t) {
observer.onNext(t);
}
});
}

看下subscribe((Subscriber<? super T>)observer)方法,跟踪源码到

   private static <T> Subscription subscribe(Subscriber<? super T> subscriber, Observable<T> observable) {
if (subscriber == null) {
throw new IllegalArgumentException("observer can not be null");
}
if (observable.onSubscribe == null) {
throw new IllegalStateException("onSubscribe function can not be null.");
}
subscriber.onStart();//onStart()是个空方法,由我们自己调用的时候来实现它
if (!(subscriber instanceof SafeSubscriber)) {//SafeSubscriber是对subscriber包装,subscriber里的onCompleted()或onError(Throwable e)方法被执行后,就不会执行onNext(T args)方法
// assign to `observer` so we return the protected version
subscriber = new SafeSubscriber<T>(subscriber);
}
try {
// allow the hook to intercept and/or decorate
hook.onSubscribeStart(observable, observable.onSubscribe).call(subscriber);//完成订阅回调
return hook.onSubscribeReturn(subscriber);
} catch (Throwable e) {
.
.
.
}
return Subscriptions.unsubscribed();
}
}

看下Subscriber类源码

public abstract class Subscriber<T> implements Observer<T>, Subscription {
private final SubscriptionList subscriptions;//订阅事件的集合,保存的是所有观察者订阅的事件,如果有取消订阅的时候,SubscriptionList里会有事件被删除,如果没有订阅事件,SubscriptionList也就为空
}

看下Subscription接口

public interface Subscription {
void unsubscribe();//解绑(取消订阅)的方法
boolean isUnsubscribed();//判断是否已经解绑(取消订阅)了
}

操作符
变换:通俗讲就是将事件序列中的对象或者整个序列进行加工处理,转换成不同的事件或者事件序列
map:就是用来把一个事件转换为另一个事件的
先看下just方法

public static <T> Observable<T> just(final T value) {
return ScalarSynchronousObservable.create(value);
}

看下ScalarSynchronousObservable.create(value)方法

 public static <T> ScalarSynchronousObservable<T> create(T t) {
return new ScalarSynchronousObservable<T>(t);
}

再看下ScalarSynchronousObservable

 protected ScalarSynchronousObservable(final T t) {
super(new OnSubscribe<T>() {//OnSubscribe就是通知观察者来产生不同行为的
@Override
public void call(Subscriber<? super T> s) {
s.setProducer(createProducer(s, t));
}
});
this.t = t;
}

看下map操作符

public final <R> Observable<R> map(Func1<? super T, ? extends R> func) {
return lift(new OperatorMap<T, R>(func));//lift方法是所有操作符的核心,所有操作符内部都会用到lift相关原理,本质是对事件的处理和再发送
}

看下lift(new OperatorMap<T, R>(func))方法

 public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
return new Observable<R>(new OnSubscribe<R>() { //生成新的被观察者对象,仅仅起到代理作用
@Override
public void call(Subscriber<? super R> o) {//call方法接收外部传的观察者Subscribertry {
Subscriber<? super T> st = hook .onLift(operator).call(o);//到调用拿到我们之前生成的Observable
try {
// new Subscriber created and being subscribed with so 'onStart' it
st.onStart();
onSubscribe.call(st);//完成订阅工作
} catch (Throwable e) {
// localized capture of errors rather than it skipping all operators 
// and ending up in the try/catch of the subscribe method which then
// prevents onErrorResumeNext and other similar approaches to error handling
Exceptions.throwIfFatal(e);
st.onError(e);
}
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
// if the lift function failed all we can do is pass the error to the final Subscriber
// as we don't have the operator available to us
o.onError(e);
}
}
});
}

看下onSubscribe.call(st)方法,在OperatorMap<T, R> 类里

 @Override
public Subscriber<? super T> call(final Subscriber<? super R> o) {
return new Subscriber<T>(o) { //创建新的观察者对象,但是没有完成订阅操作
@Override
public void onCompleted() {
o.onCompleted();
}
@Override
public void onError(Throwable e) {
o.onError(e);
}
@Override
public void onNext(T t) {
try {
o.onNext(transformer.call(t));
} catch (Throwable e) {
Exceptions.throwOrReport(e, this, t);
}
}
};
}

看下OperatorMap<T, R>类里的transformer

final Func1<? super T, ? extends R> transformer;

看下 Func1<? super T, ? extends R>接口

public interface Func1<T, R> extends Function {
R call(T t); //将泛型T转换为泛型R
}

看下flatMap操作符,flatMap会将传入的事件对象转换成Observable类型,
不会直接发送这个Observable,而是将这个Observable激活让他自己开始发送事件,
每一个创建出来的Observable发送的事件,都会被汇入到同一个Observable
flatMap会将传入的事件对象转换成Observable类型,Map是传入頞事件对象装换其它的Bitamp,这是这两个操作符的区别,共同点都是进行转换的

Rxjava线程控制
Android多程编程原则
1.不要阻塞uI线程(耗时操作让子线程操作)
2.不要在UI线程之外访问UI组件(子线程更新完了,UI更新让UI操作)
解决:Handler和Asynctask
缺点:如果项目里使都使用Handler和Asynctask,主要是代码不好维护
Rxjava遵循的是线程不变的原则,就是在哪个线程调用,就在哪个线程生产事件,在哪个线程生产了事件,就在哪个线程消费事件,Rxjava通过Schedulers来进行调度

Rxjava线程Schedulers

1.Schedulers.immediate():表示的是在当前线程运行,可以理解为不切换任何线程
2.Schedulers.newThread():表示总是去启用新线程,然后在新线程中,执行相应的操作
3.Schedulers.io():io操作使用的,一般的行为模式和Schedulers.newThread()差不多,区别就是Schedulers.io()内部使用了无数量上限的线程池,可以重用更多空闲的线程,所以大多数情况下Schedulers.io()比Schedulers.newThread()效率更高
4.Schedulers.computatio():主要是计算使用的,指的是CPU密集性的计算,
5.AndroidSchedulers.mainThread():将指定的操作放到Android主线程中进行执行

Rxjava如何进行线程控制
1.subScribrOn():指定订阅观察者时所发生的线程,就是Observable的onSubscribe,激活时所处的线程,只能调用一次
2.observeOn():指定SubscribeOn所运行在的线程,事件消费所在的线程,支持多次调用

看下subScribrOn()方法

public final Observable<T> subscribeOn(Scheduler scheduler) {//返回的是Observable<T>,和Map操作符一样,都是创建一个新的观察者
if (this instanceof ScalarSynchronousObservable) {
return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);
}
return create(new OperatorSubscribeOn<T>(this, scheduler));
}

看下peratorSubscribeOn(this, scheduler)类的源码

public final class OperatorSubscribeOn<T> implements OnSubscribe<T> { //实现OnSubscribe<T>
final Scheduler scheduler;
final Observable<T> source;
public OperatorSubscribeOn(Observable<T> source, Scheduler scheduler) {//Scheduler进行线控制
this.scheduler = scheduler;
this.source = source;
}
@Override
public void call(final Subscriber<? super T> subscriber) {
final Worker inner = scheduler.createWorker();
subscriber.add(inner);
inner.schedule(new Action0() {
@Override
public void call() {
final Thread t = Thread.currentThread();
Subscriber<T> s = new Subscriber<T>(subscriber) {//根据传进来的subscriber,创建新的Subscriber
@Override
public void onNext(T t) {
subscriber.onNext(t);//根据新的Subscriber,调用onNext(t)通知到目标Subscriber,这样完成了整个subscribeOn过程
}
@Override
public void onError(Throwable e) {
try {
subscriber.onError(e);
} finally {
inner.unsubscribe();
}
}
@Override
public void onCompleted() {
try {
subscriber.onCompleted();
} finally {
inner.unsubscribe();
}
}
@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);//取消订阅
}
});
}
}

看下Worker源码

public abstract static class Worker implements Subscription {}//实现Subscription ,Subscription 前面提到过接口里含有取消订阅和判断是否取消订阅的方法,取消订阅作用就是取消订阅了以后,就不会接收被观察者所发的事件了

看下scheduler.createWorker()方法

public abstract Worker createWorker();

抽象方法,需要子类去实现的,我们选择NewThreadScheduler的实现

   @Override
public Worker createWorker() {
return new NewThreadWorker(THREAD_FACTORY);
}

看下NewThreadWorker(THREAD_FACTORY)源码

 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;
}

看下 OperatorSubscribeOn里的setProducer方法

                    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);
}
});
}
}
});
}
};

看下 inner.schedule()方法,是个抽象方法,看下NewThreadWorker的实现

 @Override
public Subscription schedule(final Action0 action) {
return schedule(action, 0, null);
}

看下schedule(action, 0, null)重载方法

 @Override
public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit) {
if (isUnsubscribed) {
return Subscriptions.unsubscribed();
}
return scheduleActual(action, delayTime, unit);
}

看下scheduleActual(action, delayTime, unit)方法

  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) {
f = executor.submit(run);//还是通过线程池来进行进一步操作
} else {
f = executor.schedule(run, delayTime, unit);//还是通过线程池来进行进一步操作
}
run.add(f);
return run;
}

看下 source.unsafeSubscribe(s)方法

   public final Subscription unsafeSubscribe(Subscriber<? super T> subscriber) {
try {
// new Subscriber so onStart it
subscriber.onStart();
// allow the hook to intercept and/or decorate
hook.onSubscribeStart(this, onSubscribe).call(subscriber);//调用call(subscriber)方法,表明整个subscribeOn方法已经完成
return hook.onSubscribeReturn(subscriber);
} catch (Throwable e) {
.
.
.
}
}

subscribeOn方法总结:
1.会新生成一个Observable
2.onSubscribe会在目标Subscriber订阅时使用传入的Scheduler的worker作为线程调度执行者
3.在对应的线程中通知原始Observable发送消息给这个过程中临时生成的Subscriber
4.这个Subscriber又会通知到目标Subscriber,从而完成我们的subscribeOn的过程

看下observeOn方法

public final Observable<T> observeOn(Scheduler scheduler) {
if (this instanceof ScalarSynchronousObservable) {
return ((ScalarSynchronousObservable<T>)this).scalarScheduleOn(scheduler);
}
return lift(new OperatorObserveOn<T>(scheduler, false));//调用了lift方法
}

看下OperatorObserveOn源码

public final class OperatorObserveOn<T> implements Operator<T, T> {//实现了Operator<T, T>接口,所以我们看下call方法
@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<T> parent = new ObserveOnSubscriber<T>(scheduler, child, delayError);
parent.init();
return parent;
}
}
}

看下ObserveOnSubscriber源码

`private static final class ObserveOnSubscriber<T> extends Subscriber<T> implements Action0 {`//实现了Subscriber观察者接口
   @Override
public void onNext(final T t) {
if (isUnsubscribed() || finished) {
return;
}
if (!queue.offer(on.next(t))) {//判断有没有成功缓存到队列
onError(new MissingBackpressureException());
return;
}
schedule();//进行真正的线程切换
}

}
看下 schedule()方法

   protected void schedule() {
if (counter.getAndIncrement() == 0) {
recursiveScheduler.schedule(this);//跟踪源码,schedule是个抽象方法,以NewThreadWorker里实现schedule抽象方法为例,最后还是会走到前面讲过的scheduleActual(action, delayTime, unit)方法
}
}

subscribeOn和observeOn
subscribeOn是通过新建Observable的方式,使用OnSubscribe类的方式去做到线程切换的
observeOn是通过operator操作符的形式去完成线程切换的,所以它的作用域和其他操作符一样,是调用observerOn之后的链路

subscribeOn和observeOn的调用次数
subscribeOn()的位置放在那里都可以,但是它只能调用一次,原因就是subscribeOn是通过新建observeOn的方式
observeOn()指定的是它之后的操作所在的线程,通过observeOn()的多次调用,程序实现了线程的多次切换

人已赞赏
Android文章

Android开发LeakCanary源码解析

2021-2-1 21:45:27

Android文章

Android开发ButterKnife源码解析

2021-2-1 22:39:53

0 条回复 A文章作者 M管理员
    暂无讨论,说说你的看法吧
个人中心
购物车
优惠劵
今日签到
有新私信 私信列表
搜索