Dubbo之线程池设计

前言

Dubbo的线程模型中可使用4种线程池

CachedThreadPool
LimitedThreadPool
FixedThreadPool
EagerThreadPool

想深入了解线程池原理的同学，可以阅读我的线程池那些事专栏。

在Dubbo中什么时候会用到线程池

我们的线程主要执行2种逻辑，一是普通IO事件，比如建立连接，断开连接，二是请求IO事件，执行业务逻辑。
在Dubbo的Dispatcher扩展点会使用到这些线程池，Dispatcher这个扩展点用于决定Netty ChannelHandler中的那些事件在Dubbo提供的线程池中执行。

源码解析

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基本一致。
如果一定要说区别，那就是FixedThreadPool等到创建完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，然后通过workQueue的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的默认配置，我们公司的业务量还没到需要对线程池进行特殊配置的地步。本文主要目的是，通过一个成熟框架对线程池的配置点，指导我们在实际使用线程池中需要注意的点。

Dubbo之线程池设计