编程知识 cdmana.com

Android之Rxjava2-源码解析,flutter安装包优化

public static boolean isDisposed(Disposable d) {
//判断Disposable类型的变量的引用是否为DISPOSED
//就可以判断这个连接器是否中断
return d == DISPOSED;
}

public static boolean dispose(AtomicReference<Disposable> field) {
Disposable current = field.get();
Disposable d = DISPOSED;
if (current != d) {
//把field设置为DISPOSED
current = field.getAndSet(d);
if (current != d) {
if (current != null) {
current.dispose();
}
return true;
}
}
return false;
}

}

可以看到DisposableHelper是个枚举类,并且只有一个值DISPOSED。dispose方法就是把一个原子引用的field设为DISPOSED,这就是中断状态。而isDisposed()就是根据这个标志来判断是否中断的。

再回过头来看CreateEmiiter类的onNext这些方法

@Override
public void onNext(T t) {
//省略无关代码

if (!isDisposed()) {
//如果没有dispose(),才会调用onNext()
observer.onNext(t);
}
}

@Override
public void onError(Throwable t) {
if (!tryOnError(t)) {
//如果dispose()了,会调用到这里,即最终会崩溃
RxJavaPlugins.onError(t);
}
}

@Override
public boolean tryOnError(Throwable t) {
//省略无关代码
if (!isDisposed()) {
try {
//如果没有dispose(),才会调用onError()
observer.onError(t);
} finally {
//onError()之后会dispose()
dispose();
}
//如果没有dispose(),返回true
return true;
}
//如果dispose()了,返回false
return false;
}

@Override
public void onComplete() {
if (!isDisposed()) {
try {
//如果没有dispose(),才会调用onComplete()
observer.onComplete();
} finally {
//onComplete()之后会dispose()
dispose();
}
}
}

很容易得出,

  • 如果没有dispose,observer的onNext才会被调用
  • onError与onComplete方法互斥,只能其中一个调用到,因为调用其中一个,就会把连接切断,dispose
  • 先onError后onComplete中是onComplete不会被调用,反过来的话,就会崩溃,因为onError中抛出了异常,实际上,dispose了后调用onError都会崩

再看一下操作符Map

public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
ObjectHelper.requireNonNull(mapper, “mapper is null”);
return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
}

public final class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {
final Function<? super T, ? extends U> function;

public ObservableMap(ObservableSource<T> source, Function<? super T, ? extends U> function) {
super(source);
this.function = function;
}

@Override
public void subscribeActual(Observer<? super U> t) {
source.subscribe(new MapObserver<T, U>(t, function));
}

static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
final Function<? super T, ? extends U> mapper;

MapObserver(Observer<? super U> actual, Function<? super T, ? extends U> mapper) {
super(actual);
this.mapper = mapper;
}

@Override
public void onNext(T t) {
if (done) {
return;
}

if (sourceMode != NONE) {
downstream.onNext(null);
return;
}

U v;

try {
v = ObjectHelper.requireNonNull(mapper.apply(t), “The mapper function returned a null value.”);
} catch (Throwable ex) {
fail(ex);
return;
}
downstream.onNext(v);
}


}
}

可以看到,操作符其实和上面分析的特殊情况下的一样的,这里就省略分析了。
Android之Rxjava2-源码解析,flutter安装包优化_Android

##三.Rxjava线程切换
我们一般是这么使用的

Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
emitter.onNext(1);
emitter.onComplete();
}
}).map(new Function<Integer, Integer>() {
@Override
public Integer apply(Integer integer) throws Exception {
return integer+1;
}
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<Integer>() {

});

通过subscribeOn来切换上游线程,observeOn来切换下游线程。

那么在源码中,是怎么的呢?

subscribeOn源码分析
Schedulers.io()

subscribeOn类型有好几种,这里就随便选择了Schedulers.io()来分析,别的其实都差不多的,分析了一个就行了。

@NonNull
public static Scheduler io() {
//又是hook,就相当于IO
return RxJavaPlugins.onIoScheduler(IO);
}

public final class Schedulers {

@NonNull
static final Scheduler IO;

static final class IoHolder {
static final Scheduler DEFAULT = new IoScheduler();
}

static {
//又是hook,就相当于new IOTask
IO = RxJavaPlugins.initIoScheduler(new IOTask());

}


static final class IOTask implements Callable<Scheduler> {
@Override
public Scheduler call() throws Exception {
return IoHolder.DEFAULT;
}
}
}

可以看到,最后这里就相当于new IoScheduler,先不看它的具体实现。

subscribeOn

我们继续看subscribeOn的源码

public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, “scheduler is null”);
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}

可以看到和前面一样,就相当于返回new ObservableSubscribeOn

public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;

public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
super(source);
this.scheduler = scheduler;
}

@Override
public void subscribeActual(final Observer<? super T> observer) {
final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(observer);

observer.onSubscribe(parent);

 //外层的parent.setDisposable是为了创建连接器,以便以后切断等控制的,可以只看里面
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
}

}

它的构造就是把source和scheduler两个都保存一下,在后面要用到的。

接下来我们来看订阅过程,虽然这里是线程切换,但是其实它也只是个操作符,和我们前面分析的是一样的,订阅过程和上面也是一样的,所以我们可以知道,当订阅发生后,ObservableSubscribeOn的subscribeActual方法就会被调用。

同样的,subscribeActual方法中,它把我们自定义的下游观察者observer包装成了SubscribeOnObserver对象,然后调用observer的onSubscribe方法,可以看到,目前为止,还没有发生任何的线程相关的东西,所以observer的onSubscribe()方法是运行在当前线程中的,那我们重点来看一下parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));方法。

我们先来看一下SubscribeTask类

//是ObservableSubscribeOn的内部类,实现runnable接口,看到这,我们嗅到了线程的味道
final class SubscribeTask implements Runnable {
private final SubscribeOnObserver<T> parent;

SubscribeTask(SubscribeOnObserver<T> parent) {
this.parent = parent;
}

@Override
public void run() {
//这是的source就是我们自定义的Observable对象,就是ObservableCreate
source.subscribe(parent);
}
}

可以看到,这个类非常简单,实现了Runnable接口,在run方法中调用source.subscribe(parent);,这是个链式调用,会一层一层调用上去。

再来看scheduler.scheduleDirect

这是线程切换的核心部分了,一定要仔细看

public Disposable scheduleDirect(@NonNull Runnable run) {
return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS);
}

//run就是SubscribeTask
public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
//createWorker在Schedule类中是个抽象方法,所以实现是在子类中
//所以这个方法就是在IOSchedule中实现的
//worker中可以执行runnabale
final Worker w = createWorker();

//实际上decoratedRun还是个run对象,也就是SubscribeTask
final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

//runnable和worker包装成一个DisposeTask
DisposeTask task = new DisposeTask(decoratedRun, w);

//Worker执行这个Task
w.schedule(task, delay, unit);

return task;
}

上面的代码注释已经写得非常详细了,scheduleDirect方法就是,new一个worker,然后使用这个worker来执行task线程。

再看一下IoIoScheduler中,createWorker以及shedule的过程

public Worker createWorker() {
//new一个EventLoopWorker并传一个worker的缓存池进去
return new EventLoopWorker(pool.get());
}

static final class EventLoopWorker extends Scheduler.Worker {
private final CompositeDisposable tasks;
private final CachedWorkerPool pool;
private final ThreadWorker threadWorker;

final AtomicBoolean once = new AtomicBoolean();

EventLoopWorker(CachedWorkerPool pool) {
this.pool = pool;
this.tasks = new CompositeDisposable();
//从缓存worker池中取一个worker出来
this.threadWorker = pool.get();
}

@NonNull
@Override
public Disposable schedule(@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) {
if (tasks.isDisposed()) {
// don’t schedule, we are unsubscribed
return EmptyDisposable.INSTANCE;
}

//Runnable交给threadWorker去执行
return threadWorker.scheduleActual(action, delayTime, unit, tasks);
}
}

注意的是,不同的Scheduler类会有不同的Worker实现,因为Scheduler类最终都是交由worker来执行调度的,不过分析起来差别不大。

接下来我们看worker的缓存池操作

static final class CachedWorkerPool implements Runnable {

ThreadWorker get() {
if (allWorkers.isDisposed()) {
return SHUTDOWN_THREAD_WORKER;
}
while (!expiringWorkerQueue.isEmpty()) {
//缓冲池不为空,就从缓存池中取一个threadWorker
ThreadWorker threadWorker = expiringWorkerQueue.poll();
if (threadWorker != null) {
return threadWorker;
}
}

// No cached worker found, so create a new one.
//为空就一个并返回去
ThreadWorker w = new ThreadWorker(threadFactory);
allWorkers.add(w);
return w;
}

}

再看worker的执行代码threadWorker.scheduleActual

代码跟进,会发现实现在它的父类NewThreadWorker中

public class NewThreadWorker extends Scheduler.Worker implements Disposable {
private final ScheduledExecutorService executor;

volatile boolean disposed;

public NewThreadWorker(ThreadFactory threadFactory) {
//在构造中创建一个ScheduledExecutorService对象
//可以通过它来使用线程池
executor = SchedulerPoolFactory.create(threadFactory);
}

public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, @NonNull TimeUnit unit, @Nullable DisposableContainer parent) {
//这是decoratedRun就相当于run
Runnable decoratedRun = RxJavaPlugins.onSchedule(run);

//将decoratedRun包装成一个新对象ScheduledRunnable
ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);

if (parent != null) {
if (!parent.add(sr)) {
return sr;
}
}

Future<?> f;
try {
if (delayTime <= 0) {
//线程池中立即执行ScheduledRunnable
f = executor.submit((Callable<Object>)sr);
} else {
//线程池中延迟执行ScheduledRunnable
f = executor.schedule((Callable<Object>)sr, delayTime, unit);
}
sr.setFuture(f);
} catch (RejectedExecutionException ex) {

}

return sr;
}

}

这里的executor就是使用线程池来执行任务,最终subscribeTask的run方法会在线程池中被执行,即上游的Observable的subscribe方法会在IO线程中调用了。

小结

  • Observer的onSubscribe方法运行在当前线程中,因为源码中并没有线程切换
  • 如果设置了subscribeOn(指定线程),那么Observable中的subscribe方法将会运行在指定线程中。
  • 当多个subscribeOn调用时,因为从源码可知,线程的切换是从下往上的,最后也就是链式调用的第一个切换过程,才是有效的切换

observeOn源码分析

.observeOn(AndroidSchedulers.mainThread())

AndroidSchedulers.mainThread()
同样的,我们先看AndroidSchedulers.mainThread()的源码

public static Scheduler mainThread() {
return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD);
}
private static final Scheduler MAIN_THREAD = RxAndroidPlugins.initMainThreadScheduler(
new Callable<Scheduler>() {
@Override public Scheduler call() throws Exception {
return MainHolder.DEFAULT;
}
});
private static final class MainHolder {
static final Scheduler DEFAULT
= new HandlerScheduler(new Handler(Looper.getMainLooper()), false);
}

这一段代码相信如果是看了上面的源码分析的话,一眼就能看出来,其实就相当于new HandlerScheduler(new Handler(Looper.getMainLooper()), false);,把一个主线程的Handler包装进了HandlerScheduler中。

observeOn
然后我们继续看observeOn的源码

public final Observable<T> observeOn(Scheduler scheduler) {
return observeOn(scheduler, false, bufferSize());
}
public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
ObjectHelper.requireNonNull(scheduler, “scheduler is null”);
ObjectHelper.verifyPositive(bufferSize, “bufferSize”);
return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
}

通过源码也可以知道,这里相当于直接new ObservableObserveOn

public final class ObservableObserveOn<T> extends AbstractObservableWithUpstream<T, T> {
final Scheduler scheduler;
final boolean delayError;
final int bufferSize;
public ObservableObserveOn(ObservableSource<T> source, Scheduler scheduler, boolean delayError, int bufferSize) {
super(source);
this.scheduler = scheduler;
this.delayError = delayError;
this.bufferSize = bufferSize;
}

@Override
protected void subscribeActual(Observer<? super T> observer) {
//判断是否是当前线程
if (scheduler instanceof TrampolineScheduler) {
//是当前线程的话,直接调用下游的subscribe方法
//也就是调用下一个Observable的subscibe方法
source.subscribe(observer);
} else {
//创建worker
//本例中的schedule为HandlerScheduler
Scheduler.Worker w = scheduler.createWorker();

//这里和上面分析有点类似,会将worker包装到ObserveOnObserver中
//注意:source.subscribe没有涉及到worker,所以还是在之间设置的线程中执行
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
}

}

首先,判断是否已经在要切换的线程上了,如果是的话,那么直接调用。如果不是,那么使用HandlerScheduler包装一下worker,然后通过worker来把下游的事件进行切换,直接上游订阅,不做线程操作。

我们来看ObserveOnObserver类的源码

static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T>
implements Observer<T>, Runnable {

ObserveOnObserver(Observer<? super T> actual, Scheduler.Worker worker, boolean delayError, int bufferSize) {
this.downstream = actual;
this.worker = worker;
this.delayError = delayError;
this.bufferSize = bufferSize;
}

@Override
public void onNext(T t) {
if (done) {
return;
}

if (sourceMode != QueueDisposable.ASYNC) {
//将信息存入队列中
queue.offer(t);
}
schedule();
}

void schedule() {
if (getAndIncrement() == 0) {
//在这里调用
worker.schedule(this);
}
}

void drainNormal() {
int missed = 1;

//存储消息的队列
final SimpleQueue<T> q = queue;
//这里的downstram实际就是下游的observer
final Observer<? super T> a = downstream;

for (; {
if (checkTerminated(done, q.isEmpty(), a)) {
return;
}

for (; {
boolean d = done;
T v;

try {
//从队列中取出消息
v = q.poll();
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
disposed = true;
upstream.dispose();
q.clear();
a.onError(ex);
worker.dispose();
return;
}
boolean empty = v == null;

if (checkTerminated(d, empty, a)) {
return;
}

if (empty) {
break;
}

//调用下游的Observer的onNext
a.onNext(v);
}

missed = addAndGet(-missed);
if (missed == 0) {
break;
}
}
}

@Override
public void run() {
//outputFused默认是false
if (outputFused) {
drainFused();
} else {
//所以默认调用drainNormal
drainNormal();
}
}

}

由上面链式调用的分析可以知道,source.subscribe(observer)被调用时,会调用它

CreateEmitter<T> parent = new CreateEmitter<T>(observer);
observer.onSubscribe(parent);

try {
source.subscribe(parent);
} catch (Throwable ex) {
Exceptions.throwIfFatal(ex);
parent.onError(ex);
}

可以知道,会通过发射器把上游的事件都发送到下游,然后订阅,所以ObserveOnObserver这个类中,onNext这个方法就会调用,然后执行schedule,最后执行worker.schedule(this);,因为传入的runnable是this,也就是最后线程后调用到本类的run方法,执行run,最终执行drainNormal()方法。
那我们再来看一下worker中是怎么create以及调用的。
在上面我们已经知道,这个schedule是HandlerSchedule

final class HandlerScheduler extends Scheduler {
private final Handler handler;
private final boolean async;

HandlerScheduler(Handler handler, boolean async) {
this.handler = handler;
this.async = async;
}

@Override
public Worker createWorker() {
//主线程的handler
return new HandlerWorker(handler, async);
}
}

接着看HandlerWorker的schedule方法

private static final class HandlerWorker extends Worker {
private final Handler handler;
private final boolean async;

private volatile boolean disposed;

HandlerWorker(Handler handler, boolean async) {
this.handler = handler;
this.async = async;
}

@Override
@SuppressLint(“NewApi”) // Async will only be true when the API is available to call.
public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
if (run == null) throw new NullPointerException(“run == null”);
if (unit == null) throw new NullPointerException(“unit == null”);

if (disposed) {
最后为了帮助大家深刻理解Android相关知识点的原理以及面试相关知识,这里放上相关的我搜集整理的24套腾讯、字节跳动、阿里、百度2020-2021面试真题解析,我把技术点整理成了视频和PDF(实际上比预期多花了不少精力),包知识脉络 + 诸多细节

还有?高级架构技术进阶脑图、Android开发面试专题资料?帮助大家学习提升进阶,也节省大家在网上搜索资料的时间来学习,也可以分享给身边好友一起学习。

Android之Rxjava2-源码解析,flutter安装包优化_移动开发_02

Android之Rxjava2-源码解析,flutter安装包优化_Android_03

Android之Rxjava2-源码解析,flutter安装包优化_程序员_04

点击:

 Android架构视频+BAT面试专题PDF+学习笔记》即可免费获取~

网上学习 Android的资料一大堆,但如果学到的知识不成体系,遇到问题时只是浅尝辄止,不再深入研究,那么很难做到真正的技术提升。希望这份系统化的技术体系对大家有一个方向参考。

版权声明
本文为[mb61c1dbbb44788]所创,转载请带上原文链接,感谢
https://blog.51cto.com/u_15466190/4928936

Scroll to Top