1. 项目概述ChatGLM2-6B是清华大学知识工程组KEG与智谱AI联合研发的开源双语对话大模型作为ChatGLM-6B的第二代版本它在多个方面都有显著提升。这个62亿参数规模的模型在保持高效推理能力的同时大幅提升了中英文对话质量特别适合个人开发者和中小企业进行本地化部署。提示虽然ChatGLM2-6B对硬件要求相对友好但想要流畅运行仍需# 1. 概述本文分享Dubbo 的线程池策略。在 《精尽 Dubbo 源码分析 —— 线程池》 一文中我们已经分享了四种线程池策略fixed固定大小线程池启动时建立线程不关闭一直持有。cached缓存线程池空闲一分钟自动删除需要时重建。limited可伸缩线程池但池中的线程数只会增长不会收缩。只增长不收缩的目的是为了避免收缩时突然来了大流量引起的性能问题。eager优先创建Worker线程池。在任务数大于corePoolSize但是小于maximumPoolSize时优先创建Worker来处理任务。当任务数大于maximumPoolSize时将任务放入阻塞队列中。阻塞队列充满时抛出RejectedExecutionException。(相比于cached:cached在任务数量超过maximumPoolSize时直接抛出异常而不是将任务放入阻塞队列)本文分享的Dubbo 的线程池策略是在上述四种的基础上添加拒绝策略所以胖友在阅读本文之前请先确保理解了这四种线程池策略的实现。2. ThreadPoolcom.alibaba.dubbo.common.threadpool.ThreadPool线程池接口。代码如下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 url)方法获得对应的线程池的执行器。3. FixedThreadPoolcom.alibaba.dubbo.common.threadpool.support.fixed.FixedThreadPool实现 ThreadPool 接口固定大小线程池启动时建立线程不关闭一直持有。代码如下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 SynchronousQueueRunnable() : (queues 0 ? new LinkedBlockingQueueRunnable() : new LinkedBlockingQueueRunnable(queues)), new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url)); } }第 7 行线程名。优先从URL.threadname参数获得如果为空默认为Dubbo。第 9 行线程数。优先从URL.threads参数获得如果为空默认为200。第 11 行队列数。优先从URL.queues参数获得。queues 0 SynchronousQueue 对象。queues 0 LinkedBlockingQueue 对象队列大小为Integer.MAX_VALUE。queues 0 LinkedBlockingQueue 对象队列大小为queues。第 12 至 15 行创建线程池执行器。corePoolSize和maximumPoolSize参数使用threads。keepAliveTime参数0 毫秒。unit参数TimeUnit.MILLISECONDS 。workQueue参数根据queues来生成对应的阻塞队列。threadFactory参数创建 NamedThreadFactory 对象用于生成线程名。handler参数创建 AbortPolicyWithReport 对象用于当任务添加到线程池中被拒绝时。4. CachedThreadPoolcom.alibaba.dubbo.common.threadpool.support.cached.CachedThreadPool实现 ThreadPool 接口缓存线程池空闲一定时长自动删除需要时重建。代码如下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 SynchronousQueueRunnable() : (queues 0 ? new LinkedBlockingQueueRunnable() : new LinkedBlockingQueueRunnable(queues)), new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url)); } }第 7 行线程名。优先从URL.threadname参数获得如果为空默认为Dubbo。第 9 行核心线程数。优先从URL.corethreads参数获得如果为空默认为0。第 11 行最大线程数。优先从URL.threads参数获得如果为空默认为Integer.MAX_VALUE。第 13 行队列数。优先从URL.queues参数获得如果为空默认为0。第 15 行线程空闲时长。优先从URL.alive参数获得如果为空默认为60 * 1000毫秒。第 16 至 20 行创建线程池执行器。corePoolSize参数使用cores。maximumPoolSize参数使用threads。keepAliveTime参数alive毫秒。unit参数TimeUnit.MILLISECONDS 。workQueue参数根据queues来生成对应的阻塞队列。threadFactory参数创建 NamedThreadFactory 对象用于生成线程名。handler参数创建 AbortPolicyWithReport 对象用于当任务添加到线程池中被拒绝时。5. LimitedThreadPoolcom.alibaba.dubbo.common.threadpool.support.limited.LimitedThreadPool实现 ThreadPool 接口可伸缩线程池但池中的线程数只会增长不会收缩。只增长不收缩的目的是为了避免收缩时突然来了大流量引起的性能问题。代码如下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 SynchronousQueueRunnable() : (queues 0 ? new LinkedBlockingQueueRunnable() : new LinkedBlockingQueueRunnable(queues)), new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url)); } }第 7 行线程名。优先从URL.threadname参数获得如果为空默认为Dubbo。第 9 行核心线程数。优先从URL.corethreads参数获得如果为空默认为0。第 11 行最大线程数。优先从URL.threads参数获得如果为空默认为200。第 13 行队列数。优先从URL.queues参数获得如果为空默认为0。第 14 至 18 行创建线程池执行器。corePoolSize参数使用cores。maximumPoolSize参数使用threads。keepAliveTime参数Long.MAX_VALUE毫秒。unit参数TimeUnit.MILLISECONDS 。workQueue参数根据queues来生成对应的阻塞队列。threadFactory参数创建 NamedThreadFactory 对象用于生成线程名。handler参数创建 AbortPolicyWithReport 对象用于当任务添加到线程池中被拒绝时。6. EagerThreadPoolcom.alibaba.dubbo.common.threadpool.support.eager.EagerThreadPool实现 ThreadPool 接口优先创建Worker线程池。在任务数大于corePoolSize但是小于maximumPoolSize时优先创建Worker来处理任务。当任务数大于maximumPoolSize时将任务放入阻塞队列中。阻塞队列充满时抛出RejectedExecutionException。(相比于cached:cached在任务数量超过maximumPoolSize时直接抛出异常而不是将任务放入阻塞队列)。代码如下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 TaskQueueRunnable taskQueue new TaskQueueRunnable(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; } }第 7 行线程名。优先从URL.threadname参数获得如果为空默认为Dubbo。第 9 行核心线程数。优先从URL.corethreads参数获得如果为空默认为0。第 11 行最大线程数。优先从URL.threads参数获得如果为空默认为Integer.MAX_VALUE。第 13 行队列数。优先从URL.queues参数获得如果为空默认为0。第 15 行线程空闲时长。优先从URL.alive参数获得如果为空默认为60 * 1000毫秒。第 17 至 25 行创建线程池执行器。corePoolSize参数使用cores。maximumPoolSize参数使用threads。keepAliveTime参数alive毫秒。unit参数TimeUnit.MILLISECONDS 。workQueue参数创建 TaskQueue 对象队列大小为queues。threadFactory参数创建 NamedThreadFactory 对象用于生成线程名。handler参数创建 AbortPolicyWithReport 对象用于当任务添加到线程池中被拒绝时。第 26 行调用TaskQueue#setExecutor(ExecutorService executor)方法设置执行器到队列中。因为TaskQueue 需要在队列满了时添加新的线程。6.1 TaskQueuecom.alibaba.dubbo.common.threadpool.support.eager.TaskQueue实现java.util.concurrent.LinkedBlockingQueue类任务队列。代码如下public class TaskQueueR extends Runnable extends LinkedBlockingQueueRunnable { 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); } // 当前线程数小于最大线程数创建新线程 if (currentPoolThreadSize executor.getMaximumPoolSize()) { return false; } // 提交到队列 return super.offer(runnable); } }executor属性执行器。通过#setExecutor(ExecutorService executor)方法设置。#offer(Runnable runnable)方法第 24 行若executor未设置抛出 RejectedExecutionException 异常。第 27 行调用Executor#getPoolSize()方法获得当前线程数。第 29 至 31 行调用EagerThreadPoolExecutor#getSubmittedTaskCount()方法获得已提交任务数。若已提交任务数小于当前线程数提交任务到队列。即有线程空闲可以提交到队列中。第 33 至 35 行当前线程数小于最大线程数返回false促使EagerThreadPoolExecutor 创建新线程。第 37 行提交到队列。6.2 EagerThreadPoolExecutorcom.alibaba.dubbo.common.threadpool.support.eager.EagerThreadPoolExecutor实现 ThreadPoolExecutor 类积极线程池执行器。代码如下public class EagerThreadPoolExecutor extends ThreadPoolExecutor { /** * 已提交任务数量 */ private final AtomicInteger submittedTaskCount new AtomicInteger(0); public EagerThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, TaskQueueRunnable 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属性已提交任务数量。这个属性是 EagerThreadPoolExecutor 最关键的属性。#execute(Runnable command)方法第 33 行提交任务数 1 。第 36 行调用父类的#execute(Runnable command)方法提交任务执行。【发生拒绝异常】第 37 至 47 行发生拒绝异常调用TaskQueue#retryOffer(Runnable o, long timeout, TimeUnit unit)方法尝试重新将任务放到队列中。若提交失败提交任务数 - 1 并抛出 RejectedExecutionException 异常。【发生其他异常】第 48 至 51 行提交任务数 - 1 并抛出该异常。#afterExecute(Runnable r, Throwable t)方法第 25 行任务执行完成提交任务数 - 1 。7. AbortPolicyWithReportcom.alibaba.dubbo.common.threadpool.support.AbortPolicyWithReport实现java.util.concurrent.ThreadPoolExecutor.AbortPolicy拒绝策略实现类。打印 JStack 分析线程状态。代码如下1: public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy { 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: * 信号量大小为 1 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: // 获得当前时间 51: long now System.currentTimeMillis(); 52: 53: // 每 10 分钟打印一次 54: // Reset every one hour to avoid flood. 55: if (now - lastPrintTime 10 * 60 * 1000) { 56: return; 57: } 58: 59: // 获得信号量 60: if (!guard.tryAcquire()) { 61: return; 62: } 63: 64: // 创建线程池后台执行打印 JStack 65: ExecutorService pool Executors.newSingleThreadExecutor(); 66: pool.execute(new Runnable() { 67: Override 68: public void run() { 69: // 获得系统 70: String OS System.getProperty(os.name).toLowerCase(); 71: // 获得系统进程集合 72: String dumpCommand null; 73: if (OS.contains(sunos) || OS.contains(linux)) { 74: dumpCommand kill -3 getPID(); 75: } else if (OS.contains(win)) { 76: dumpCommand tasklist; 77: } 78: 79: // 打印 JStack 80: if (dumpCommand ! null) { 81: try { 82: Process process Runtime.getRuntime().exec(dumpCommand); 83: BufferedReader reader new BufferedReader(new InputStreamReader(process.getInputStream())); 84: String line; 85: try { 86: while ((line reader.readLine()) ! null) { 87: logger.error(line); 88: } 89: } finally { 90: reader.close(); 91: } 92: } catch (Throwable t) { 93: logger.error(t.getMessage(), t); 94: } 95: } 96: 97: // 打印线程和线程池状态 98: try { 99: MapThread, StackTraceElement[] allThreads Thread.getAllStackTraces(); 100: if (allThreads ! null !allThreads.isEmpty()) { 101: for (Map.EntryThread, StackTraceElement[] entry : allThreads.entrySet()) { 102: Thread thread entry.getKey(); 103: // 跳过当前线程 104: if (thread.getId() Thread.currentThread().getId()) { 105: continue; 106: } 107: // 打印线程 108: logger.error(thread: thread.getName() , state: thread.getState()); 109: for (StackTraceElement stackTraceElement : entry.getValue()) { 110: logger.error( stackTraceElement.toString()); 111: } 112: } 113: } 114: } catch (Throwable t) { 115: logger.error(t.getMessage(), t); 116: } 117: 118: // 释放信号量 119: guard.release(); 120: // 记录最后打印时间 121: lastPrintTime System.currentTimeMillis(); 122: } 123: }); 124: // 关闭线程池 125: // must shutdown thread pool ,if not will lead to OOM 126: pool.shutdown(); 127: 128: } 129: 130: /** 131: * 获得当前进程的 PID 132: * 133: * return PID 134: */ 135: private int getPID() { 136: String processName java.lang.management.ManagementFactory.getRuntimeMXBean().getName(); 137: if (processName.contains()) { 138: return Integer.parseInt(processName.substring(0, processName.indexOf())); 139: } else { 140: return 0; 141: } 142: } 143: 144: }#rejectedExecution(Runnable r, ThreadPoolExecutor e)实现方法第 32 至 39 行打印告警日志。例如[14/04/19 03:40:05:005 CST] main WARN support.AbortPolicyWithReport: [DUBBO] Thread pool is EXHAUSTED! Thread Name: DubboServerHandler-172.16.132.166:20880, Pool Size: 200 (active: 200, core: 200, max: 200, largest: 200), Task: 295 (completed: 95), Executor status:(isShutdown:false, isTerminated:false, isTerminating:false), in dubbo://172.16.132.166:20880!通过日志我们可以看到目前线程池的状态包括线程池大小、活跃线程数、任务总数、完成数量等等。第 41 行调用#dumpJStack()方法打印JStack分析线程状态。第 43 行抛出 RejectedExecutionException 异常。#dumpJStack()方法第 55 至 57 行每 10 分钟打印一次。第 60 至 62 行获得信号量。保证同一时间有且仅有一个线程执行打印。第 65 行创建单线程的线程池后台执行打印 JStack 。第 66 至 123 行调用Runtime#exec(command)方法打印 JStack 日志。注意Windows 和 Linux 的日志格式不同。第 97 至 116 行打印线程和线程池状态。第 119 行释放信号量。第 121 行记录最后打印时间。第 126 行关闭线程池。避免内存泄露。#getPID()方法获得当前进程的 PID 。666. 彩蛋 又是一篇炒冷饭的文章。 Dubbo线程池策略与拒绝策略详解
1. 项目概述ChatGLM2-6B是清华大学知识工程组KEG与智谱AI联合研发的开源双语对话大模型作为ChatGLM-6B的第二代版本它在多个方面都有显著提升。这个62亿参数规模的模型在保持高效推理能力的同时大幅提升了中英文对话质量特别适合个人开发者和中小企业进行本地化部署。提示虽然ChatGLM2-6B对硬件要求相对友好但想要流畅运行仍需# 1. 概述本文分享Dubbo 的线程池策略。在 《精尽 Dubbo 源码分析 —— 线程池》 一文中我们已经分享了四种线程池策略fixed固定大小线程池启动时建立线程不关闭一直持有。cached缓存线程池空闲一分钟自动删除需要时重建。limited可伸缩线程池但池中的线程数只会增长不会收缩。只增长不收缩的目的是为了避免收缩时突然来了大流量引起的性能问题。eager优先创建Worker线程池。在任务数大于corePoolSize但是小于maximumPoolSize时优先创建Worker来处理任务。当任务数大于maximumPoolSize时将任务放入阻塞队列中。阻塞队列充满时抛出RejectedExecutionException。(相比于cached:cached在任务数量超过maximumPoolSize时直接抛出异常而不是将任务放入阻塞队列)本文分享的Dubbo 的线程池策略是在上述四种的基础上添加拒绝策略所以胖友在阅读本文之前请先确保理解了这四种线程池策略的实现。2. ThreadPoolcom.alibaba.dubbo.common.threadpool.ThreadPool线程池接口。代码如下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 url)方法获得对应的线程池的执行器。3. FixedThreadPoolcom.alibaba.dubbo.common.threadpool.support.fixed.FixedThreadPool实现 ThreadPool 接口固定大小线程池启动时建立线程不关闭一直持有。代码如下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 SynchronousQueueRunnable() : (queues 0 ? new LinkedBlockingQueueRunnable() : new LinkedBlockingQueueRunnable(queues)), new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url)); } }第 7 行线程名。优先从URL.threadname参数获得如果为空默认为Dubbo。第 9 行线程数。优先从URL.threads参数获得如果为空默认为200。第 11 行队列数。优先从URL.queues参数获得。queues 0 SynchronousQueue 对象。queues 0 LinkedBlockingQueue 对象队列大小为Integer.MAX_VALUE。queues 0 LinkedBlockingQueue 对象队列大小为queues。第 12 至 15 行创建线程池执行器。corePoolSize和maximumPoolSize参数使用threads。keepAliveTime参数0 毫秒。unit参数TimeUnit.MILLISECONDS 。workQueue参数根据queues来生成对应的阻塞队列。threadFactory参数创建 NamedThreadFactory 对象用于生成线程名。handler参数创建 AbortPolicyWithReport 对象用于当任务添加到线程池中被拒绝时。4. CachedThreadPoolcom.alibaba.dubbo.common.threadpool.support.cached.CachedThreadPool实现 ThreadPool 接口缓存线程池空闲一定时长自动删除需要时重建。代码如下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 SynchronousQueueRunnable() : (queues 0 ? new LinkedBlockingQueueRunnable() : new LinkedBlockingQueueRunnable(queues)), new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url)); } }第 7 行线程名。优先从URL.threadname参数获得如果为空默认为Dubbo。第 9 行核心线程数。优先从URL.corethreads参数获得如果为空默认为0。第 11 行最大线程数。优先从URL.threads参数获得如果为空默认为Integer.MAX_VALUE。第 13 行队列数。优先从URL.queues参数获得如果为空默认为0。第 15 行线程空闲时长。优先从URL.alive参数获得如果为空默认为60 * 1000毫秒。第 16 至 20 行创建线程池执行器。corePoolSize参数使用cores。maximumPoolSize参数使用threads。keepAliveTime参数alive毫秒。unit参数TimeUnit.MILLISECONDS 。workQueue参数根据queues来生成对应的阻塞队列。threadFactory参数创建 NamedThreadFactory 对象用于生成线程名。handler参数创建 AbortPolicyWithReport 对象用于当任务添加到线程池中被拒绝时。5. LimitedThreadPoolcom.alibaba.dubbo.common.threadpool.support.limited.LimitedThreadPool实现 ThreadPool 接口可伸缩线程池但池中的线程数只会增长不会收缩。只增长不收缩的目的是为了避免收缩时突然来了大流量引起的性能问题。代码如下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 SynchronousQueueRunnable() : (queues 0 ? new LinkedBlockingQueueRunnable() : new LinkedBlockingQueueRunnable(queues)), new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url)); } }第 7 行线程名。优先从URL.threadname参数获得如果为空默认为Dubbo。第 9 行核心线程数。优先从URL.corethreads参数获得如果为空默认为0。第 11 行最大线程数。优先从URL.threads参数获得如果为空默认为200。第 13 行队列数。优先从URL.queues参数获得如果为空默认为0。第 14 至 18 行创建线程池执行器。corePoolSize参数使用cores。maximumPoolSize参数使用threads。keepAliveTime参数Long.MAX_VALUE毫秒。unit参数TimeUnit.MILLISECONDS 。workQueue参数根据queues来生成对应的阻塞队列。threadFactory参数创建 NamedThreadFactory 对象用于生成线程名。handler参数创建 AbortPolicyWithReport 对象用于当任务添加到线程池中被拒绝时。6. EagerThreadPoolcom.alibaba.dubbo.common.threadpool.support.eager.EagerThreadPool实现 ThreadPool 接口优先创建Worker线程池。在任务数大于corePoolSize但是小于maximumPoolSize时优先创建Worker来处理任务。当任务数大于maximumPoolSize时将任务放入阻塞队列中。阻塞队列充满时抛出RejectedExecutionException。(相比于cached:cached在任务数量超过maximumPoolSize时直接抛出异常而不是将任务放入阻塞队列)。代码如下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 TaskQueueRunnable taskQueue new TaskQueueRunnable(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; } }第 7 行线程名。优先从URL.threadname参数获得如果为空默认为Dubbo。第 9 行核心线程数。优先从URL.corethreads参数获得如果为空默认为0。第 11 行最大线程数。优先从URL.threads参数获得如果为空默认为Integer.MAX_VALUE。第 13 行队列数。优先从URL.queues参数获得如果为空默认为0。第 15 行线程空闲时长。优先从URL.alive参数获得如果为空默认为60 * 1000毫秒。第 17 至 25 行创建线程池执行器。corePoolSize参数使用cores。maximumPoolSize参数使用threads。keepAliveTime参数alive毫秒。unit参数TimeUnit.MILLISECONDS 。workQueue参数创建 TaskQueue 对象队列大小为queues。threadFactory参数创建 NamedThreadFactory 对象用于生成线程名。handler参数创建 AbortPolicyWithReport 对象用于当任务添加到线程池中被拒绝时。第 26 行调用TaskQueue#setExecutor(ExecutorService executor)方法设置执行器到队列中。因为TaskQueue 需要在队列满了时添加新的线程。6.1 TaskQueuecom.alibaba.dubbo.common.threadpool.support.eager.TaskQueue实现java.util.concurrent.LinkedBlockingQueue类任务队列。代码如下public class TaskQueueR extends Runnable extends LinkedBlockingQueueRunnable { 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); } // 当前线程数小于最大线程数创建新线程 if (currentPoolThreadSize executor.getMaximumPoolSize()) { return false; } // 提交到队列 return super.offer(runnable); } }executor属性执行器。通过#setExecutor(ExecutorService executor)方法设置。#offer(Runnable runnable)方法第 24 行若executor未设置抛出 RejectedExecutionException 异常。第 27 行调用Executor#getPoolSize()方法获得当前线程数。第 29 至 31 行调用EagerThreadPoolExecutor#getSubmittedTaskCount()方法获得已提交任务数。若已提交任务数小于当前线程数提交任务到队列。即有线程空闲可以提交到队列中。第 33 至 35 行当前线程数小于最大线程数返回false促使EagerThreadPoolExecutor 创建新线程。第 37 行提交到队列。6.2 EagerThreadPoolExecutorcom.alibaba.dubbo.common.threadpool.support.eager.EagerThreadPoolExecutor实现 ThreadPoolExecutor 类积极线程池执行器。代码如下public class EagerThreadPoolExecutor extends ThreadPoolExecutor { /** * 已提交任务数量 */ private final AtomicInteger submittedTaskCount new AtomicInteger(0); public EagerThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, TaskQueueRunnable 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属性已提交任务数量。这个属性是 EagerThreadPoolExecutor 最关键的属性。#execute(Runnable command)方法第 33 行提交任务数 1 。第 36 行调用父类的#execute(Runnable command)方法提交任务执行。【发生拒绝异常】第 37 至 47 行发生拒绝异常调用TaskQueue#retryOffer(Runnable o, long timeout, TimeUnit unit)方法尝试重新将任务放到队列中。若提交失败提交任务数 - 1 并抛出 RejectedExecutionException 异常。【发生其他异常】第 48 至 51 行提交任务数 - 1 并抛出该异常。#afterExecute(Runnable r, Throwable t)方法第 25 行任务执行完成提交任务数 - 1 。7. AbortPolicyWithReportcom.alibaba.dubbo.common.threadpool.support.AbortPolicyWithReport实现java.util.concurrent.ThreadPoolExecutor.AbortPolicy拒绝策略实现类。打印 JStack 分析线程状态。代码如下1: public class AbortPolicyWithReport extends ThreadPoolExecutor.AbortPolicy { 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: * 信号量大小为 1 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: // 获得当前时间 51: long now System.currentTimeMillis(); 52: 53: // 每 10 分钟打印一次 54: // Reset every one hour to avoid flood. 55: if (now - lastPrintTime 10 * 60 * 1000) { 56: return; 57: } 58: 59: // 获得信号量 60: if (!guard.tryAcquire()) { 61: return; 62: } 63: 64: // 创建线程池后台执行打印 JStack 65: ExecutorService pool Executors.newSingleThreadExecutor(); 66: pool.execute(new Runnable() { 67: Override 68: public void run() { 69: // 获得系统 70: String OS System.getProperty(os.name).toLowerCase(); 71: // 获得系统进程集合 72: String dumpCommand null; 73: if (OS.contains(sunos) || OS.contains(linux)) { 74: dumpCommand kill -3 getPID(); 75: } else if (OS.contains(win)) { 76: dumpCommand tasklist; 77: } 78: 79: // 打印 JStack 80: if (dumpCommand ! null) { 81: try { 82: Process process Runtime.getRuntime().exec(dumpCommand); 83: BufferedReader reader new BufferedReader(new InputStreamReader(process.getInputStream())); 84: String line; 85: try { 86: while ((line reader.readLine()) ! null) { 87: logger.error(line); 88: } 89: } finally { 90: reader.close(); 91: } 92: } catch (Throwable t) { 93: logger.error(t.getMessage(), t); 94: } 95: } 96: 97: // 打印线程和线程池状态 98: try { 99: MapThread, StackTraceElement[] allThreads Thread.getAllStackTraces(); 100: if (allThreads ! null !allThreads.isEmpty()) { 101: for (Map.EntryThread, StackTraceElement[] entry : allThreads.entrySet()) { 102: Thread thread entry.getKey(); 103: // 跳过当前线程 104: if (thread.getId() Thread.currentThread().getId()) { 105: continue; 106: } 107: // 打印线程 108: logger.error(thread: thread.getName() , state: thread.getState()); 109: for (StackTraceElement stackTraceElement : entry.getValue()) { 110: logger.error( stackTraceElement.toString()); 111: } 112: } 113: } 114: } catch (Throwable t) { 115: logger.error(t.getMessage(), t); 116: } 117: 118: // 释放信号量 119: guard.release(); 120: // 记录最后打印时间 121: lastPrintTime System.currentTimeMillis(); 122: } 123: }); 124: // 关闭线程池 125: // must shutdown thread pool ,if not will lead to OOM 126: pool.shutdown(); 127: 128: } 129: 130: /** 131: * 获得当前进程的 PID 132: * 133: * return PID 134: */ 135: private int getPID() { 136: String processName java.lang.management.ManagementFactory.getRuntimeMXBean().getName(); 137: if (processName.contains()) { 138: return Integer.parseInt(processName.substring(0, processName.indexOf())); 139: } else { 140: return 0; 141: } 142: } 143: 144: }#rejectedExecution(Runnable r, ThreadPoolExecutor e)实现方法第 32 至 39 行打印告警日志。例如[14/04/19 03:40:05:005 CST] main WARN support.AbortPolicyWithReport: [DUBBO] Thread pool is EXHAUSTED! Thread Name: DubboServerHandler-172.16.132.166:20880, Pool Size: 200 (active: 200, core: 200, max: 200, largest: 200), Task: 295 (completed: 95), Executor status:(isShutdown:false, isTerminated:false, isTerminating:false), in dubbo://172.16.132.166:20880!通过日志我们可以看到目前线程池的状态包括线程池大小、活跃线程数、任务总数、完成数量等等。第 41 行调用#dumpJStack()方法打印JStack分析线程状态。第 43 行抛出 RejectedExecutionException 异常。#dumpJStack()方法第 55 至 57 行每 10 分钟打印一次。第 60 至 62 行获得信号量。保证同一时间有且仅有一个线程执行打印。第 65 行创建单线程的线程池后台执行打印 JStack 。第 66 至 123 行调用Runtime#exec(command)方法打印 JStack 日志。注意Windows 和 Linux 的日志格式不同。第 97 至 116 行打印线程和线程池状态。第 119 行释放信号量。第 121 行记录最后打印时间。第 126 行关闭线程池。避免内存泄露。#getPID()方法获得当前进程的 PID 。666. 彩蛋 又是一篇炒冷饭的文章。