RxJava自带线程池监控和管理的探索之旅_嘘_不要惊到我的bug的专栏文章_微信文章

背景

RxJava很方便的切换主子线程、指定任务运行的线程，在这个便利之后还隐藏着很多问题。比如IO scheduler是一个无上限线程池，如果短时间并发量过大，在手机端可能出现OOM或者pthread_create错误。另外，在实际业务中我们需要对执行的业务进行优先级区分，以便优先级高的任务先执行，想实现这个需求必然需要对RxJava默认的scheduler进行改造。本文将从RxJava IO scheduler分析、介绍线程池相关知识、如何对IO scheduler进行改造等方向进行介绍，并且对应用旧代码做到无侵入式的替换。

线程池相关知识回顾

1. 线程以及线程池的含义

在介绍主体内容之前，我们先回顾下线程池的相关知识，这样能更好的理解本文章内容。从字面意思上来说，线程池肯定是一个装着线程的"池"，小则是鱼塘，装的少，但是家里没矿只能承包这么大的鱼塘；当然如果是大佬，说不定这一片海都是他的。线程池肯定不是简单的承载容纳线程的池子，既然作为类似仓库的属性，必然有管理之意。

线程是作为一个任务的执行承载者，接收来自程序的诸多执行请求，其中子线程是合理利用cpu性能避免阻塞主线程的存在。好东西都是不可贪多，线程如果不加以管理，肯定会被程序各处的代码随意创建，这样会浪费或者影响cpu某个时刻的性能，甚至导致当前进程出现异常。

阻塞队列，这是线程池待执行任务的容器，负责管理要执行的任务。阻塞二字说明它的入队和出队是可控的，所谓阻塞，在某些情况下会挂起线程（即阻塞），一旦条件满足，被挂起的线程又会自动被唤醒。按照队列的数据结构进行出入某一个任务时会阻塞，这样在多线程环境下才更安全的生产任务和消费任务。

比较通俗的一个例子有仓库提取货物，如果仓库里只有有限的几个小车运输货物，此时有很多运输员来提货，肯定要遇到争夺小车、等待小车。小车类似于cpu的核数，也可以理解为线程池允许创建的总数。总而言之，当前仓库（线程池）只有有限的几辆小车同时工作，每个运输员（程序代码块）想要获取货物（执行代码）就必须要争夺小车资源。当有空闲小车时，会按照一定规则分配给某个经销商，可能是队列的简单先后入队等待顺序，也可能是优先级（毕竟氪金无敌VIP）。当然有可能是大佬，家里矿多，说小车不够，我去买，这样会形成一个无限制上限线程池模型。不过这样做有一种风险就是仓库体积无法随意扩容（整个程序所承载的机器性能有限），买了太多小车放不下，然后整个仓库就瘫痪了，这时候可能OOM或pthread_create错误。小车每次使用完后，都会继续被分配到下一个任务，当然如果经销商的事情都处理完了，可能就都闲置了，有可能晚上没活，仓库就把小车都封存起来，整理回收到固定地方（超时闲置后回收非核心线程），有可能留下几辆预先说好的小车以便晚上有紧急货物时处理（核心线程常驻）。如下图所示，大致概念如下。绿色的取货车可能是因为取货车不够，临时采购或者借调的，类似于线程池临时开启新线程。红色的取货车区域是核心线程，有限的。

一般核心线程数根据CPU数量来确定，线程池数大于CPU数量，看似是并发执行任务，其实是操作系统帮我们在按照一定时间片进行调度来执行任务，达到一种同时执行的效果，所以大量线程同时执行对CPU负载性能要求，会让机器达到50%甚至100%高负荷运作，此时整机机器发生出错的概率增大。所以同时执行很多任务导致频繁切换线程本身也是一种额外的开销，不建议如此操作，尤其是部分任务是低优先级且不重要、可延迟的。

单个线程大概1MB左右开销，在Java内部开发版的JDK中，加入fiber这种新的任务调度模型，开销只有200KB左右，通过100W级任务调度测试，据介绍性能比thread优异很多，release版预测还会提高，特性类似于协程。不过我们Android不太可能使用付费JDK。

通过上面的描述我们简单的了解线程池与线程之间的关系，线程和调用者之间的联系，并且线程池运作时是有自己的规则设定，调用者和每个被管理的线程必须遵守规则。

2. 线程池的种类

java通过Executors提供了四种常用线程池，这四种本质上也是封装了自定义线程池的基本参数，简化了创建流程，提供特定的功能。

先介绍下自定义线程池的基本参数和含义，这样更好理解下面的java封装好的线程池。

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler)
复制代码

一个任务被提交时，它的大致流程都是相似的。下面是流程图（下图取自文章https://juejin.im/post/5b052dd7f265da0ba567e7f1）

CachedThreadPool 缓存线程池这是一种线程数无上限的线程池，可以在有空闲线程时复用，无空闲时新建线程。默认情况下，60s回收空闲线程，并且阻塞队列为SynchronousQueue（这是一种无容量阻塞队列，当拿到任务入队时就判断是否有线程可以执行，如果有就立刻出队执行，否则就阻塞等待）

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}
复制代码

FixedThreadPool 固定大小线程池线程池的线程总数是有上限的，当初始化线程池时可以设置它的容量，如果待执行任务超出总数就需要在队列中等待了。

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}
复制代码

SchedulerThreadPool 定时周期任务线程池它很多参数都和固定大小线程池一样，除了阻塞队列选用了DelayQueue，这是一种按照延时长短排序的队列。

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE,
          DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
          new DelayedWorkQueue());
}
复制代码

SingleThreadExecutor 单线程池只有一个线程，所以队列里的任务按照队列的出队规则逐个执行，队列采用的是一个链表结构的队列。

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}
复制代码

3.自定义线程池相关类介绍

一般都是继承ThreadPoolExecutor类（不继承也可以，但是继承是为了做更多的方法执行监控），然后根据需要设置下面7种参数。

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }
复制代码

比较重要的就是阻塞队列workQueue和拒绝策略handler的选择，这个会在后续RxJava的IO scheduler监控方案里再次介绍。java提供的默认队列的种类有无限大小和有限，带优先级、带延时等，根据需要可以选择不同类型的队列。ThreadFactory 是构造新的thread的工厂，这里自定义一个可以进行新建线程的监控。拒绝策略handler有提供四种默认的策略，也可以自己实现接口RejectedExecutionHandler自己做特殊策略，比如移交任务到另外一个执行者，或者判断下这个任务的重要性，然后再抛弃。

//ThreadPoolExecutor.CallerRunsPolicy：在调用者所在线程执行该任务

        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                r.run();
            }
        }
//ThreadPoolExecutor.AbortPolicy:放弃执行任务，抛出RejectedExecutionException异常。

        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            throw new RejectedExecutionException("Task " + r.toString() +
                                                 " rejected from " +
                                                 e.toString());
        }

//ThreadPoolExecutor.DiscardPolicy：放弃执行任务，不抛出异常。

        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
        }

//ThreadPoolExecutor.DiscardOldestPolicy：poll出队一个最早任务，然后尝试执行它

 public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            if (!e.isShutdown()) {
                e.getQueue().poll();
                e.execute(r);
            }
        }

复制代码

IO scheduler的弊端

RxJavaSchedulersHook类里会生成IO Scheduler，默认调用CachedThreadScheduler。里面的CachedWorkerPool维护了一个线程管理的队列expiringWorkerQueue，默认是每隔60s就去通过evictor清除已经过期的线程，线程池没有上限。因此如果短时间内有大量任务要执行，会导致不停地创建新线程，所以存在出现pthread_create、OOM、耗费大量系统资源造成卡顿等问题。

  
public final class CachedThreadScheduler extends Scheduler implements SchedulerLifecycle {
    private static final long KEEP_ALIVE_TIME;
    private static final TimeUnit KEEP_ALIVE_UNIT = TimeUnit.SECONDS;

    static final ThreadWorker SHUTDOWN_THREADWORKER;

    static final CachedWorkerPool NONE;

    final ThreadFactory threadFactory;

    final AtomicReference<CachedWorkerPool> pool;

    static {
        SHUTDOWN_THREADWORKER = new ThreadWorker(RxThreadFactory.NONE);
        SHUTDOWN_THREADWORKER.unsubscribe();

        NONE = new CachedWorkerPool(null, 0, null);
        NONE.shutdown();

        KEEP_ALIVE_TIME = Integer.getInteger("rx.io-scheduler.keepalive", 60);
    }

    static final class CachedWorkerPool {
        private final ThreadFactory threadFactory;
        private final long keepAliveTime;
        private final ConcurrentLinkedQueue<ThreadWorker> expiringWorkerQueue;
        private final CompositeSubscription allWorkers;
        private final ScheduledExecutorService evictorService;
        private final Future<?> evictorTask;

        CachedWorkerPool(final ThreadFactory threadFactory, long keepAliveTime, TimeUnit unit) {
            this.threadFactory = threadFactory;
            this.keepAliveTime = unit != null ? unit.toNanos(keepAliveTime) : 0L;
            this.expiringWorkerQueue = new ConcurrentLinkedQueue<ThreadWorker>();
            this.allWorkers = new CompositeSubscription();

            ScheduledExecutorService evictor = null;
            Future<?> task = null;
            if (unit != null) {
                evictor = Executors.newScheduledThreadPool(1, new ThreadFactory() {
                    @Override public Thread newThread(Runnable r) {
                        Thread thread = threadFactory.newThread(r);
                        thread.setName(thread.getName() + " (Evictor)");
                        return thread;
                    }
                });
                NewThreadWorker.tryEnableCancelPolicy(evictor);
                task = evictor.scheduleWithFixedDelay(
                        new Runnable() {
                            @Override
                            public void run() {
                                evictExpiredWorkers();
                            }
                        }, this.keepAliveTime, this.keepAliveTime, TimeUnit.NANOSECONDS
                );
            }
            evictorService = evictor;
            evictorTask = task;
        }

        ThreadWorker get() {
            if (allWorkers.isUnsubscribed()) {
                return SHUTDOWN_THREADWORKER;
            }
            while (!expiringWorkerQueue.isEmpty()) {
                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;
        }

        void release(ThreadWorker threadWorker) {
            // Refresh expire time before putting worker back in pool
            threadWorker.setExpirationTime(now() + keepAliveTime);

            expiringWorkerQueue.offer(threadWorker);
        }
        //每60s执行一次清除队列中的已过时线程
        void evictExpiredWorkers() {
            if (!expiringWorkerQueue.isEmpty()) {
                long currentTimestamp = now();

                for (ThreadWorker threadWorker : expiringWorkerQueue) {
                    if (threadWorker.getExpirationTime() <= currentTimestamp) {
                        if (expiringWorkerQueue.remove(threadWorker)) {
                            allWorkers.remove(threadWorker);
                        }
                    } else {
                        // Queue is ordered with the worker that will expire first in the beginning, so when we
                        // find a non-expired worker we can stop evicting.
                        break;
                    }
                }
            }
        }

        long now() {
            return System.nanoTime();
        }

        void shutdown() {
            try {
                if (evictorTask != null) {
                    evictorTask.cancel(true);
                }
                if (evictorService != null) {
                    evictorService.shutdownNow();
                }
            } finally {
                allWorkers.unsubscribe();
            }
        }
    }

    public CachedThreadScheduler(ThreadFactory threadFactory) {
        this.threadFactory = threadFactory;
        this.pool = new AtomicReference<CachedWorkerPool>(NONE);
        start();
    }

    @Override
    public void start() {
        CachedWorkerPool update =
            new CachedWorkerPool(threadFactory, KEEP_ALIVE_TIME, KEEP_ALIVE_UNIT);
        if (!pool.compareAndSet(NONE, update)) {
            update.shutdown();
        }
    }
    @Override
    public void shutdown() {
        for (;;) {
            CachedWorkerPool curr = pool.get();
            if (curr == NONE) {
                return;
            }
            if (pool.compareAndSet(curr, NONE)) {
                curr.shutdown();
                return;
            }
        }
    }

    @Override
    public Worker createWorker() {
        return new EventLoopWorker(pool.get());
    }

    static final class EventLoopWorker extends Scheduler.Worker implements Action0 {
        private final CompositeSubscription innerSubscription = new CompositeSubscription();
        private final CachedWorkerPool pool;
        private final ThreadWorker threadWorker;
        final AtomicBoolean once;

        EventLoopWorker(CachedWorkerPool pool) {
            this.pool = pool;
            this.once = new AtomicBoolean();
            this.threadWorker = pool.get();
        }

        @Override
        public void unsubscribe() {
            if (once.compareAndSet(false, true)) {
                // unsubscribe should be idempotent, so only do this once

                // Release the worker _after_ the previous action (if any) has completed
                threadWorker.schedule(this);
            }
            innerSubscription.unsubscribe();
        }

        @Override
        public void call() {
            pool.release(threadWorker);
        }

        @Override
        public boolean isUnsubscribed() {
            return innerSubscription.isUnsubscribed();
        }

        @Override
        public Subscription schedule(Action0 action) {
            return schedule(action, 0, null);
        }

        @Override
        public Subscription schedule(final Action0 action, long delayTime, TimeUnit unit) {
            if (innerSubscription.isUnsubscribed()) {
                // don't schedule, we are unsubscribed
                return Subscriptions.unsubscribed();
            }

            ScheduledAction s = threadWorker.scheduleActual(new Action0() {
                @Override
                public void call() {
                    if (isUnsubscribed()) {
                        return;
                    }
                    action.call();
                }
            }, delayTime, unit);
            innerSubscription.add(s);
            s.addParent(innerSubscription);
            return s;
        }
    }

    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里最终设置executor的初始化构造，这里可以看到是一个定时周期任务线程池，核心线程为1.

   /* package */
    public NewThreadWorker(ThreadFactory threadFactory) {
        ScheduledExecutorService exec = Executors.newScheduledThreadPool(1, threadFactory);
        // Java 7+: cancelled future tasks can be removed from the executor thus avoiding memory leak
        boolean cancelSupported = tryEnableCancelPolicy(exec);
        if (!cancelSupported && exec instanceof ScheduledThreadPoolExecutor) {
            registerExecutor((ScheduledThreadPoolExecutor)exec

RxJava自带线程池监控和管理的探索之旅

正文