「DUBBO系列」线程池策略源码分析

朱雀 2021-09-24 09:14 351阅读 0赞

# 1 文章概述 #

本系列文章已经分析了DUBBO线程模型实现原理，本文简单进行回顾。我们知道DUBBO提供五种线程模型

all
    所有消息都派发到业务线程池，包括请求，响应，连接事件，断开事件，心跳
    
    direct
    所有消息都不派发到业务线程池，全部在IO线程直接执行
    
    message
    只有请求响应消息派发到业务线程池，其它连接断开事件，心跳等消息直接在IO线程执行
    
    execution
    只有请求消息派发到业务线程池，响应和其它连接断开事件，心跳等消息直接在IO线程执行
    
    connection
    在IO线程上将连接断开事件放入队列，有序逐个执行，其它消息派发到业务线程池

不同线程模型会选择使用IO线程还是业务线程，那么业务线程池采用什么线程池策略是本文需要回答的问题。DUBBO提供了多种线程池策略，选择线程池策略需要在配置文件指定threadpool属性

<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" />

不同线程池策略会创建不同线程池，不同线程池处理任务方式也不相同

fixed
    固定线程数量
    
    cached
    线程空闲一分钟会被回收，当新请求到来时会创建新线程
    
    limited
    线程个数随着任务增加而增加，但不会超过最大阈值。空闲线程不会被回收
    
    eager
    当所有核心线程数都处于忙碌状态时，优先创建新线程执行任务

# 2 线程池策略确认时机 #

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构造函数

public class WrappedChannelHandler implements ChannelHandlerDelegate {
    
        public WrappedChannelHandler(ChannelHandler handler, URL url) {
            this.handler = handler;
            this.url = url;
            // 获取线程池自适应扩展点默认FixedThreadPool
            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);
        }
    }

如果配置指定threadpool属性，扩展点加载器会从URL获取属性值加载对应线程池策略

@SPI("fixed")
    public interface ThreadPool {
        @Adaptive({Constants.THREADPOOL_KEY})
        Executor getExecutor(URL url);
    }

# 3 源码分析 #

## 3.1 FixedThreadPool ##

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));
        }
    }

## 3.2 CachedThreadPool ##

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.3 LimitedThreadPool ##

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));
        }
    }

## 3.4 EagerThreadPool ##

通过自定义线程执行器可以自定义线程处理策略，我们进行一个对比。

ThreadPoolExecutor普通线程执行器。当线程池核心线程达到阈值时新任务放入队列。当队列已满开启新线程处理。当前线程数达到最大线程数时执行拒绝策略。

EagerThreadPoolExecutor自定义线程执行器。当线程池核心线程达到阈值时，新任务不会放入队列而是开启新线程进行处理（要求当前线程数没有超过最大线程数）。当前线程数达到最大线程数时任务放入队列。队列已满执行拒绝策略。

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;
        }
    }
    
    public class EagerThreadPoolExecutor extends ThreadPoolExecutor {
    
        // 提交任务个数
        private final AtomicInteger submittedTaskCount = new AtomicInteger(0);
    
        @Override
        public void execute(Runnable command) {
            if (command == null) {
                throw new NullPointerException();
            }
            // 任务数自增
            submittedTaskCount.incrementAndGet();
            try {
                // 调用父类方法执行线程任务
                super.execute(command);
            }
            // 抛出拒绝异常
            catch (RejectedExecutionException rx) {
                final TaskQueue queue = (TaskQueue) super.getQueue();
                try {
                    // 任务重新放入队列
                    if (!queue.retryOffer(command, 0, TimeUnit.MILLISECONDS)) {
                        // 任务重新放入队列失败抛出异常
                        submittedTaskCount.decrementAndGet();
                        throw new RejectedExecutionException("Queue capacity is full.", rx);
                    }
                } catch (InterruptedException x) {
                    submittedTaskCount.decrementAndGet();
                    throw new RejectedExecutionException(x);
                }
            } catch (Throwable t) {
                submittedTaskCount.decrementAndGet();
                throw t;
            }
        }
    }
    
    public class TaskQueue<R extends Runnable> extends LinkedBlockingQueue<Runnable> {
    
        @Override
        public boolean offer(Runnable runnable) {
            if (executor == null) {
                throw new RejectedExecutionException("The task queue does not have executor");
            }
    
            // 当前线程数
            int currentPoolThreadSize = executor.getPoolSize();
    
            // 任务数 < 当前线程数表示存在空闲worker线程则任务放入队列等待worker线程处理
            if (executor.getSubmittedTaskCount() < currentPoolThreadSize) {
                return super.offer(runnable);
            }
    
            // 当前线程数 < 最大线程数返回false表示创建worker线程
            if (currentPoolThreadSize < executor.getMaximumPoolSize()) {
                return false;
            }
            // 当前线程数 > 最大线程数任务放入队列
            return super.offer(runnable);
        }
    
        public boolean retryOffer(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {
            if (executor.isShutdown()) {
                throw new RejectedExecutionException("Executor is shutdown!");
            }
            // 任务重试放入队列
            return super.offer(o, timeout, unit);
        }
    }

# 4 文章总结 #

本文分析了DUBBO线程池策略源码，可以根据不同业务场景选择不同线程池策略。后续文章我们深入分析「DUBBO线程池打满问题」请继续关注。

> 扫描二维码关注公众号【IT徐胖子】获取更多互联网和技术干货，感谢各位支持

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dvc2hpeHV5ZQ_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dvc2hpeHV5ZQ_size_16_color_FFFFFF_t_70]: /images/20210923/aedee28d717443779dd3877e7b2a911b.png