上一篇文章：Rxjava2源码浅析（一）
分析了最基础的一套流程，今天呢就略加一些常用的操作吧。
使用范例
上次我们在建立关系的时候就是这样光秃秃的一句话
observable.subscribe(observer);
这在平时使用的时候显然是不够用的，Rxjava的优势是什么？切换线程肯定要算一个啊，所以我们看一下下面这种使用方法。
observable.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(observer);
这样就完成了在newThread中运行被观察者，在主线程中观察。下面就进入源码时间了~
先从参数开始看起：
public static Scheduler newThread() {
return RxJavaPlugins.onNewThreadScheduler(NEW_THREAD);
}
然后又调用了
@NonNull
public static Scheduler onNewThreadScheduler(@NonNull Scheduler defaultScheduler) {
Function<? super Scheduler, ? extends Scheduler> f = onNewThreadHandler;
if (f == null) {
return defaultScheduler;
}
return apply(f, defaultScheduler);
}
这里的onNewThreadHandler又是什么呢？
static volatile Function<? super Scheduler, ? extends Scheduler> onNewThreadHandler;
可以看到是一个类型转换的Funtion<>，暂时不用去管它，因为现在肯定是为空的，所以就会返回我们传进去的defaultScheduler也就是NEW_THREAD，这个是Schedulers内的一个变量
static final Scheduler NEW_THREAD;
这个NEW_THREAD又是什么呢？跟踪它在Schedulers内部的调用。
NEW_THREAD = RxJavaPlugins.initNewThreadScheduler(new Callable<Scheduler>() {
@Override
public Scheduler call() throws Exception {
return NewThreadHolder.DEFAULT;
}
});
发现了有个static字段内部对它进行了初始化操作。
还是从参数开始分析，发现这是一个Callable对象，和Runable大同小异不过可以返回结果，这里就在call方法中返回了一个Scheduler对象NewThreadHolder.DEFAULT
static final class NewThreadHolder {
static final Scheduler DEFAULT = new NewThreadScheduler();
}
这里就将创建线程的任务从Schedulers移到了NewThreadScheduler
static {
int priority = Math.max(Thread.MIN_PRIORITY, Math.min(Thread.MAX_PRIORITY,
Integer.getInteger(KEY_NEWTHREAD_PRIORITY,Thread.NORM_PRIORITY)));
THREAD_FACTORY = new RxThreadFactory(THREAD_NAME_PREFIX, priority);
}
private static final RxThreadFactory THREAD_FACTORY;
public NewThreadScheduler() {
this(THREAD_FACTORY);
}
public NewThreadScheduler(ThreadFactory threadFactory) {
this.threadFactory = threadFactory;
}
我们看到NewThreadScheduler初始化中还包含着RxThreadFactory的初始化，至于它的作用到现在还不得而知，暂且搁置，回到主方法中。接着从从subscribeon开始分析：
public final Observable<T> subscribeOn(Scheduler scheduler) {
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
return RxJavaPlugins.onAssembly(new ObservableSubscribeOn<T>(this, scheduler));
}
上来第一行还是老样子先进行非空判断，然后第二行这个我们是不是也看着很眼熟呢？没错。。。
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
ObjectHelper.requireNonNull(source, "source is null");
return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}
这跟第一个步骤中的create长的简直一模一样。所以重点就来到了
new ObservableSubscribeOn<T>(this, scheduler)
该方法为：
public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
super(source);
this.scheduler = scheduler;
}
暂时没啥好说的。。。先跳过。。。
subscribeon就先搁置一下，
然后看observeron,参数就不分析了，和subscribeon的流程大同小异,唯一不同的就是其中是通过Looper.getMainLooper()来得到主线程的。后面有机会的话也会专门来写一篇文章记录一下自己关于Looper和Handler的线程和消息机制的学习。
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.CUSTOM)
public final Observable<T> observeOn(Scheduler scheduler) {
return observeOn(scheduler, false, bufferSize());
}
这里的buffersize
public static int bufferSize() {
return Flowable.bufferSize();
}
public static int bufferSize() {
return BUFFER_SIZE;
}
static final int BUFFER_SIZE;
static {
BUFFER_SIZE = Math.max(16, Integer.getInteger("rx2.buffer-size", 128));
}
可以看到这个buffersize就是一个缓冲区的大小，一个int类型的参数，不过和flowable牵扯在了一起，这个flowable也是Rxjava2的新特性--背压，这里先不多讲了，都放到后面的文章来说。主要来看这个observeon
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));
}
它的前两行
ObjectHelper.requireNonNull(scheduler, "scheduler is null");
ObjectHelper.verifyPositive(bufferSize, "bufferSize");
都是验证合理性的。第一行出现很多次就不用说了。第二行
public static int verifyPositive(int value, String paramName) {
if (value <= 0) {
throw new IllegalArgumentException(paramName + " > 0 required but it was " + value);
}
return value;
}
就是说我们的buffersize一定要是一个正数，这也是合情合理且应该的。
然后就又看到我们的老朋友了RxJavaPlugins.onAssembly。。。
我们前面的分析也有了经验，所以这里会return 里面的参数 new ObservableObserveOn< T >(this, scheduler, delayError, bufferSize)向上转型为一个Observable，之前也分析过，
ObservableObserveOn < T > 中会重写
@Override
protected void subscribeActual(Observer<? super T> observer) {
if (scheduler instanceof TrampolineScheduler) {
source.subscribe(observer);
} else {
Scheduler.Worker w = scheduler.createWorker();
source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
}
}
这里的scheduler.createWorker是AndroidScheduler.MainThread
@Override
public Worker createWorker() {
return new HandlerWorker(handler);
}
而这个createWork（）
@Override
public Worker createWorker() {
return new HandlerWorker(handler);
}
private static final class HandlerWorker extends Worker {
private final Handler handler;
private volatile boolean disposed;
HandlerWorker(Handler handler) {
this.handler = handler;
}
@Override
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) {
return Disposables.disposed();
}
run = RxJavaPlugins.onSchedule(run);
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
Message message = Message.obtain(handler, scheduled);
message.obj = this; // Used as token for batch disposal of this worker's runnables.
handler.sendMessageDelayed(message, Math.max(0L, unit.toMillis(delay)));
// Re-check disposed state for removing in case we were racing a call to dispose().
if (disposed) {
handler.removeCallbacks(scheduled);
return Disposables.disposed();
}
return scheduled;
}
@Override
public void dispose() {
disposed = true;
handler.removeCallbacksAndMessages(this /* token */);
}
@Override
public boolean isDisposed() {
return disposed;
}
}
在new ObserveOnObserver< T >(observer, w, delayError, bufferSize)中，我们就选两个方法来看一下就可以了
@Override
public void onSubscribe(Disposable s) {
if (DisposableHelper.validate(this.s, s)) {
this.s = s;
if (s instanceof QueueDisposable) {
@SuppressWarnings("unchecked")
QueueDisposable<T> qd = (QueueDisposable<T>) s;
int m = qd.requestFusion(QueueDisposable.ANY | QueueDisposable.BOUNDARY);
if (m == QueueDisposable.SYNC) {
sourceMode = m;
queue = qd;
done = true;
actual.onSubscribe(this);
schedule();
return;
}
if (m == QueueDisposable.ASYNC) {
sourceMode = m;
queue = qd;
actual.onSubscribe(this);
return;
}
}
queue = new SpscLinkedArrayQueue<T>(bufferSize);
actual.onSubscribe(this);
}
}
@Override
public void onNext(T t) {
if (done) {
return;
}
if (sourceMode != QueueDisposable.ASYNC) {
queue.offer(t);
}
schedule();
}
其中的schedule方法比较重要
void schedule() {
if (getAndIncrement() == 0) {
worker.schedule(this);
}
}
而这个schedule
@NonNull
public Disposable schedule(@NonNull Runnable run) {
return schedule(run, 0L, TimeUnit.NANOSECONDS);
}
就调用了之前的HandlerScheduler里面的schedule方法
@Override
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) {
return Disposables.disposed();
}
run = RxJavaPlugins.onSchedule(run);
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
Message message = Message.obtain(handler, scheduled);
message.obj = this; // Used as token for batch disposal of this worker's runnables.
handler.sendMessageDelayed(message, Math.max(0L, unit.toMillis(delay)));
// Re-check disposed state for removing in case we were racing a call to dispose().
if (disposed) {
handler.removeCallbacks(scheduled);
return Disposables.disposed();
}
return scheduled;
}
其中
run = RxJavaPlugins.onSchedule(run);
ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
Message message = Message.obtain(handler, scheduled);
message.obj = this;
// Used as token for batch disposal of this worker's runnables.
handler.sendMessageDelayed(message, Math.max(0L, unit.toMillis(delay)));
就完成了在MainThread中的观察，因为我们的handler就是刚刚new Handler（）的时候传入了Looper.getMainLooper（）。
至于subscribeon是如何切换线程的。。还没看出来。。明天再说吧。。。
网友评论