前言
Dubbo的IO模型中提供了4种线程池，下面我会一一介绍。推荐先阅读我前几篇的线程池解析，更有助于理解Dubbo中的线程池设计。
在Dubbo中什么时候会用到线程池
图片上看起来只有服务端用到了ThreadPool，实际上客户端调用的时候也会用到ThreadPool
Dubbo的线程池扩展用作业务线程池，在Dubbo的Dispatcher模块会用到这些线程池，Dispatcher这个模块用于决定Netty ChannelHandler 那些事件的处理需要在业务线程池执行，对于那些耗时的阶段推荐在业务线程池运行，而不是在IO线程中，Dubbo中默认使用的线程池为LimitedThreadPool。
下面先看下Dubbo提供的4种线程池的源码
源码解析
CachedThreadPool
public class CachedThreadPool implements ThreadPool {
@Override
public Executor getExecutor(URL url) {
String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
int cores = url.getParameter(Constants.CORE_THREADS_KEY, Constants.DEFAULT_CORE_THREADS);
int threads = url.getParameter(Constants.THREADS_KEY, Integer.MAX_VALUE);
int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
int alive = url.getParameter(Constants.ALIVE_KEY, Constants.DEFAULT_ALIVE);
return new ThreadPoolExecutor(cores, threads, alive, TimeUnit.MILLISECONDS,
queues == 0 ? new SynchronousQueue<Runnable>() :
(queues < 0 ? new LinkedBlockingQueue<Runnable>()
: new LinkedBlockingQueue<Runnable>(queues)),
new NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url));
}
}
缓冲线程池，默认配置如下
| 配置 | 配置值 |
|---|---|
| corePoolSize | 0 |
| maximumPoolSize | Integer.MAX_VALUE |
| keepAliveTime | 60s |
| workQueue | 根据queue决定是SynchronousQueue还是LinkedBlockingQueue，默认queue=0，所以是SynchronousQueue |
| threadFactory | NamedInternalThreadFactory |
| rejectHandler | AbortPolicyWithReport |
就默认配置来看，和Executors创建的差不多，存在内存溢出风险。NamedInternalThreadFactory主要用于修改线程名，方便我们排查问题。AbortPolicyWithReport对拒绝的任务打印日志，也是方便排查问题。
LimitedThreadPool
public class LimitedThreadPool implements ThreadPool {
@Override
public Executor getExecutor(URL url) {
String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
int cores = url.getParameter(Constants.CORE_THREADS_KEY, Constants.DEFAULT_CORE_THREADS);
int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);
int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
return new ThreadPoolExecutor(cores, threads, Long.MAX_VALUE, TimeUnit.MILLISECONDS,
queues == 0 ? new SynchronousQueue<Runnable>() :
(queues < 0 ? new LinkedBlockingQueue<Runnable>()
: new LinkedBlockingQueue<Runnable>(queues)),
new NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url));
}
}
| 配置 | 配置值 |
|---|---|
| corePoolSize | 0 |
| maximumPoolSize | 200 |
| keepAliveTime | Long.MAX_VALUE,相当于无限长 |
| workQueue | 根据queue决定是SynchronousQueue还是LinkedBlockingQueue，默认queue=0，所以是SynchronousQueue |
| threadFactory | NamedInternalThreadFactory |
| rejectHandler | AbortPolicyWithReport |
从keepAliveTime的配置可以看出来，LimitedThreadPool线程池的特性是线程数只会增加不会减少。
FixedThreadPool
public class FixedThreadPool implements ThreadPool {
@Override
public Executor getExecutor(URL url) {
String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);
int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
return new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS,
queues == 0 ? new SynchronousQueue<Runnable>() :
(queues < 0 ? new LinkedBlockingQueue<Runnable>()
: new LinkedBlockingQueue<Runnable>(queues)),
new NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url));
}
}
| 配置 | 配置值 |
|---|---|
| corePoolSize | 200 |
| maximumPoolSize | 200 |
| keepAliveTime | 0 |
| workQueue | 根据queue决定是SynchronousQueue还是LinkedBlockingQueue，默认queue=0，所以是SynchronousQueue |
| threadFactory | NamedInternalThreadFactory |
| rejectHandler | AbortPolicyWithReport |
Dubbo的默认线程池，固定200个线程，就配置来看和LimitedThreadPool没什么区别，因为这两种线程池都是只增加。
EagerThreadPool
public class EagerThreadPool implements ThreadPool {
@Override
public Executor getExecutor(URL url) {
String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
int cores = url.getParameter(Constants.CORE_THREADS_KEY, Constants.DEFAULT_CORE_THREADS);
int threads = url.getParameter(Constants.THREADS_KEY, Integer.MAX_VALUE);
int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
int alive = url.getParameter(Constants.ALIVE_KEY, Constants.DEFAULT_ALIVE);
// init queue and executor
TaskQueue<Runnable> taskQueue = new TaskQueue<Runnable>(queues <= 0 ? 1 : queues);
EagerThreadPoolExecutor executor = new EagerThreadPoolExecutor(cores,
threads,
alive,
TimeUnit.MILLISECONDS,
taskQueue,
new NamedInternalThreadFactory(name, true),
new AbortPolicyWithReport(name, url));
taskQueue.setExecutor(executor);
return executor;
}
}
| 配置 | 配置值 |
|---|---|
| corePoolSize | 0 |
| maximumPoolSize | Integer.MAX_VALUE |
| keepAliveTime | 60s |
| workQueue | 自定义实现TaskQueue，默认长度为1，使用时要自己配置下 |
| threadFactory | NamedInternalThreadFactory |
| rejectHandler | AbortPolicyWithReport |
我们知道，当线程数量达到corePoolSize之后，只有当workqueue满了之后，才会增加工作线程。
这个线程池就是对这个特性做了优化，首先继承ThreadPoolExecutor实现EagerThreadPoolExecutor，对当前线程池提交的任务数submittedTaskCount进行记录。
其次是通过自定义TaskQueue作为workQueue，它会在提交任务时判断是否currentPoolSize<submittedTaskCount<maxPoolSize，然后通过它的offer方法返回false导致增加工作线程。
为什么返回false会增加工作线程，我们回顾下ThreadPoolExecutor的execute方法
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
//在这一步如果offer方法返回false，那么会进入到下一个else分支判断
//如果这个时候当前工作线程数没有达到上限，那么就会增加一个工作线程
//对于普通的workQueue在没有满的情况下是不会返回false的
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
else if (!addWorker(command, false))
reject(command);
}
然后看下TaskQueue的offer方法逻辑
public boolean offer(Runnable runnable) {
//TaskQueue持有executor引用，用于获取当前提交任务数
if (executor == null) {
throw new RejectedExecutionException("The task queue does not have executor!");
}
int currentPoolThreadSize = executor.getPoolSize();
//如果提交任务数小于当前工作线程数，说明当前工作线程足够处理任务，将提交的任务插入到工作队列
if (executor.getSubmittedTaskCount() < currentPoolThreadSize) {
return super.offer(runnable);
}
//如果提交任务数大于当前工作线程数并且小于最大线程数，说明提交的任务量线程已经处理不过来，那么需要增加线程数，返回false
if (currentPoolThreadSize < executor.getMaximumPoolSize()) {
return false;
}
//工作线程数到达最大线程数，插入到workqueue
return super.offer(runnable);
}
再看下修改的EagerThreadPoolExecutor如何统计提交的任务
public int getSubmittedTaskCount() {
return submittedTaskCount.get();
}
//任务执行完毕后，submittedTaskCount--
@Override
protected void afterExecute(Runnable r, Throwable t) {
submittedTaskCount.decrementAndGet();
}
@Override
public void execute(Runnable command) {
if (command == null) {
throw new NullPointerException();
}
//执行任务前，submittedTaskCount++
submittedTaskCount.incrementAndGet();
try {
super.execute(command);
} catch (RejectedExecutionException rx) {
//失败尝试重新加入到workqueue
final TaskQueue queue = (TaskQueue) super.getQueue();
try {
if (!queue.retryOffer(command, 0, TimeUnit.MILLISECONDS)) {
//如果重新加入失败，那么抛出异常，并且统计数-1
submittedTaskCount.decrementAndGet();
throw new RejectedExecutionException("Queue capacity is full.");
}
} catch (InterruptedException x) {
submittedTaskCount.decrementAndGet();
throw new RejectedExecutionException(x);
}
} catch (Throwable t) {
// decrease any way
submittedTaskCount.decrementAndGet();
}
}
最后
一般来讲使用Dubbo的默认配置，我们公司的业务量还没到需要对线程池进行特殊配置的地步。本文主要目的是，通过一个成熟框架对线程池的配置点，指导我们在实际使用线程池中需要注意的点。
下面是我公众号，大家可以关注下。
image
网友评论