源码分析Dubbo线程池实现原理

中间件兴趣圈 2022-11-10

微信公众号：[中间件兴趣圈]
作者简介：《RocketMQ技术内幕》作者

本文主要分析Dubbo线程池的构建过程，主要介绍官方文档中有关于ThreadPool的种类：

fixed
固定大小线程池，启动时建立线程，不关闭，一致持有。（缺省）
cached
缓存线程池，空闲一分钟，线程会消费，需要时重新创建新线程。
limited
可伸缩线程池，但池中的线程数只会增长不会收缩。
eager
优先使用线程来执行新提交任务。（渴望立即执行，而不是进入队列排队执行）。
配置标签：< dubbo:protocol threadpool = "fixed" ../>

由于各种类型的线程池，内部是根据规则创建不同的ThreadPoolExecutor对象，那我们先简单回顾一下线程池的基本知识。

线程池核心属性

int corePoolSize
线程池核心线程数、常驻线程数
int maximumPoolSize
线程池中最大线程数量
long keepAliveTime
线程保持活跃时间,（如果线程创建，并空闲指定值后，线程会被回收，0表示不开启该特性,其范围针对超过corePoolSize的线程）
TimeUnit unit
keepAliveTime的时间单位。
BlockingQueue< Runnable> workQueue
任务队列
ThreadFactory threadFactory
线程工厂类，一般通过该线程工厂，为线程命名，以便区分线程。
RejectedExecutionHandler handler)
拒绝策略。

线程池任务提交流程

如果线程池中线程数量小于corePoolSize，则创建一个线程来执行该任务。
如果线程池中的线程大于等于corePoolSize，则尝试将任务放入队列中。
如果成功将任务放入队列，则本次提交任务正常结束，如果放入任务队列失败则继续下一步。
如果线程池中的线程数量小于最大线程数，则创建先的线程，否则执行拒绝策略。

更多有关线程池，请参考作者的另外一篇博文：https://blog.csdn.net/prestigeding/article/details/53929713

解析来我们将一一分析Dubbo支持的线程池类型。

fixed

固定大小线程池。

 1public class FixedThreadPool implements ThreadPool {
 2    @Override
 3    public Executor getExecutor(URL url) {
 4        String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
 5        int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);
 6        int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
 7        return new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS,
 8                queues == 0 ? new SynchronousQueue<Runnable>() :
 9                        (queues < 0 ? new LinkedBlockingQueue<Runnable>()
10                                : new LinkedBlockingQueue<Runnable>(queues)),
11                new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url));
12    }
13}

实现要点：
1、首先获取可配置参数threadname、threads、queues三个参数，分别代表线程池中线程名前缀、线程中最大线程数量、任务队列长度。

2、要实现fixed固定大小线程池，故名思议，就是线程池自创建以来，线程数量始终保持一致。其实现要点是，corePoolSize、maximumPoolSize相等，并且其值等于threads(默认200)，并且keepAliveTime=0，表示线程始终活跃。

3、任务队列，如果queues 为0，则使用SynchronousQueue，如果小于0，则使用无界队列，如果大于0，则创建容量为LinkedBlockingQueue的队列，超过容量，则拒绝入队。

4、线程工厂，NamedThreadFactory，主要设置线程名称，默认为Dubbo-thread-序号。

5、拒绝策略AbortPolicyWithReport，其主要是如果拒绝任务，首先会打印出详细日志，包含线程池的核心参数，并且会dump jstack日志，日志文件默认存储在user.home/Dubbo_JStack.log.timestamp，可以通过dump.directory属性配置，可通过<dubbo:protocol> <dubbo:parameter key =“” value = ""/> < /dubbo:protocol>。

与该线程池相关的配置属性：threadname、theadpool、threads、queues。

这里再简单介绍如果队列长度为0（默认），为什么是选用SynchronousQueue队列。

SynchronousQueue的一个简单理解：调用offer、put之前，必须先调用take，也就是先调用take方法的线程阻塞，然后当别的线程调用offer之后，调用take的线程被唤醒，如果没有线程调用take方法，一个线程调用offer方法，则会返回false，并不会将元素添加到SynchronousQueue队列中，因为SynchronousQueue内部的队列长度为0。

cached

缓存线程池，线程空闲后会被回收。

 1public class CachedThreadPool implements ThreadPool {
 2    @Override
 3    public Executor getExecutor(URL url) {
 4        String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
 5        int cores = url.getParameter(Constants.CORE_THREADS_KEY, Constants.DEFAULT_CORE_THREADS);
 6        int threads = url.getParameter(Constants.THREADS_KEY, Integer.MAX_VALUE);
 7        int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
 8        int alive = url.getParameter(Constants.ALIVE_KEY, Constants.DEFAULT_ALIVE);
 9        return new ThreadPoolExecutor(cores, threads, alive, TimeUnit.MILLISECONDS,
10                queues == 0 ? new SynchronousQueue<Runnable>() :
11                        (queues < 0 ? new LinkedBlockingQueue<Runnable>()
12                                : new LinkedBlockingQueue<Runnable>(queues)),
13                new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url));
14    }
15}

实现要点：既然要实现线程可以被回收，则必然要设置keepAliveTime。
故对应线程池核心参数设置，对应如下：

corePoolSize：通过参数corethreads设置，默认为0
maximumPoolSize：通过参数threads设置，默认200
keepAliveTime：通过参数alive设置，默认为60 * 1000
workQueue ：通过queues参数设置，默认为0
其他与fixed相同，则不重复介绍

limited

可伸缩线程池，其特征：线程数只增不减。

 1public class LimitedThreadPool implements ThreadPool {
 2    @Override
 3    public Executor getExecutor(URL url) {
 4        String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
 5        int cores = url.getParameter(Constants.CORE_THREADS_KEY, Constants.DEFAULT_CORE_THREADS);
 6        int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);
 7        int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
 8        return new ThreadPoolExecutor(cores, threads, Long.MAX_VALUE, TimeUnit.MILLISECONDS,
 9                queues == 0 ? new SynchronousQueue<Runnable>() :
10                        (queues < 0 ? new LinkedBlockingQueue<Runnable>()
11                                : new LinkedBlockingQueue<Runnable>(queues)),
12                new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url));
13    }
14}

与cached不同的是，limited线程池中线程一旦创建，就不回收，与cached不同的就是keepAliveTime的取值不同，limited取值为：Long.MAX_VALUE,其他与cached相同。

eager

其核心实现主要由TaskQueue、EagerThreadPoolExecutor共同完成。

首先，我们关注一下TaskQueued的offer方法。

 1public boolean offer(Runnable runnable) {
 2        if (executor == null) {
 3            throw new RejectedExecutionException("The task queue does not have executor!");
 4        }
 5
 6        int currentPoolThreadSize = executor.getPoolSize();     // @1
 7        // have free worker. put task into queue to let the worker deal with task.
 8        if (executor.getSubmittedTaskCount() < currentPoolThreadSize) {   // @2
 9            return super.offer(runnable);
10        }
11
12        // return false to let executor create new worker.
13        if (currentPoolThreadSize < executor.getMaximumPoolSize()) {    // @3
14            return false;
15        }
16
17        // currentPoolThreadSize >= max     // @4
18        return super.offer(runnable); 
19    }

代码@1：获取当前线程池中线程的数量。

代码@2：如果当前已提交到线程池中的任务数量小于当前存在在的线程数，则走默认的提交流程。

代码@3：如果当前已提交到线程中的数量大于当前的线程池，并线程池中数量并未达到线程池允许创建的最大线程数时，则返回false，并不入队，其效果是会创建新的线程来执行。

代码@4：如果当前线程池中的线程已达到允许创建的最大线程数后，走默认的提交任务逻辑。

其次看一下EagerThreadPoolExecutor#execute

 1public void execute(Runnable command) {
 2        if (command == null) {
 3            throw new NullPointerException();
 4        }
 5        // do not increment in method beforeExecute!
 6        submittedTaskCount.incrementAndGet();       // @1 
 7        try {
 8            super.execute(command);
 9        } catch (RejectedExecutionException rx) {
10            // retry to offer the task into queue.
11            final TaskQueue queue = (TaskQueue) super.getQueue();
12            try {
13                if (!queue.retryOffer(command, 0, TimeUnit.MILLISECONDS)) {
14                    submittedTaskCount.decrementAndGet();
15                    throw new RejectedExecutionException("Queue capacity is full.");
16                }
17            } catch (InterruptedException x) {
18                submittedTaskCount.decrementAndGet();
19                throw new RejectedExecutionException(x);
20            }
21        } catch (Throwable t) {
22            // decrease any way
23            submittedTaskCount.decrementAndGet();   // @2
24        }
25    }

其核心实现逻辑：如果提交任务失败，则再走一次默认的任务提交流程。
最总后结一下Eager的核心特性。

 1public class EagerThreadPool implements ThreadPool {
 2    @Override
 3    public Executor getExecutor(URL url) {
 4        String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
 5        int cores = url.getParameter(Constants.CORE_THREADS_KEY, Constants.DEFAULT_CORE_THREADS);
 6        int threads = url.getParameter(Constants.THREADS_KEY, Integer.MAX_VALUE);
 7        int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
 8        int alive = url.getParameter(Constants.ALIVE_KEY, Constants.DEFAULT_ALIVE);
 9
10        // init queue and executor
11        TaskQueue<Runnable> taskQueue = new TaskQueue<Runnable>(queues <= 0 ? 1 : queues);
12        EagerThreadPoolExecutor executor = new EagerThreadPoolExecutor(cores,
13                threads,
14                alive,
15                TimeUnit.MILLISECONDS,
16                taskQueue,
17                new NamedThreadFactory(name, true),
18                new AbortPolicyWithReport(name, url));
19        taskQueue.setExecutor(executor);
20        return executor;
21    }
22}

其核心特性如下：
1、首先，其配置参数与cached类型的线程池相同，说明eager也是基于缓存的。

2、eager与cached类型线程池不同的一点是，提交任务后，线程优先于队列，默认的提交流程是如果线程数达到核心线程数后，新提交的任务是首先进入队列，但eager是优先创建线程来执行，这有点与公平锁，非公平锁一样的概念了。

Dubbo的线程池种类就介绍到这里。

广告：作者新书《RocketMQ技术内幕》已上市

《RocketMQ技术内幕》已出版上市，目前可在主流购物平台（京东、天猫等）购买，本书从源码角度深度分析了RocketMQ NameServer、消息发送、消息存储、消息消费、消息过滤、主从同步HA、事务消息；在实战篇重点介绍了RocketMQ运维管理界面与当前支持的39个运维命令；并在附录部分罗列了RocketMQ几乎所有的配置参数。本书得到了RocketMQ创始人、阿里巴巴Messaging开源技术负责人、Linux OpenMessaging 主席的高度认可并作序推荐。目前是国内第一本成体系剖析RocketMQ的书籍。
新书7折优惠！7折优惠！7折优惠！

更多文章请关注微信公众号：

推荐关注微信公众号：RocketMQ官方微信公众号：

薄公子低调成台湾女婿 23日已在台举办婚礼

警察殴打打人学生，舆论撕裂的背后

你手放哪呢，出生啊

薅广电羊毛！100元话费实付94.6元，还有电费96.9充100元！招团长~

警察踢打校园欺凌者：当事人不愿返校，派出所拒收锦旗

源码分析Dubbo线程池实现原理

线程池核心属性

线程池任务提交流程

fixed

cached

limited

eager

您可能也对以下帖子感兴趣

薄公子低调成台湾女婿 23日已在台举办婚礼

警察殴打打人学生，舆论撕裂的背后

你手放哪呢，出生啊​

薅广电羊毛！100元话费实付94.6元，还有电费96.9充100元！招团长~

警察踢打校园欺凌者：当事人不愿返校，派出所拒收锦旗

生成图片，分享到微信朋友圈

源码分析Dubbo线程池实现原理

线程池核心属性

线程池任务提交流程

fixed

cached

limited

eager

您可能也对以下帖子感兴趣

你手放哪呢，出生啊