这篇文章来看看rxjava指定线程的时候，那些线程是怎么创建出来的，先来看Schedulers.io(),看看这个io线程的创建：
public final class Schedulers {
...
@NonNull
static final Scheduler IO;
static {
//RxJavaPlugins类是rxjava里很强大的工具类，之前我们多次看到它的身影
//这里又用到了它来创建Scheduler对象
IO = RxJavaPlugins.initIoScheduler(new IOTask());
}
public static Scheduler io() {
//RxJavaPlugins的onIoScheduler()方法，这种方法看着又是一个包装，没有包装则直接返回入参IO自身
return RxJavaPlugins.onIoScheduler(IO);
}
//静态内部类IOTask
static final class IOTask implements Callable<Scheduler> {
@Override
public Scheduler call() throws Exception {
return IoHolder.DEFAULT;
}
}
//静态内部类IoHolder
static final class IoHolder {
//静态变量DEFAULT ，最终是在这里创建的IoScheduler
static final Scheduler DEFAULT = new IoScheduler();
}
...
}
再来看看RxJavaPlugins里的initIoScheduler()方法：
public static Scheduler initIoScheduler(@NonNull Callable<Scheduler> defaultScheduler) {
ObjectHelper.requireNonNull(defaultScheduler, "Scheduler Callable can't be null");
Function<? super Callable<Scheduler>, ? extends Scheduler> f = onInitIoHandler;
if (f == null) {
//onInitIoHandler不另外设置则为空，callRequireNonNull方法返回的就是IoScheduler
return callRequireNonNull(defaultScheduler);
}
return applyRequireNonNull(f, defaultScheduler);
}
static Scheduler callRequireNonNull(@NonNull Callable<Scheduler> s) {
try {
//s.call()不为空则返回s.call()，也就是IoScheduler
return ObjectHelper.requireNonNull(s.call(), "Scheduler Callable result can't be null");
} catch (Throwable ex) {
throw ExceptionHelper.wrapOrThrow(ex);
}
}
绕来绕去，最终Schedulers.io()拿到的Scheduler就是IoScheduler，由上篇文章可知subscribeOn()方法后最终会这样用到IoScheduler（里面的scheduleDirect()方法就是开了线程在执行Runnable）：
//这个scheduler就是IoScheduler,SubscribeTask就是个ruannable，run()方法运行了Observable的subscribe()方法
parent.setDisposable(scheduler.scheduleDirect(new SubscribeTask(parent)));
先看看scheduleDirect()方法，这个方法是父类Scheduler里的：
public abstract class Scheduler {
...
@NonNull
public Disposable scheduleDirect(@NonNull Runnable run) {
return scheduleDirect(run, 0L, TimeUnit.NANOSECONDS);
}
@NonNull
public Disposable scheduleDirect(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) {
//createWorker()是抽象方法，需要看子类的实现
final Worker w = createWorker();
//返回装饰过的runable，其实就是返回了自身run
final Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
//继续封装成DisposeTask ，decoratedRun的装饰类
DisposeTask task = new DisposeTask(decoratedRun, w);
//执行了task方法
w.schedule(task, delay, unit);
return task;
}
...
}
看看IoScheduler的createWorker()方法：
public final class IoScheduler extends Scheduler {
...
@Override
public Worker createWorker() {
//新建了一个EventLoopWorker对象
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();
this.threadWorker = pool.get();
}
@Override
public void dispose() {
if (once.compareAndSet(false, true)) {
tasks.dispose();
// releasing the pool should be the last action
pool.release(threadWorker);
}
}
@Override
public boolean isDisposed() {
return once.get();
}
@NonNull
@Override
//最终线程是在这里执行的
public Disposable schedule(@NonNull Runnable action, long delayTime, @NonNull TimeUnit unit) {
//解除订阅了，返回EmptyDisposable
if (tasks.isDisposed()) {
// don't schedule, we are unsubscribed
return EmptyDisposable.INSTANCE;
}
//最后靠的是ThreadWorker(线程工作者) 的scheduleActual()(实际安排执行的方法)
return threadWorker.scheduleActual(action, delayTime, unit, tasks);
}
}
//静态内部类ThreadWorker，继承自NewThreadWorker，NewThreadWorker很重要
static final class ThreadWorker extends NewThreadWorker {
//封装了一个截至时间，主要的方法还是在父类
private long expirationTime;
ThreadWorker(ThreadFactory threadFactory) {
super(threadFactory);
this.expirationTime = 0L;
}
public long getExpirationTime() {
return expirationTime;
}
public void setExpirationTime(long expirationTime) {
this.expirationTime = expirationTime;
}
}
...
}
看看父类NewThreadWorker ：
public class NewThreadWorker extends Scheduler.Worker implements Disposable {
...
private final ScheduledExecutorService executor;
public NewThreadWorker(ThreadFactory threadFactory) {
executor = SchedulerPoolFactory.create(threadFactory);
}
public ScheduledRunnable scheduleActual(final Runnable run, long delayTime, @NonNull TimeUnit unit, @Nullable DisposableContainer parent) {
//返回自身run
Runnable decoratedRun = RxJavaPlugins.onSchedule(run);
//装饰decoratedRun，ScheduledRunnable 便于管理生命周期，这里不深究，本质还是runnable
ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, parent);
if (parent != null) {
if (!parent.add(sr)) {
return sr;
}
}
Future<?> f;
try {
if (delayTime <= 0) {
//熟悉的身影，executor是个线程池，最终在这里开辟了异步线程来执行subscribe()方法，
//从而达到异步线程的目的，submit()立马执行
f = executor.submit((Callable<Object>)sr);
} else {
//用schedule()来执行延迟线程
f = executor.schedule((Callable<Object>)sr, delayTime, unit);
}
sr.setFuture(f);
} catch (RejectedExecutionException ex) {
if (parent != null) {
parent.remove(sr);
}
RxJavaPlugins.onError(ex);
}
return sr;
}
...
}
上面executor的创建：
public static ScheduledExecutorService create(ThreadFactory factory) {
final ScheduledExecutorService exec = Executors.newScheduledThreadPool(1, factory);//核心线程为1的线程池
if (PURGE_ENABLED && exec instanceof ScheduledThreadPoolExecutor) {
ScheduledThreadPoolExecutor e = (ScheduledThreadPoolExecutor) exec;
POOLS.put(e, exec);
}
return exec;
}
io线程开启总结：
IoScheduler更多的还是做一些协调作用，真正用到的是 NewThreadWorker 类的线程池来开辟线程执行异步操作。
下篇分析AndroidSchdulers.mainThread()，看看它是怎么切换到android主线程的。
网友评论