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

Posted Java后端笔记

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了线程池那些事之Dubbo线程池设计相关的知识,希望对你有一定的参考价值。

前言

Dubbo的IO模型中提供了4种线程池,下面我会一一介绍。推荐先阅读我前几篇的线程池解析,更有助于理解Dubbo中的线程池设计。

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

图片上看起来只有服务端用到了ThreadPool,实际上客户端调用的时候也会用到ThreadPool

Dubbo的线程池扩展用作业务线程池,在Dubbo的Dispatcher模块会用到这些线程池,Dispatcher这个模块用于决定Netty ChannelHandler 那些事件的处理需要在业务线程池执行,对于那些耗时的阶段推荐在业务线程池运行,而不是在IO线程中,Dubbo中默认使用的线程池为LimitedThreadPool。

下面先看下Dubbo提供的4种线程池的源码

源码解析

CachedThreadPool

 
   
   
 
  1. public 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 NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url));

  14.    }

  15. }

缓冲线程池,默认配置如下

配置 配置值
corePoolSize 0
maximumPoolSize Integer.MAX_VALUE
keepAliveTime 60s
workQueue 根据queue决定是SynchronousQueue还是LinkedBlockingQueue,默认queue=0,所以是SynchronousQueue
threadFactory NamedInternalThreadFactory
rejectHandler AbortPolicyWithReport

就默认配置来看,和Executors创建的差不多,存在内存溢出风险。NamedInternalThreadFactory主要用于修改线程名,方便我们排查问题。AbortPolicyWithReport对拒绝的任务打印日志,也是方便排查问题。

LimitedThreadPool

 
   
   
 
  1. public 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 NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url));

  13.    }

  14. }

配置 配置值
corePoolSize 0
maximumPoolSize 200
keepAliveTime Long.MAX_VALUE,相当于无限长
workQueue 根据queue决定是SynchronousQueue还是LinkedBlockingQueue,默认queue=0,所以是SynchronousQueue
threadFactory NamedInternalThreadFactory
rejectHandler AbortPolicyWithReport

从keepAliveTime的配置可以看出来,LimitedThreadPool线程池的特性是线程数只会增加不会减少。

FixedThreadPool

 
   
   
 
  1. public 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 NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url));

  12.    }

  13. }

配置 配置值
corePoolSize 200
maximumPoolSize 200
keepAliveTime 0
workQueue 根据queue决定是SynchronousQueue还是LinkedBlockingQueue,默认queue=0,所以是SynchronousQueue
threadFactory NamedInternalThreadFactory
rejectHandler AbortPolicyWithReport

Dubbo的默认线程池,固定200个线程,就配置来看和LimitedThreadPool没什么区别,因为这两种线程池都是只增加。

EagerThreadPool

 
   
   
 
  1. public 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.        // init queue and executor

  10.        TaskQueue<Runnable> taskQueue = new TaskQueue<Runnable>(queues <= 0 ? 1 : queues);

  11.        EagerThreadPoolExecutor executor = new EagerThreadPoolExecutor(cores,

  12.                threads,

  13.                alive,

  14.                TimeUnit.MILLISECONDS,

  15.                taskQueue,

  16.                new NamedInternalThreadFactory(name, true),

  17.                new AbortPolicyWithReport(name, url));

  18.        taskQueue.setExecutor(executor);

  19.        return executor;

  20.    }

  21. }

配置 配置值
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导致增加工作线程。< p="" style="box-sizing: border-box;"></submittedtaskcount<maxpoolsize,然后通过它的offer方法返回false导致增加工作线程。<>

为什么返回false会增加工作线程,我们回顾下ThreadPoolExecutor的execute方法

 
   
   
 
  1. public void execute(Runnable command) {

  2.        if (command == null)

  3.            throw new NullPointerException();

  4.        int c = ctl.get();

  5.        if (workerCountOf(c) < corePoolSize) {

  6.            if (addWorker(command, true))

  7.                return;

  8.            c = ctl.get();

  9.        }

  10.        //在这一步如果offer方法返回false,那么会进入到下一个else分支判断

  11.        //如果这个时候当前工作线程数没有达到上限,那么就会增加一个工作线程

  12.        //对于普通的workQueue在没有满的情况下是不会返回false的

  13.        if (isRunning(c) && workQueue.offer(command)) {

  14.            int recheck = ctl.get();

  15.            if (! isRunning(recheck) && remove(command))

  16.                reject(command);

  17.            else if (workerCountOf(recheck) == 0)

  18.                addWorker(null, false);

  19.        }

  20.        else if (!addWorker(command, false))

  21.            reject(command);

  22.    }

然后看下TaskQueue的offer方法逻辑

 
   
   
 
  1. public boolean offer(Runnable runnable) {

  2.        //TaskQueue持有executor引用,用于获取当前提交任务数

  3.        if (executor == null) {

  4.            throw new RejectedExecutionException("The task queue does not have executor!");

  5.        }

  6.        int currentPoolThreadSize = executor.getPoolSize();

  7.        //如果提交任务数小于当前工作线程数,说明当前工作线程足够处理任务,将提交的任务插入到工作队列

  8.        if (executor.getSubmittedTaskCount() < currentPoolThreadSize) {

  9.            return super.offer(runnable);

  10.        }

  11.        //如果提交任务数大于当前工作线程数并且小于最大线程数,说明提交的任务量线程已经处理不过来,那么需要增加线程数,返回false

  12.        if (currentPoolThreadSize < executor.getMaximumPoolSize()) {

  13.            return false;

  14.        }

  15.        //工作线程数到达最大线程数,插入到workqueue

  16.        return super.offer(runnable);

  17.    }

再看下修改的EagerThreadPoolExecutor如何统计提交的任务

 
   
   
 
  1.    public int getSubmittedTaskCount() {

  2.        return submittedTaskCount.get();

  3.    }

  4.    //任务执行完毕后,submittedTaskCount--

  5.    @Override

  6.    protected void afterExecute(Runnable r, Throwable t) {

  7.        submittedTaskCount.decrementAndGet();

  8.    }

  9.    @Override

  10.    public void execute(Runnable command) {

  11.        if (command == null) {

  12.            throw new NullPointerException();

  13.        }

  14.        //执行任务前,submittedTaskCount++

  15.        submittedTaskCount.incrementAndGet();

  16.        try {

  17.            super.execute(command);

  18.        } catch (RejectedExecutionException rx) {

  19.           //失败尝试重新加入到workqueue

  20.            final TaskQueue queue = (TaskQueue) super.getQueue();

  21.            try {

  22.                if (!queue.retryOffer(command, 0, TimeUnit.MILLISECONDS)) {

  23.                    //如果重新加入失败,那么抛出异常,并且统计数-1

  24.                    submittedTaskCount.decrementAndGet();

  25.                    throw new RejectedExecutionException("Queue capacity is full.");

  26.                }

  27.            } catch (InterruptedException x) {

  28.                submittedTaskCount.decrementAndGet();

  29.                throw new RejectedExecutionException(x);

  30.            }

  31.        } catch (Throwable t) {

  32.            // decrease any way

  33.            submittedTaskCount.decrementAndGet();

  34.        }

  35.    }

最后

一般来讲使用Dubbo的默认配置,我们公司的业务量还没到需要对线程池进行特殊配置的地步。本文主要目的是,通过一个成熟框架对线程池的配置点,指导我们在实际使用线程池中需要注意的点。


以上是关于线程池那些事之Dubbo线程池设计的主要内容,如果未能解决你的问题,请参考以下文章

线程池那些事之ScheduledThreadPoolExecutor

线程池那些事之Future

Dubbo之线程池设计

JDK线程池Tomcat线程池Dubbo线程池的不同

稳定性 耗时 监控原因分析-- dubbo rpc 框架 的线程池,io 连接模型. 客户端,服务端

dubbo-线程池监控