SingleScheduler是RxJava2新增的Scheduler。SingleScheduler中有一个属性叫作executor,它是使用AtomicReference包装的ScheduledExecutorService。
补充:AtomicReference类的作用:AtomicReference则对应普通的对象引用,即保证你在修改对象引用时的线程安全性;对” 对象
”进行原子操作
final AtomicReference<ScheduledExecutorService> executor = new AtomicReference<ScheduledExecutorService>(); 复制代码
在SingleScheduler构造函数中,Executor会调用lazySet().
/** * @param threadFactory thread factory to use for creating worker threads. Note that this takes precedence over any * system properties for configuring new thread creation. Cannot be null. */ public SingleScheduler(ThreadFactory threadFactory) { this.threadFactory = threadFactory; executor.lazySet(createExecutor(threadFactory)); } 复制代码
************ 分割线 ************
其中**lazySet()**是 AtomicReference
中的方法,用于修改引用对象:
// AtomicRefence类 /** * Sets to the given value. * * @param newValue the new value */ public final void set(V newValue) { value = newValue; } /** * Eventually sets to the given value. * * @param newValue the new value * @since 1.6 */ public final void lazySet(V newValue) { U.putOrderedObject(this, VALUE, newValue); } 复制代码
AtomicReferences中set()和lazySet()区别: set()
会 立刻修改旧值
,别的线程可以立刻看到更新后的值;而 lazySet()
不会立刻(但是 最终会
)修改旧值,别的线程看到新值的时间会延迟一些。
************ 分割线 ************
它的createExecutor()用于创建工作线程,可以看到通过SchedulerPoolFactory来创建ScheduledExecutorService。
// SingleScheduler类 static ScheduledExecutorService createExecutor(ThreadFactory threadFactory) { return SchedulerPoolFactory.create(threadFactory); } 复制代码
通过SchedulerPoolFactory类的create(ThreadFactory factory)来创建单线程的线程
// SchedulerPoolFactory类
/**
* Creates a ScheduledExecutorService with the given factory.
* @param factory the thread factory
* @return the ScheduledExecutorService
*/
public static ScheduledExecutorService create(ThreadFactory factory) {
// 创建单线程
final ScheduledExecutorService exec = Executors.newScheduledThreadPool(1, factory);
if (exec instanceof ScheduledThreadPoolExecutor) {
ScheduledThreadPoolExecutor e = (ScheduledThreadPoolExecutor) exec;
POOLS.put(e, exec);
}
return exec;
}
复制代码
在SingleScheduler中,每次使用ScheduledExecutorService时,其实是使用executor.get()。所以说,single拥有一个线程单例。
SingleScheduler会创建一个ScheduledWorker,ScheduledWorker使用JDK的ScheduledExecutorService作为executor。
下面是ScheduledWorker的schedule()方法,使用ScheduledExecutorService的submit()或schedule()来执行runnable。
@NonNull @Override public Disposable schedule(@NonNull Runnable run, long delay, @NonNull TimeUnit unit) { if (disposed) { return EmptyDisposable.INSTANCE; } Runnable decoratedRun = RxJavaPlugins.onSchedule(run); ScheduledRunnable sr = new ScheduledRunnable(decoratedRun, tasks); tasks.add(sr); try { Future<?> f; if (delay <= 0L) { /** * 立即执行则执行submit()方法 * Submits a value-returning task for execution and returns a * Future representing the pending results of the task. The * Future's {@code get} method will return the task's result upon * successful completion. */ f = executor.submit((Callable<Object>)sr); } else { /** * 需延迟执行,则执行schedule()方法 * Creates and executes a ScheduledFuture that becomes enabled after the * given delay. */ f = executor.schedule((Callable<Object>)sr, delay, unit); } sr.setFuture(f); } catch (RejectedExecutionException ex) { dispose(); RxJavaPlugins.onError(ex); return EmptyDisposable.INSTANCE; } return sr; } 复制代码
2.ComputationScheduler
ComputationScheduler使用FixedSchedulerPool作为线程池,并且FixedSchedulerPool被AtomicReference包装了一下。
从ComputationScheduler的源码中可以看出, MAX_THREADS
是 CPU的数目
。 FixedSchedulerPool
可以理解为拥有 固定数量的线程池
(有点类似线程池中的FixedThreadPool),数量为MAX_THREADS。
static { MAX_THREADS = cap(Runtime.getRuntime().availableProcessors(), Integer.getInteger(KEY_MAX_THREADS, 0)); ... } static int cap(int cpuCount, int paramThreads) { return paramThreads <= 0 || paramThreads > cpuCount ? cpuCount : paramThreads; } 复制代码
ComputationScheduler类会创建一个EventLoopWorker。
@NonNull @Override public Worker createWorker() { return new EventLoopWorker(pool.get().getEventLoop()); } 复制代码
其中getEventLoop()是FixedSchedulerPool中的方法,返回了FixedSchedulerPool中的一个PoolWorker。
注: FixedSchedulerPool
和 EventLoopWorker
都为 ComputationScheduler
的内部类
// EventLoopWorker类中的方法
public PoolWorker getEventLoop() {
int c = cores;
if (c == 0) {
return SHUTDOWN_WORKER;
}
// simple round robin, improvements to come
return eventLoops[(int)(n++ % c)];
}
复制代码
PoolWorker继承自NewThreadWorker,也是线程数为1的ScheduledExecutorService。
3.IoScheduler
IoScheduler使用CachedWorkerPool作为线程池,并且CacheWorkerPool也被AtomicReference包装了一下。
CachedWorkerPool是基于RxThreadFactory这个ThreadFactory来创建的。
static {
...
WORKER_THREAD_FACTORY = new RxThreadFactory(WORKER_THREAD_NAME_PREFIX, priority);
...
NONE = new CachedWorkerPool(0, null, WORKER_THREAD_FACTORY);
...
}
复制代码
在RxThreadFactory中,由prefix和incrementAndGet()来创建新线程的名称
后续补充,先写到这儿
原文
https://juejin.im/post/5eee285ef265da02d812d843
本站部分文章源于互联网,本着传播知识、有益学习和研究的目的进行的转载,为网友免费提供。如有著作权人或出版方提出异议,本站将立即删除。如果您对文章转载有任何疑问请告之我们,以便我们及时纠正。PS:推荐一个微信公众号: askHarries 或者qq群:474807195,里面会分享一些资深架构师录制的视频录像:有Spring,MyBatis,Netty源码分析,高并发、高性能、分布式、微服务架构的原理,JVM性能优化这些成为架构师必备的知识体系。还能领取免费的学习资源,目前受益良多

转载请注明原文出处:Harries Blog™ » RxJava之Scheduler (二)