Dubbo源码解析线程那些事Springboot

线程池那些事之Dubbo线程池设计

2018-11-07  本文已影响2人  土豆肉丝盖浇饭

前言

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
上一篇 下一篇

猜你喜欢

热点阅读