RxJava之Scheduler (二)

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 (二)

赞 (0)
分享到:更多 ()

评论 0

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址