From 57362ffcbcb58ec99d9582a3df431c8b3b263be5 Mon Sep 17 00:00:00 2001 From: "calvin.xiao" Date: Mon, 7 Dec 2015 21:39:29 +0800 Subject: [PATCH] fix thread pools bug and add one more builder --- .../modules/utils/ThreadPoolBuilder.java | 84 +++++++++++++++++-- 1 file changed, 77 insertions(+), 7 deletions(-) diff --git a/modules/utils/src/main/java/org/springside/modules/utils/ThreadPoolBuilder.java b/modules/utils/src/main/java/org/springside/modules/utils/ThreadPoolBuilder.java index 434399fa..1a33a48a 100644 --- a/modules/utils/src/main/java/org/springside/modules/utils/ThreadPoolBuilder.java +++ b/modules/utils/src/main/java/org/springside/modules/utils/ThreadPoolBuilder.java @@ -36,7 +36,7 @@ public class ThreadPoolBuilder { * * 2. 第poolSize次任务提交后, 新增任务放入Queue中, Pool中的所有线程从Queue中take任务执行. * - * 2.a Queue默认为无限长的LinkedBlockingQueue， 也可以设置queueSize换成有界的队列. + * 2.a Queue默认为无限长的LinkedBlockingQueue, 也可以设置queueSize换成有界的队列. * * 2.b 如果使用有界队列, 当队列满了之后,会调用RejectHandler进行处理, 默认为AbortPolicy，抛出RejectedExecutionException异常. * 其他可选的Policy包括静默放弃当前任务(Discard)，放弃Queue里最老的任务(DisacardOldest)，或由主线程来直接执行(CallerRuns). @@ -80,7 +80,7 @@ public class ThreadPoolBuilder { if (queueSize == 0) { queue = new LinkedBlockingQueue(); } else { - new ArrayBlockingQueue(queueSize); + queue = new ArrayBlockingQueue(queueSize); } if (threadFactory == null) { @@ -99,18 +99,18 @@ public class ThreadPoolBuilder { /** * 创建CachedThreadPool. * - * 1. 任务提交时, 如果线程数还没达到minSize即创建新线程并绑定任务(即minSize次提交后线程总数必达到minSize，不会重用之前的线程) + * 1. 任务提交时, 如果线程数还没达到minSize即创建新线程并绑定任务(即minSize次提交后线程总数必达到minSize, 不会重用之前的线程) * - * 1.a minSize默认为0, 可进行设置保证有基本的线程处理请求. + * 1.a minSize默认为0, 可设置保证有基本的线程处理请求不被回收. * * 2. 第minSize次任务提交后, 新增任务提交进SynchronousQueue后，如果没有空闲线程立刻处理，则会创建新的线程, 直到总线程数达到上限. * - * 2.a maxSize默认为Integer.Max，可进行设置. + * 2.a maxSize默认为Integer.Max, 可进行设置. * - * 2.b 如果设置了maxSize, 当总线程数达到上限, 会调用RejectHandler进行处理, 默认为AbortPolicy，抛出RejectedExecutionException异常. + * 2.b 如果设置了maxSize, 当总线程数达到上限, 会调用RejectHandler进行处理, 默认为AbortPolicy, 抛出RejectedExecutionException异常. * 其他可选的Policy包括静默放弃当前任务(Discard)，或由主线程来直接执行(CallerRuns). * - * 3. minSize以上，maxSize以下的线程，如果在keepAliveTime中都poll不到任务执行将会被结束掉, keeAliveTime默认为60秒，可设置. + * 3. minSize以上, maxSize以下的线程, 如果在keepAliveTime中都poll不到任务执行将会被结束掉, keeAliveTime默认为60秒, 可设置. */ public static class CachedThreadPoolBuilder { @@ -160,4 +160,74 @@ public class ThreadPoolBuilder { new SynchronousQueue(), threadFactory, rejectHandler); } } + + /** + * 可同时设置min/max/queue Size的线程池, 仅用于特殊场景. + * + * 比如并发要求非常高，觉得SynchronousQueue的性能太差. + * + * 比如平常使用Core线程工作，如果满了先放queue，queue满再开临时线程，此时queue的长度一定要按项目需求设好. + */ + public static class ConfigurableThreadPoolBuilder { + + private int minSize = 0; + private int maxSize = Integer.MAX_VALUE; + private int queueSize = 0; + private int keepAliveSecs = 60; + + private ThreadFactory threadFactory = null; + private RejectedExecutionHandler rejectHandler; + + public ConfigurableThreadPoolBuilder setMinSize(int minSize) { + this.minSize = minSize; + return this; + } + + public ConfigurableThreadPoolBuilder setMaxSize(int maxSize) { + this.maxSize = maxSize; + return this; + } + + public ConfigurableThreadPoolBuilder setQueueSize(int queueSize) { + this.queueSize = queueSize; + return this; + } + + public ConfigurableThreadPoolBuilder setKeepAliveSecs(int keepAliveSecs) { + this.keepAliveSecs = keepAliveSecs; + return this; + } + + public ConfigurableThreadPoolBuilder setThreadFactory(ThreadFactory threadFactory) { + this.threadFactory = threadFactory; + return this; + } + + public ConfigurableThreadPoolBuilder setRejectHanlder(RejectedExecutionHandler rejectHandler) { + this.rejectHandler = rejectHandler; + return this; + } + + public ExecutorService build() { + + BlockingQueue queue = null; + if (queueSize == 0) { + queue = new LinkedBlockingQueue(); + } else { + queue = new ArrayBlockingQueue(queueSize); + } + + if (threadFactory == null) { + threadFactory = Executors.defaultThreadFactory(); + } + + if (rejectHandler == null) { + rejectHandler = defaultRejectHandler; + } + + return new ThreadPoolExecutor(minSize, maxSize, keepAliveSecs, TimeUnit.SECONDS, queue, threadFactory, + rejectHandler); + } + } + }