RxJava自带线程池监控和管理的探索之旅
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);
}
executor = exec;
}
public ScheduledAction scheduleActual(final Action0 action, long delayTime, TimeUnit unit) {
Action0 decoratedAction = RxJavaHooks.onScheduledAction(action);
ScheduledAction run = new ScheduledAction(decoratedAction);
Future<?> f;
if (delayTime <= 0) {
f = executor.submit(run);
} else {
f = executor.schedule(run, delayTime, unit);
}
run.add(f);
return run;
}
public ScheduledThreadPoolExecutor(int corePoolSize,
ThreadFactory threadFactory) {
super(corePoolSize, Integer.MAX_VALUE,
DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
new DelayedWorkQueue(), threadFactory);
}
复制代码
技术方案分析
本线程管理方案特性
- 几乎0侵入代码，只需要在Application初始化。如果要使用优先级，则需要对改变优先级的任务的代码块传入参数，属于正常侵入。
- 线程创建、执行的流程优化，更适合应用APP利用线程池执行任务
- 线程的执行代码块定位和耗时统计等基本信息收集，以及以此为基础的任务追踪频度，比如某个阶段的线程开启情况和任务耗时，任务执行频度统计。
- 避免pthread_create、Rxjava IO scheduler线程开启过多导致的OOM等问题
- 任务优先级执行，可在性能不足时丢弃执行可抛弃类型的任务
- 提供当前应用总线程数和其他线程名字的打印输出
- 可监控子线程中再次切换线程的情况，避免多余线程开销
本方案提供一些自定义的基础类，也可以选择实现接口，自由替换为自己想要的策略。
方案分析
主要是利用RxJava本身RxJavaPlugins做替换线程池的操作。然后自定义一些scheduler所需要的类，完成相关逻辑。
RxJavaPlugins.getInstance().registerSchedulersHook(new RxJavaSchedulersHookImpl()); 复制代码
管理逻辑类:IOTaskPriorityType、IOMonitorManager、AppSchedulers IOMonitorManager：负责调用基础监控类，并且替换默认RxJava的IO scheduler，作为主要的逻辑管理类，负责初始化整个监控模块，以及配置基本参数。
AppSchedulers作为替换Schedulers的存在，用于应用层显式地使用自定义IO Scheduler，区别于原来的RxJava的Sheduler.IO。
IOTaskPriorityType：定义不同的任务类型，用于区分管理。
大致结构如下图所示。
改造后的任务执行流程也做了相应的改变，有别于上面提到的原生流程。对什么时候创建线程做了调整，优先在限制线程池大小范围内创建新线程执行任务，而不是以前的无限创建或者先阻塞队列等待执行。为了方便对比，把上面的图片也拷贝下来。
使用说明
关于custominterface包：自定义IO scheduler或者scheduler所使用的线程池时，需要关注这个包下面的接口和抽象类 customScheduler包：已经自定义好的scheduler相关以及提供的基础线程池，可以参考这里的实现，去自定义应用自己的线程池管理的scheduler
大部分时候你只需要关心IOMonitorManager这个入口管理类，其它只在需要自定义或者策略改动时才修改。
/*
* description 自定义IO线程监控管理类，配置监视器的基本参数，开启各种debug方法等
* 监控要求：
* 1、自定义的scheduler需要继承AbstractScheduler
* 2、参考或者直接使用ExecutorSchedulerWorker创建任务
* 3、如果是要替换原始RxJava的IO线程池，需要额外实现IThreadPool，创建自己的线程池类
* 4、IThreadPool的实现类里，线程池的构造使用MonitorThreadPoolExecutor类，便于监控
* 5、编写新的scheduler参考自定义的IOScheduler类
* 6、如果需要自定义SchedulerWork，需要实现Runnable, IBaseWork 接口，继承Scheduler.Worker
* 7、默认提供IOThreadPool和LimitCoreThreadPool两种基础线程池，还有自定义的IOScheduler（用来替换原本RxJava的IOScheduler）
* 8、IOTaskPriorityType优先级类型，RxJava observable在subscribeOn时可以选择传入
* 9、AbstractRejectedExecutionHandler可以做一些拒绝任务的策略动作
*
* @see ExecutorSchedulerWorker
* 使用方式：
* 1、所有public方法提供配置参数
* 2、在基础参数配置完后调用此方法startMonitor
* modify by
*/
//必须在应用第一次使用observable之前设置，这里会替换rxjava的默认IO scheduler。
// 如果只调用setReplaceIOScheduler方法，则替换时用基础库里自带的自定义IO scheduler
//LimitCoreThreadPool不是基础库默认的IO scheduler实现,一般都是替换线程池实现，不直接修改自定义的IO scheduler
IOMonitorManager.getInstance().setReplaceIOScheduler(true)
.setIOThreadPool(LimitCoreThreadPool.getInstance()
.build(2, BuildConfig.DEBUG ? 35 : 35, 15, 1000, false));
List<String> targetList = new ArrayList<>(2);
targetList.add("com.xtc");
//配置基本的监视器参数，下面参数可以在IOMonitorManager的startMonitor后修改
IOMonitorManager.getInstance().setCostTimeLimit(0)
//超出这个活跃线程数就输出到日志
.setThreadActiveCountLimit(30)
//打印当前被监控的线程池信息
.setLogThreadPoolInfo(BuildConfig.DEBUG)
//打印其它非RxJava的IO线程信息
.setLogOtherThreadInfo(true)
.setPackageName(getPackageName())
// .setTargetNameList(list)
//监控器轮询时间，每隔这么久打印一些线程信息
.setMonitorIntervalTime(10)
//是否输出更多日志信息，看方法注释
.setLogMoreInfo(BuildConfig.DEBUG)
//监控器是否启用
.setMonitorEnable(BuildConfig.DEBUG)
//一般在调试时才开启，监控子线程重复切换线程
.setLogRepeatChangeThread(false)
//打印所有任务执行情况，适合打桩分析当前时间段有哪些任务触发
.setLogAllTaskRunningInfo(true)
//是否过滤堆栈
.setFilterStack(!BuildConfig.DEBUG);
复制代码
正文到此结束
- 本文标签： db description 数据 CEO 开发 http Action id 并发 测试 grep 调试 注释 操作系统 ask Atom IO zab update find UI 时间 参数 统计 queue 图片 缓存 ssh tar 配置 需求 list core 线程池 executor 拒绝策略 java 安全 代码 value src 任务调度 cache ArrayList logo https cat 多线程 Android build IOS plugin 本质 ip 进程 管理 App ThreadPoolExecutor 文章 ORM bug cpu负载 线程 IDE CTO final ACE Service 模型
- 版权声明： 本文为互联网转载文章，出处已在文章中说明(部分除外)。如果侵权，请联系本站长删除，谢谢。
- 本文海报： 生成海报一 生成海报二
热门推荐
相关文章
近期评论
-
666
-
文章和留言都翻到11页了 没有OOM
-
-
朋友，翻页到11页，及以后，会出现OOM，无法访问
-
-
-
-
版本号是多少，你可以下载哪个代码仓库，jdk选1.8 直接跑就行
-
-
Loading...
![[HBLOG]公众号](https://www.liuhaihua.cn/img/qrcode_gzh.jpg)
