DUBBO线程池策略
### 基本概念
上个章节分析了线程模型，我们知道不同的线程模型会选择使用还是IO线程还是业务线程。如果使用业务线程池，那么使用什么线程池策略是本章节需要回答的问题。DUBBO官网线程派发模型图展示了线程模型和线程池策略的关系：
DUBBO提供了多种线程池策略，选择线程池策略需要在配置文件指定threadpool属性
```java
<dubbo:protocol name="dubbo" threadpool="fixed" threads="100" />
<dubbo:protocol name="dubbo" threadpool="cached" threads="100" />
<dubbo:protocol name="dubbo" threadpool="limited" threads="100" />
<dubbo:protocol name="dubbo" threadpool="eager" threads="100" />
```
不同线程池策略会创建不同特性的线程池：
```java
fixed
包含固定个数线程
cached
线程空闲一分钟会被回收，当新请求到来时会创建新线程
limited
线程个数随着任务增加而增加，但不会超过最大阈值。空闲线程不会被回收
eager
当所有核心线程数都处于忙碌状态时，优先创建新线程执行任务，而不是立即放入队列
```
### 确定时机
本文我们以AllDispatcher为例分析线程池策略在什么时候确定：
```java
public class AllDispatcher implements Dispatcher {
public static final String NAME = "all";
@Override
public ChannelHandler dispatch(ChannelHandler handler, URL url) {
return new AllChannelHandler(handler, url);
}
}
public class AllChannelHandler extends WrappedChannelHandler {
public AllChannelHandler(ChannelHandler handler, URL url) {
super(handler, url);
}
}
```
在WrappedChannelHandler构造函数中如果配置指定了threadpool属性，扩展点加载器会从URL获取属性值加载对应线程池策略，默认策略为fixed：
```java
public class WrappedChannelHandler implements ChannelHandlerDelegate {
public WrappedChannelHandler(ChannelHandler handler, URL url) {
this.handler = handler;
this.url = url;
// 获取线程池自适应扩展点
executor = (ExecutorService) ExtensionLoader.getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url);
String componentKey = Constants.EXECUTOR_SERVICE_COMPONENT_KEY;
if (Constants.CONSUMER_SIDE.equalsIgnoreCase(url.getParameter(Constants.SIDE_KEY))) {
componentKey = Constants.CONSUMER_SIDE;
}
DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension();
dataStore.put(componentKey, Integer.toString(url.getPort()), executor);
}
}
@SPI("fixed")
public interface ThreadPool {
@Adaptive({Constants.THREADPOOL_KEY})
Executor getExecutor(URL url);
}
```
### 源码分析
(1) FixedThreadPool
```java
public class FixedThreadPool implements ThreadPool {
@Override
public Executor getExecutor(URL url) {
// 线程名称
String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
// 线程个数默认200
int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);
// 队列容量默认0
int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
// 队列容量等于0使用阻塞队列SynchronousQueue
// 队列容量小于0使用无界阻塞队列LinkedBlockingQueue
// 队列容量大于0使用有界阻塞队列LinkedBlockingQueue
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));
}
}
```
(2) CachedThreadPool
```java
public class CachedThreadPool implements ThreadPool {
@Override
public Executor getExecutor(URL url) {
// 获取线程名称
String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
// 核心线程数默认0
int cores = url.getParameter(Constants.CORE_THREADS_KEY, Constants.DEFAULT_CORE_THREADS);
// 最大线程数默认Int最大值
int threads = url.getParameter(Constants.THREADS_KEY, Integer.MAX_VALUE);
// 队列容量默认0
int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
// 线程空闲多少时间被回收默认1分钟
int alive = url.getParameter(Constants.ALIVE_KEY, Constants.DEFAULT_ALIVE);
// 队列容量等于0使用阻塞队列SynchronousQueue
// 队列容量小于0使用无界阻塞队列LinkedBlockingQueue
// 队列容量大于0使用有界阻塞队列LinkedBlockingQueue
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));
}
}
```
(3) LimitedThreadPool
```java
public class LimitedThreadPool implements ThreadPool {
@Override
public Executor getExecutor(URL url) {
// 获取线程名称
String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
// 核心线程数默认0
int cores = url.getParameter(Constants.CORE_THREADS_KEY, Constants.DEFAULT_CORE_THREADS);
// 最大线程数默认200
int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);
// 队列容量默认0
int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
// 队列容量等于0使用阻塞队列SynchronousQueue
// 队列容量小于0使用无界阻塞队列LinkedBlockingQueue
// 队列容量大于0使用有界阻塞队列LinkedBlockingQueue
// keepalive时间设置Long.MAX_VALUE表示不回收空闲线程
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));
}
}
```
(4) EagerThreadPool
我们知道ThreadPoolExecutor是普通线程执行器。当线程池核心线程达到阈值时新任务放入队列，当队列已满开启新线程处理，当前线程数达到最大线程数时执行拒绝策略。
但是EagerThreadPool自定义线程执行策略，当线程池核心线程达到阈值时，新任务不会放入队列而是开启新线程进行处理（要求当前线程数没有超过最大线程数）。当前线程数达到最大线程数时任务放入队列。
```java
public class EagerThreadPool implements ThreadPool {
@Override
public Executor getExecutor(URL url) {
// 线程名
String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
// 核心线程数默认0
int cores = url.getParameter(Constants.CORE_THREADS_KEY, Constants.DEFAULT_CORE_THREADS);
// 最大线程数默认Int最大值
int threads = url.getParameter(Constants.THREADS_KEY, Integer.MAX_VALUE);
// 队列容量默认0
int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
// 线程空闲多少时间被回收默认1分钟
int alive = url.getParameter(Constants.ALIVE_KEY, Constants.DEFAULT_ALIVE);
// 初始化自定义线程池和队列重写相关方法
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;
}
}
```
公众号进入网站体验更佳
