java线程池实践
- 2021 年 5 月 9 日
- 笔记
- JAVA, threadpool
线程池大家都很熟悉，无论是平时的业务开发还是框架中间件都会用到，大部分都是基于JDK线程池ThreadPoolExecutor做的封装，
都会牵涉到这几个核心参数的设置：核心线程数，等待(任务)队列，最大线程数，拒绝策略等。
但如果线程池设置不当就会引起一系列问题, 下面就说下我最近碰到的问题。
案件还原
比如你有一个项目中有个接口部分功能使用了线程池，这个功能会去调用多个第三方接口，都有一定的耗时，为了不影响主流程的性能，不增加整体响应时间,所以放在线程池里和主线程并行执行，等线程池里的任务执行完通过future.get的方式获取线程池里的线程执行结果，然后合并到主流程的结果里返回，大致流程如下：
线程池参数为：
- coresize:50
- max:200
- queuesize:1
- keepalivetime:60s
- 拒绝策略为reject
假设每次请求提交5个task到线程池，平均每个task是耗时50ms
没过一会就收到了线程池满了走了拒绝策略的报错
结合你对线程池的了解，先思考下为什么
线程池的工作流程如下：
根据这个我们来列一个时间线
1. 项目刚启动 第1次请求(每次5个task提交到线程池)，创建5个核心线程
2. 第2次请求 继续创建5个（共10个核心线程了）
3. 直到第10次 核心线程数会达满50个
4. 核心线程处理完之后核心线程会干嘛呢
根据 jdk1.8的线程池的源码：
线程池的线程处理处理了交给它的task之后，它会去getTask()
源码如下：
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
//注意这段
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
请注意上面代码中的bool类型的timed的赋值逻辑，
由于allowCoreThreadTimeOut默认为false，也就是说：
只要创建的线程数量超过了核心线程数，那么干完手上活后的线程（不管是核心线程，还是超过队列后新开的线程）就会走进
//线程状态为 timedwaiting
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)
由于我们上面步骤里面还没有超过coresize所以会走进
//线程状态为 waiting
workQueue.take()
所以答案是：上面步骤干活的核心线程处理完之后核心线程会进入waiting状态，
只要队列一有活就会被唤醒去干活。
5. 到第11次的时候
好家伙，到这步骤的时候 ，核心线程数已满，那么就往队列里面塞，但是设置的queuesize=1，
每次有5个task，那就是说往队列里面塞1个，剩下4个(别较真我懂你意思)要创建新的max线程了。
结果：
- 核心线程数：50
- 队列：1
- max线程：4个
因为50个核心线程在waiting中，所以队列只要一add，就会立马被消费，假设消费的这个核心线程名字是小A。
这里要细品一下：
这里已经总线程数大于核心线程数了，那么getTask()里面
// timed=true
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
那幺小A干完活就会走进
//线程状态为 timedwaiting
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)
此处核心线程小A就会变成timedwaiting的状态（keepalive设置的是60s）
6. 到第12次的时候
继续往队列塞1个，创建4个max线程，max线程已经有8个了
这里 又会有一个新的核心线程小B ，会变成timedwaiting状态了
max线程们干完手上的活后，也会去调用getTask() 也会进入timedwaiting状态
因为queuesize=1，狼多肉少
7. 继续下去，那么最终会变成
max满了，线程们都在timedwaiting（keepalive设置的是60s）
新的提交就会走拒绝策略了
问题总结
其实核心与非核心对于线程池来说都是一样的，只要一旦线程数超过了核心线程数，那么线程就会走进timewaiting
把queuesize调大就好了？
这里又有一个新的注意点：
上面举例的是I/O密集型业务，queuesize不是越大越好的，
因为：
线程池新创建的线程会优先处理新请求进来的任务，而不是去处理队列里的任务，队列里的任务只能等核心线程数忙完了才能被执行，这样可能造成队列里的任务长时间等待，导致队列积压，尤其是I/O密集场景
慎用CallRunnerPolicy这个拒绝策略
一定得理解这个策略会带来什么影响，
先看下这个拒绝策略的源码
如果你提交线程池的任务即时失败也没有关系的话，用这个拒绝策略是致命的，
因为一旦超过线程池的负载后开始吞噬tomcat线程。
用future.get的方式慎用DiscardPolicy这个拒绝策略
如果需要得到线程池里的线程执行结果，使用future的方式，拒绝策略不建议使用DiscardPolicy，这种丢弃策略虽然不执行子线程的任务，
但是还是会返回future对象(其实在这种情况下我们已经不需要线程池返回的结果了)，然后后续代码即使判断了future!=null也没用，
这样的话还是会走到future.get()方法，如果get方法没有设置超时时间会导致一直阻塞下去
类似下面的伪代码：
// 如果线程池已满，新的请求会被直接执行拒绝策略,此时如果拒绝策略设置的是DiscardPolicy丢弃任务,
// 则还是会返回future对象, 这样的话后续流程还是可能会走到get获取结果的逻辑
Future<String> future = executor.submit(() -> {
// 业务逻辑,比如调用第三方接口等操作
return result;
});
// 主流程调用逻辑
if(future != null) // 如果拒绝策略是DiscardPolicy还是会走到下面代码
future.get(超时时间); // 调用方阻塞等待结果返回，直到超时
推荐解决方案
1. 用动态线程池，可以动态修改coresize，maxsize，queuesize，keepalivetime
- 对线程池的核心指标进行埋点监控，可以通过继承 ThreadPoolExecutor 然后Override掉beforeExecute，afterExecute，shutdown，shutdownNow方法，进行埋点记录到es
- 可以埋点的数据有：
包括线程池运行状态、核心线程数、最大线程数、任务等待数、已完成任务数、线程池异常关闭等信息
| 名称 | 含义 |
|---|---|
| core_pool_size | 定义的核心线程总数 |
| max_pool_size | 定义的maxpoolsize |
| keep_alive_time | 定义的keepalivetime |
| current_pool_size | 当前线程池总线程数 |
| queue_wait_size | 当前队列中等待处理的个数 |
| active_count | 当前run状态的线程数 |
| completed_count | 当前线程池中的每个线程处理的task数的叠加值 |
| task_count | 等于completed_count加上queue_wait_size |
| shutdown | 当前线程池的状态是否关闭 |
| useRate | 当前线程池利用率：((active_count * 1.0 / max_pool_size) * 100) |
基于以上数据，我们可以实时监控和排查定位问题
参考代码：
/**
* 自定义线程池<p>
* 1.监控线程池状态及异常关闭等情况<p>
* 2.监控线程池运行时的各项指标, 比如:任务执行时间、任务等待数、已完成任务数、任务异常信息、核心线程数、最大线程数等<p>
* author: maoyingxu
*/
public class ThreadPoolExt extends ThreadPoolExecutor{
private TimeUnit timeUnit;
public ThreadPoolExt(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
this.timeUnit = unit;
}
@Override
protected void beforeExecute(Thread t, Runnable r) {
monitor("ThreadPool monitor data:"); // 监控线程池运行时的各项指标
}
@Override
protected void afterExecute(Runnable r, Throwable ex) {
// 记录线程池执行任务的时间
ELKLogUtils.addAppendedValue(StoredLogTag.RUNNING_DETAIL, MessageFormat.format("ThreadPool task executeTime:{0}", executeTime));
if (ex != null) { // 监控线程池中的线程执行是否异常
LogUtils.warn("unknown exception caught in ThreadPool afterExecute:", ex);
}
}
@Override
public void shutdown() {
monitor("ThreadPool will be shutdown:"); // 线程池将要关闭事件,此方法会等待线程池中正在执行的任务和队列中等待的任务执行完毕再关闭
super.shutdown();
}
@Override
public List<Runnable> shutdownNow() {
monitor("ThreadPool going to immediately be shutdown:"); // 线程池立即关闭事件,此方法会立即关闭线程池,但是会返回队列中等待的任务
// 记录被丢弃的任务, 目前只记录日志, 后续可根据业务场景做进一步处理
List<Runnable> dropTasks = null;
try {
dropTasks = super.shutdownNow();
ELKLogUtils.addAppendedValue(StoredLogTag.RUNNING_DETAIL, MessageFormat.format("{0}ThreadPool discard task count:{1}{2}",
System.lineSeparator(), dropTasks!=null ? dropTasks.size() : 0, System.lineSeparator()));
} catch (Exception e) {
LogUtils.addClogException("ThreadPool shutdownNow error", e);
}
return dropTasks;
}
/**
* 监控线程池运行时的各项指标, 比如:任务等待数、任务异常信息、已完成任务数、核心线程数、最大线程数等
* @param title
*/
private void monitor(String title){
try {
// 线程池监控信息记录, 这里需要注意写ES的时机,尤其是多个子线程的日志合并到主流程的记录方式
String threadPoolMonitor = MessageFormat.format(
"{0}{1}core pool size:{2}, current pool size:{3}, queue wait size:{4}, active count:{5}, completed task count:{6}, " +
"task count:{7}, largest pool size:{8}, max pool size:{9}, keep alive time:{10}, is shutdown:{11}, is terminated:{12}, " +
"thread name:{13}{14}",
System.lineSeparator(), title, this.getCorePoolSize(), this.getPoolSize(),
this.getQueue().size(), this.getActiveCount(), this.getCompletedTaskCount(), this.getTaskCount(), this.getLargestPoolSize(),
this.getMaximumPoolSize(), this.getKeepAliveTime(timeUnit != null ? timeUnit : TimeUnit.SECONDS), this.isShutdown(),
this.isTerminated(), Thread.currentThread().getName(), System.lineSeparator());
ELKLogUtils.addAppendedValue(StoredLogTag.RUNNING_DETAIL, threadPoolMonitor);
LogUtils.info(title, threadPoolMonitor);
ELKLogUtils.addFieldValue(APPIndexedLogTag.THREAD_POOL_USE_RATE, useRate); // ES埋点线程池使用率, useRate = (getActiveCount()/getMaximumPoolSize())*100
Cat.logEvent(key, String.valueOf(useRate)); // 报警设置
} catch (Exception e) {
LogUtils.addClogException("ThreadPool monitor error", e);
}
}
}
2. 重写线程池拒绝策略, 拒绝策略主要参考了 Dubbo的线程池拒绝策略
public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy {
// 省略部分代码
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
String msg = String.format("Thread pool is EXHAUSTED!" +
" Thread Name: %s, Pool Size: %d (active: %d, core: %d, max: %d, largest: %d), Task: %d (completed: "
+ "%d)," +
" Executor status:(isShutdown:%s, isTerminated:%s, isTerminating:%s), in %s://%s:%d!",
threadName, e.getPoolSize(), e.getActiveCount(), e.getCorePoolSize(), e.getMaximumPoolSize(),
e.getLargestPoolSize(),
e.getTaskCount(), e.getCompletedTaskCount(), e.isShutdown(), e.isTerminated(), e.isTerminating(),
url.getProtocol(), url.getIp(), url.getPort());
logger.warn(msg); // 记录最大负载情况下线程池的核心线程数,活跃数,最大线程数等参数
dumpJStack(); // 记录线程堆栈信息包括锁争用信息
throw new RejectedExecutionException(msg);
}
private void dumpJStack() {
long now = System.currentTimeMillis();
//dump every 10 minutes 每隔10分钟记录一次
if (now - lastPrintTime < TEN_MINUTES_MILLS) {
return;
}
if (!guard.tryAcquire()) { // 加锁访问
return;
}
ExecutorService pool = Executors.newSingleThreadExecutor(); // 这里单独开启一个新的线程去执行(阿里的Java开发规范不允许直接调用Executors.newSingleThreadExecutor, 估计dubbo那时候还没出开发规范...)
pool.execute(() -> {
String dumpPath = url.getParameter(DUMP_DIRECTORY, System.getProperty("user.home"));
SimpleDateFormat sdf;
String os = System.getProperty(OS_NAME_KEY).toLowerCase();
// window system don't support ":" in file name
if (os.contains(OS_WIN_PREFIX)) {
sdf = new SimpleDateFormat(WIN_DATETIME_FORMAT);
} else {
sdf = new SimpleDateFormat(DEFAULT_DATETIME_FORMAT);
}
String dateStr = sdf.format(new Date());
//try-with-resources
try (FileOutputStream jStackStream = new FileOutputStream(
new File(dumpPath, "Dubbo_JStack.log" + "." + dateStr))) {
JVMUtil.jstack(jStackStream);
} catch (Throwable t) {
logger.error("dump jStack error", t);
} finally {
guard.release();
}
lastPrintTime = System.currentTimeMillis();
});
//must shutdown thread pool ,if not will lead to OOM
pool.shutdown();
}
}
以上理解如果有误，欢迎大佬指正。
参考资料:
- Dubbo线程池拒绝策略: org.apache.dubbo.common.threadpool.support.AbortPolicyWithReport.java
- 《Java并发编程实战》
