Dubbo线程池策略详解：四种实现与应用场景

少年梁大锤

1. 嵌入式电源设计的核心挑战

在嵌入式系统开发中，电源设计往往是最容易被忽视却又最关键的环节。我见过太多项目因为电源问题导致系统不稳定，甚至批量返工。不同于通用计算机电源，嵌入式电源需要面对三大特殊挑战：

首先是空间限制。多数嵌入式设备体积有限，电源模块往往被压缩在PCB的边角位置。我曾参与一个智能家居网关项目，最初设计的电源模块占用了30%的PCB面积，后来通过改用集成度更高的PMIC（电源管理集成电路）才解决问题。

其次是动态负载特性。嵌入式系统的电流需求可能从微安级（睡眠模式）瞬间跃升至安培级（射频模块启动）。某工业传感器项目就因未考虑MCU唤醒时的电流尖峰，导致无线模块频繁掉线。后来我们通过增加470μF的MLCC电容组才稳定了电压。

第三是环境适应性。汽车电子要承受-40℃~125℃的温度变化，工业设备需要抵抗电磁干扰。有个农业物联网案例中，电源电路在实验室表现完美，但在温室大棚里却因湿度导致LDO失效。最终改用密封式DC-DC模块才解决。

2. 电源拓扑结构选型指南

2.1 线性稳压器(LDO)的应用场景

LDO因其低噪声特性，特别适合为模拟电路供电。在医疗级心电监测设备中，我们使用TPS7A4700为ADC提供超洁净的3.3V电源。但要注意：

压差(Vin-Vout)一般需保持1V以上
效率η≈Vout/Vin，当Vin=5V、Vout=3.3V时效率仅66%
-# 1. 概述

本文分享 Dubbo 的线程池策略。在《精尽 Dubbo 源码分析 —— 线程池》一文中，我们已经看到，Dubbo 提供了四种不同的线程池实现，分别是：

fixed 固定大小线程池
limited 可伸缩线程池
cached 缓存线程池
eager 优先创建Worker线程池

本文涉及的类，如下图所示：

2. ThreadPool

com.alibaba.dubbo.common.threadpool.ThreadPool ，线程池接口。代码如下：

Java复制@SPI("fixed")
public interface ThreadPool {

    /**
     * 线程池
     *
     * @param url 线程参数
     * @return 线程池
     */
    @Adaptive({Constants.THREADPOOL_KEY})
    Executor getExecutor(URL url);

}

@SPI("fixed") 注解，Dubbo SPI 拓展点，默认为 "fixed" 。
@Adaptive({Constants.THREADPOOL_KEY}) 注解，基于 Dubbo SPI Adaptive 机制，加载对应的线程池实现，使用 URL.threadpool 属性。
#getExecutor(url) 方法，获得对应的线程池的执行器。

3. FixedThreadPool

com.alibaba.dubbo.common.threadpool.support.fixed.FixedThreadPool ，实现 ThreadPool 接口，固定大小线程池，启动时建立线程，不关闭，一直持有。代码如下：

Java复制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 NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url));
    }

}

第 5 行：线程名。优先从 URL.threadname 参数获得，否则使用 "Dubbo" 。
第 7 行：线程数。优先从 URL.threads 参数获得，否则使用 200 。
第 9 行：队列数。优先从 URL.queues 参数获得，否则使用 0 。
- queues == 0 ， SynchronousQueue 对象。
- queues < 0 ， LinkedBlockingQueue 对象。
- queues > 0 ，带队列数的 LinkedBlockingQueue 对象。
第 10 至 14 行：创建线程池执行器。

4. LimitedThreadPool

com.alibaba.dubbo.common.threadpool.support.limited.LimitedThreadPool ，实现 ThreadPool 接口，可伸缩线程池，但池中的线程数只会增长不会收缩。只增长不收缩的目的是为了避免收缩时突然来了大流量引起的性能问题。代码如下：

Java复制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 NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url));
    }

}

和 FixedThreadPool 实现思路一致，差异点在 cores 和 threads 参数上。

5. CachedThreadPool

com.alibaba.dubbo.common.threadpool.support.cached.CachedThreadPool ，实现 ThreadPool 接口，缓存线程池，空闲一定时长，自动删除，需要时重建。代码如下：

Java复制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 NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url));
    }

}

和 FixedThreadPool 实现思路一致，差异点在 cores 和 threads 参数上。
第 13 行：alive 参数，线程空闲时长。优先从 URL.alive 参数获得，否则使用 60 * 1000 毫秒。

6. EagerThreadPool

com.alibaba.dubbo.common.threadpool.support.eager.EagerThreadPool ，实现 ThreadPool 接口，优先创建Worker线程池。在任务数大于 cores 时，优先创建线程到 threads 数量，而非放入队列。代码如下：

Java复制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);
        // 创建队列
        TaskQueue<Runnable> taskQueue = new TaskQueue<Runnable>(queues <= 0 ? 1 : queues);
        // 创建线程池
        EagerThreadPoolExecutor executor = new EagerThreadPoolExecutor(cores,
                threads,
                alive,
                TimeUnit.MILLISECONDS,
                taskQueue,
                new NamedThreadFactory(name, true),
                new AbortPolicyWithReport(name, url));
        taskQueue.setExecutor(executor);
        return executor;
    }

}

第 13 行：创建 TaskQueue 对象。注意，初始长度是 queues <= 0 ? 1 : queues ，避免 queues == 0 时，无法开启新的线程。
第 15 至 21 行：创建 EagerThreadPoolExecutor 对象。
第 22 行：设置 executor 到 taskQueue 中。因为，taskQueue 需要在队列满了时，调用 EagerThreadPoolExecutor 创建线程。

6.1 TaskQueue

com.alibaba.dubbo.common.threadpool.support.eager.TaskQueue ，继承 LinkedBlockingQueue 类，任务队列。代码如下：

Java复制public class TaskQueue<R extends Runnable> extends LinkedBlockingQueue<Runnable> {

    private static final long serialVersionUID = -2635853580887179627L;

    /**
     * 线程池
     */
    private EagerThreadPoolExecutor executor;

    public TaskQueue(int capacity) {
        super(capacity);
    }

    public void setExecutor(EagerThreadPoolExecutor exec) {
        executor = exec;
    }

    @Override
    public boolean offer(Runnable runnable) {
        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;
        }
        // 返回 true ，则添加到队列
        return super.offer(runnable);
    }

    /**
     * 重写 take 方法，当线程池停止时，抛出异常
     *
     * @return
     * @throws InterruptedException
     */
    @Override
    public Runnable take() throws InterruptedException {
        if (executor != null && executor.isShutdown()) {
            throw new RejectedExecutionException("Executor is shutdown!");
        }
        return super.take();
    }

    /**
     * 重写 poll 方法，当线程池停止时，抛出异常
     *
     * @param timeout
     * @param unit
     * @return
     * @throws InterruptedException
     */
    @Override
    public Runnable poll(long timeout, TimeUnit unit) throws InterruptedException {
        if (executor != null && executor.isShutdown()) {
            throw new RejectedExecutionException("Executor is shutdown!");
        }
        return super.poll(timeout, unit);
    }

    /**
     * 重写 poll 方法，当线程池停止时，抛出异常
     *
     * @return
     */
    @Override
    public Runnable poll() {
        if (executor != null && executor.isShutdown()) {
            throw new RejectedExecutionException("Executor is shutdown!");
        }
        return super.poll();
    }

}

executor 属性，线程池。通过 #setExecutor(executor) 方法设置。
#offer(runnable) 实现方法，代码如下：
- 第 22 至 24 行：若 executor 未设置，抛出 RejectedExecutionException 异常。
- 第 26 行：调用 Executor#getPoolSize() 方法，获得当前线程数。
- 第 28 至 30 行：调用 EagerThreadPoolExecutor#getSubmittedTaskCount() 方法，获得已提交任务数。若已提交任务数小于当前线程数，意味着当前有空闲线程，可以提交任务到队列中。
- 第 32 至 34 行：若当前线程数小于最大线程数，返回 false ，欺骗线程池创建新的线程。
- 第 36 行：返回 true ，提交任务到队列中。
其它方法，主要是在线程池停止时，抛出 RejectedExecutionException 异常。

6.2 EagerThreadPoolExecutor

com.alibaba.dubbo.common.threadpool.support.eager.EagerThreadPoolExecutor ，继承 ThreadPoolExecutor 类，积极线程池执行器，实现优先创建线程的特性。代码如下：

Java复制public class EagerThreadPoolExecutor extends ThreadPoolExecutor {

    /**
     * 已提交任务数量
     */
    private final AtomicInteger submittedTaskCount = new AtomicInteger(0);

    public EagerThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit, BlockingQueue<Runnable> workQueue,
                                  ThreadFactory threadFactory,
                                  RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }

    /**
     * 获取已提交任务数量
     *
     * @return 数量
     */
    public int getSubmittedTaskCount() {
        return submittedTaskCount.get();
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        submittedTaskCount.decrementAndGet();
    }

    @Override
    public void execute(Runnable command) {
        if (command == null) {
            throw new NullPointerException();
        }
        // 提交任务数 + 1
        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) {
            // 提交任务数 - 1
            submittedTaskCount.decrementAndGet();
            throw t;
        }
    }

}

submittedTaskCount 属性，已提交任务数量。
#getSubmittedTaskCount() 方法，获得已提交任务数量。
#afterExecute(r, t) 方法，执行任务完成时，调用 submittedTaskCount.decrementAndGet() ，提交任务数 - 1 。
#execute(command) 方法，执行任务。
- 第 34 行：提交任务数 + 1 。
- 第 37 行：调用父类的 #execute(command) 方法，执行任务。
- 【发生拒绝异常】第 39 至 48 行：发生拒绝异常时，调用 TaskQueue#retryOffer(command, 0, TimeUnit.MILLISECONDS) 方法，尝试重新将任务放到队列中。若提交失败，提交任务数 - 1 ，并抛出 RejectedExecutionException 异常。
- 【发生其他异常】第 49 至 52 行：发生其他异常时，提交任务数 - 1 ，并抛出异常。

7. AbortPolicyWithReport

com.alibaba.dubbo.common.threadpool.support.AbortPolicyWithReport ，实现 RejectedExecutionHandler 接口，拒绝策略实现类。打印 JStack ，分析线程状态。代码如下：

Java复制  1: public class AbortPolicyWithReport extends AbstractRejectedExecutionHandler {
  2: 
  3:     protected static final Logger logger = LoggerFactory.getLogger(AbortPolicyWithReport.class);
  4: 
  5:     /**
  6:      * 线程名
  7:      */
  8:     private final String threadName;
  9: 
 10:     /**
 11:      * URL 对象
 12:      */
 13:     private final URL url;
 14: 
 15:     /**
 16:      * 最后打印时间
 17:      */
 18:     private static volatile long lastPrintTime = 0;
 19: 
 20:     /**
 21:      * 信号量，用于保证打印日志串行
 22:      */
 23:     private static Semaphore guard = new Semaphore(1);
 24: 
 25:     public AbortPolicyWithReport(String threadName, URL url) {
 26:         this.threadName = threadName;
 27:         this.url = url;
 28:     }
 29: 
 30:     @Override
 31:     public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
 32:         // 打印告警日志
 33:         String msg = String.format("Thread pool is EXHAUSTED!" +
 34:                         " Thread Name: %s, Pool Size: %d (active: %d, core: %d, max: %d, largest: %d), Task: %d (completed: %d)," +
 35:                         " Executor status:(isShutdown:%s, isTerminated:%s, isTerminating:%s), in %s://%s:%d!",
 36:                 threadName, e.getPoolSize(), e.getActiveCount(), e.getCorePoolSize(), e.getMaximumPoolSize(), e.getLargestPoolSize(),
 37:                 e.getTaskCount(), e.getCompletedTaskCount(), e.isShutdown(), e.isTerminated(), e.isTerminating(),
 38:                 url.getProtocol(), url.getIp(), url.getPort());
 39:         logger.warn(msg);
 40:         // 打印 JStack ，分析线程状态
 41:         dumpJStack();
 42:         // 抛出 RejectedExecutionException 异常
 43:         throw new RejectedExecutionException(msg);
 44:     }
 45: 
 46:     /**
 47:      * 打印 JStack
 48:      */
 49:     private void dumpJStack() {
 50:         long now = System.currentTimeMillis();
 51: 
 52:         // 每 10 分钟，打印一次
 53:         // dump every 10 minutes
 54:         if (now - lastPrintTime < 10 * 60 * 1000) {
 55:             return;
 56:         }
 57: 
 58:         // 获得信号量
 59:         if (!guard.tryAcquire()) {
 60:             return;
 61:         }
 62: 
 63:         // 创建线程池，后台执行打印 JStack
 64:         ExecutorService pool = Executors.newSingleThreadExecutor();
 65:         pool.execute(new Runnable() {
 66:             @Override
 67:             public void run() {
 68:                 // 获得系统
 69:                 String OS = System.getProperty("os.name").toLowerCase();
 70:                 // 获得线程的 dump
 71:                 String dump;
 72:                 if (OS.contains("win")) {
 73:                     dump = "Cannot dump jstack on Windows";
 74:                 } else {
 75:                     dump = JvmUtil.jstack();
 76:                 }
 77: 
 78:                 // 打印告警日志
 79:                 try {
 80:                     String jstackLogFile = url.getParameter(Constants.DUMP_DIRECTORY, System.getProperty("user.home"));
 81:                     SimpleDateFormat sdf;
 82: 
 83:                     String OS2 = System.getProperty("os.name").toLowerCase();
 84:                     // window system don't support ":" in file name
 85:                     if(OS2.contains("win")){
 86:                         sdf = new SimpleDateFormat("yyyy-MM-dd_HH-mm-ss");
 87:                     }else {
 88:                         sdf = new SimpleDateFormat("yyyy-MM-dd_HH:mm:ss");
 89:                     }
 90: 
 91:                     String dateStr = sdf.format(new Date());
 92:                     // 写入文件
 93:                     try (FileOutputStream jstackStream = new FileOutputStream(
 94:                             jstackLogFile + "/dubbo_jstack.log" + "." + dateStr)) {
 95:                         jstackStream.write(dump.getBytes());
 96:                     }
 97:                 } catch (Throwable t) {
 98:                     logger.error("dump jstack error", t);
 99:                 } finally {
100:                     // 释放信号量
101:                     guard.release();
102:                 }
103:                 // 记录最后打印时间
104:                 lastPrintTime = System.currentTimeMillis();
105:             }
106:         });
107:         // 关闭线程池
108:         pool.shutdown();
109:     }
110: 
111: }

threadName 属性，线程名。
url 属性，URL 对象。
lastPrintTime 属性，最后打印时间。
guard 属性，信号量，用于保证打印日志串行。
#rejectedExecution(r, e) 实现方法：
- 第 32 至 39 行：打印告警日志。例如：
```
code复制[14/04/18 11:10:23:023 UTC] ThreadPoolExecutor is EXHAUSTED!
Thread Name: DubboServerHandler-172.16.196.128:20880, Pool Size: 200 (active: 200, core: 200, max: 200, largest: 200), Task: 295 (completed: 95),
Executor status:(isShutdown:false, isTerminated:false, isTerminating:false)!
```
  - 通过日志，我们可以看到，目前线程池已经满了，并且都处于活跃状态。
- 第 41 行：调用 #dumpJStack() 方法，打印 JStack ，分析线程状态。
- 第 43 行：抛出 RejectedExecutionException 异常。
#dumpJStack() 方法，打印 JStack：
- 第 53 至 56 行：每 10 分钟，打印一次。
- 第 58 至 61 行：获得信号量。若获取不到，不执行打印。
- 第 64 行：创建线程池，后台执行打印 JStack 。因为是后台打印，使用单线程即可。
- 第 65 至 106 行：调用 JvmUtil#jstack() 方法，获得线程的 dump 。
  - 第 80 行：获得 URL.dump.directory 配置，日志目录。若未配置，使用 user.home 。
  - 第 81 至 91 行：根据系统，创建不同的 SimpleDateFormat 对象。因为 Windows 系统下，文件名不支持 ":" 。
  - 第 93 至 96 行：写入文件。
  - 第 101 行：释放信号量。
  - 第 104 行：记录最后打印时间。
- 第 108 行：关闭线程池。

8. JvmUtil

com.alibaba.dubbo.common.utils.JvmUtil ，JVM 工具类。目前，仅有 JStack 功能，整体和《Dubbo 用户指南 —— 线程栈自动 dump》一致，胖友可以看看。

9. 日志

在《Dubbo 用户指南 —— 线程栈自动 dump》中，提到日志目录如下：

Linux/Mac OS: ~/logs/dubbo-jstack.log
Windows: C:\home\logs\dubbo-jstack.log

666. 彩蛋

知识星球

美滋滋。