- 浏览: 932939 次
- 性别:
- 来自: 北京
文章分类
- 全部博客 (251)
- WebService (17)
- IBatis (22)
- Hibernate (1)
- SpringMVC - 基础篇 (32)
- Spring (15)
- Java (11)
- JVM及调优 - 基础篇 (4)
- 集群 (14)
- 数据库 (17)
- WebSphere (5)
- 多线程 (4)
- 集合、容器 (2)
- DB Pool (1)
- Power Designer (5)
- Maven基础 (5)
- JS (14)
- WEB 前端 (5)
- 实用小工具 (17)
- 社会、人 (2)
- 乱七八糟 (18)
- ASM&CGLIB - 基础篇 (12)
- 缓存 (1)
- 性能 (1)
- 设计之殇 (1)
- 分布式事务 (1)
- 单点登录 (11)
- 分布式 Session (4)
- Memcached - 基础篇 (6)
最新评论
-
一笑_奈何:
楼主写的还真行不错。
扫盲贴 - J2EE集群之JNDI集群实现 -
xuezhongyu01:
博主写的很详细,但最后还是没明白,最后调用BasicDataS ...
Spring中的destroy-method方法 -
Mr梁:
commons-fileupload.jar commons- ...
SpringMVC 中文件上传 MultipartResolver -
Eywa:
总结的很不错
ORACLE CASE WHEN 及 SELECT CASE WHEN的用法 -
TryRelax:
fastjson 比 jackson 好用吧?
Spring MVC Jackson DateFormat
java.util.concurrent.ThreadPoolExecutor相关基础介绍和使用示例。
[ 一 ]、常用线程池
最常用构造方法为:
- ThreadPoolExecutor( int corePoolSize,
- int maximumPoolSize,
- long keepAliveTime,
- TimeUnit unit,
- BlockingQueue<Runnable> workQueue,
- RejectedExecutionHandler handler)
ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler)
JDK自带的配置好的线程池:
- // 固定工作线程数量的线程池
- ExecutorService executorService1 = Executors.newFixedThreadPool( 3 );
- // 一个可缓存的线程池
- ExecutorService executorService2 = Executors.newCachedThreadPool();
- // 单线程化的Executor
- ExecutorService executorService3 = Executors.newSingleThreadExecutor();
- // 支持定时的以及周期性的任务执行
- ExecutorService executorService4 = Executors.newScheduledThreadPool( 3 );
// 固定工作线程数量的线程池 ExecutorService executorService1 = Executors.newFixedThreadPool(3); // 一个可缓存的线程池 ExecutorService executorService2 = Executors.newCachedThreadPool(); // 单线程化的Executor ExecutorService executorService3 = Executors.newSingleThreadExecutor(); // 支持定时的以及周期性的任务执行 ExecutorService executorService4 = Executors.newScheduledThreadPool(3);
这些预定义好的线程池服务也是基于ThreadPoolExecutor配置的,所以我们应该从最基本的参数着手了解
,如下:
参数详细说明
:
[ 1 ]、corePoolSize
: 线程池维护线程的最少数量
[ 2 ]、maximumPoolSize
:线程池维护线程的最大数量
[ 3 ]、keepAliveTime
: 线程池维护线程所允许的空闲时间
[ 4 ]、unit
:
线程池维护线程所允许的空闲时间的单位,unit可选的参数为java.util.concurrent.TimeUnit中的几个静态属性:
- NANOSECONDS
- MICROSECONDS
- MILLISECONDS
- SECONDS
[ 5]、workQueue
:
线程池所使用的缓冲队列,常用的是:java.util.concurrent.ArrayBlockingQueue
[ 6 ]、handler
: 线程池对拒绝任务的处理策略,有四个选择如下:
- ThreadPoolExecutor.AbortPolicy():抛出java.util.concurrent.RejectedExecutionException异常
- ThreadPoolExecutor.CallerRunsPolicy():重试添加当前的任务,他会自动重复调用execute()方法
- ThreadPoolExecutor.DiscardOldestPolicy():抛弃旧的任务
- ThreadPoolExecutor.DiscardPolicy():抛弃当前的任务
[ 二 ]、详细说明
[ 1
]、当一个任务通过execute(Runnable)方法欲添加到线程池时:
- 如果此时线程池中的数量小于corePoolSize,即使线程池中的线程都处于空闲状态,也要创建新的线程来处理被添加的任务。
- 如果此时线程池中的数量等于 corePoolSize,但是缓冲队列 workQueue未满,那么任务被放入缓冲队列。
- 如果此时线程池中的数量大于corePoolSize,缓冲队列workQueue满,并且线程池中的数量小于maximumPoolSize,建新的线程来处理被添加的任务。
- 如果此时线程池中的数量大于corePoolSize,缓冲队列workQueue满,并且线程池中的数量等于maximumPoolSize,那么通过 handler所指定的策略来处理此任务。也就是:处理任务的优先级为:核心线程corePoolSize、任务队列workQueue、最大线程maximumPoolSize,如果三者都满了,使用handler处理被拒绝的任务。
- 当线程池中的线程数量大于 corePoolSize时,如果某线程空闲时间超过keepAliveTime,线程将被终止。这样,线程池可以动态的调整池中的线程数。
[ 2 ]、核心和最大池大小
ThreadPoolExecutor 将根据
corePoolSize(参见 getCorePoolSize())和 maximumPoolSize(参见
getMaximumPoolSize())设置的边界自动调整池大小。当新任务在方法 execute(java.lang.Runnable)
中提交时,如果运行的线程少于 corePoolSize,则创建新线程来处理请求,即使其他辅助线程是空闲的。如果运行的线程多于 corePoolSize 而少于
maximumPoolSize,则仅当队列满时才创建新线程。如果设置的 corePoolSize 和 maximumPoolSize
相同,则创建了固定大小的线程池。如果将 maximumPoolSize 设置为基本的无界值(如
Integer.MAX_VALUE),则允许池适应任意数量的并发任务。在大多数情况下,核心和最大池大小仅基于构造来设置,不过也可以使用
setCorePoolSize(int) 和 setMaximumPoolSize(int) 进行动态更改。
[ 3 ]、排队及策略
所有 BlockingQueue
都可用于传输和保持提交的任务。可以使用此队列与池大小进行交互:
- 如果运行的线程少于 corePoolSize,则 Executor 始终首选添加新的线程,而不进行排队。
- 如果运行的线程等于或多于 corePoolSize,则 Executor 始终首选将请求加入队列,而不添加新的线程。
- 如果无法将请求加入队列,则创建新的线程,除非创建此线程超出 maximumPoolSize,在这种情况下,任务将被拒绝。
排队有三种通用策略:
- 直接提交。工作队列的默认选项是 SynchronousQueue,它将任务直接提交给线程而不保持它们。在此,如果不存在可用于立即运行任务的线程,则试图把任务加入队列将失败,因此会构造一个新的线程。此策略可以避免在处理可能具有内部依赖性的请求集合时出现锁定。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。当命令以超过队列所能处理的平均数连续到达时,此策略允许无界线程具有增长的可能性。
- 无界队列。使用无界队列(例如,不具有预定义容量的 LinkedBlockingQueue)将导致在所有 corePoolSize 线程都忙的情况下将新任务加入队列。这样,创建的线程就不会超过 corePoolSize。(因此,maximumPoolSize 的值也就无效了。)当每个任务完全独立于其他任务,即任务执行互不影响时,适合于使用无界队列;例如,在 Web 页服务器中。这种排队可用于处理瞬态突发请求,当命令以超过队列所能处理的平均数连续到达时,此策略允许无界线程具有增长的可能性。
- 有界队列。当使用有限的 maximumPoolSizes 时,有界队列(如 ArrayBlockingQueue)有助于防止资源耗尽,但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷:使用大型队列和小型池可以最大限度地降低 CPU 使用率、操作系统资源和上下文切换开销,但是可能导致人工降低吞吐量。如果任务频繁阻塞(例如,如果它们是 I/O 边界),则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小,CPU 使用率较高,但是可能遇到不可接受的调度开销,这样也会降低吞吐量。
[ 4 ]、拒绝任务的处理策略
(这个和参数handler设置相关)
当 Executor 已经关闭,并且 Executor
将有限边界用于最大线程和工作队列容量,且已经饱和时,在方法 execute(java.lang.Runnable)
中提交的新任务将被拒绝。在以上两种情况下,execute 方法都将调用其 RejectedExecutionHandler 的
RejectedExecutionHandler.rejectedExecution(java.lang.Runnable,
java.util.concurrent.ThreadPoolExecutor) 方法。下面提供了四种预定义的处理程序策略:
- 在默认的 ThreadPoolExecutor.AbortPolicy 中,处理程序遭到拒绝将抛出运行时 RejectedExecutionException。
- 在 ThreadPoolExecutor.CallerRunsPolicy 中,线程调用运行该任务的 execute 本身。此策略提供简单的反馈控制机制,能够减缓新任务的提交速度。
- 在 ThreadPoolExecutor.DiscardPolicy 中,不能执行的任务将被删除。
- 在 ThreadPoolExecutor.DiscardOldestPolicy 中,如果执行程序尚未关闭,则位于工作队列头部的任务将被删除,然后重试执行程序(如果再次失败,则重复此过程)。
- 定义和使用其他种类的 RejectedExecutionHandler 类也是可能的,但这样做需要非常小心,尤其是当策略仅用于特定容量或排队策略时。
[ 三
]、简单示例
JavaThreadPool.java
- package michael.thread.pool;
- import java.util.Date;
- import java.util.concurrent.ArrayBlockingQueue;
- import java.util.concurrent.ThreadPoolExecutor;
- import java.util.concurrent.TimeUnit;
- /**
- * @see http://sjsky.iteye.com
- * @author michael sjsky007@gmail.com
- */
- public class JavaThreadPool {
- /**
- * @param args
- */
- public static void main(String[] args) {
- ThreadPoolExecutor threadPool = new ThreadPoolExecutor( 3 , 5 , 60 ,
- TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>( 10 ),
- new ThreadPoolExecutor.CallerRunsPolicy());
- for ( int i = 0 ; i < 10 ; i++) {
- System.out.println( "add job_" + i + " at:" + new Date());
- SimplePrintJob job = new SimplePrintJob( "job_" + i);
- threadPool.execute(job);
- }
- System.out.println( "execute all job" );
- threadPool.shutdown();
- System.out.println( "main program end-----------" );
- }
- }
package michael.thread.pool; import java.util.Date; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; /** * @see http://sjsky.iteye.com * @author michael sjsky007@gmail.com */ public class JavaThreadPool { /** * @param args */ public static void main(String[] args) { ThreadPoolExecutor threadPool = new ThreadPoolExecutor(3, 5, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(10), new ThreadPoolExecutor.CallerRunsPolicy()); for (int i = 0; i < 10; i++) { System.out.println("add job_" + i + " at:" + new Date()); SimplePrintJob job = new SimplePrintJob("job_" + i); threadPool.execute(job); } System.out.println("execute all job"); threadPool.shutdown(); System.out.println("main program end-----------"); } }
SimplePrintJob.java
- package michael.thread.pool;
- import java.util.Random;
- /**
- * @see http://sjsky.iteye.com
- * @author michael sjsky007@gmail.com
- */
- public class SimplePrintJob implements Runnable {
- private String jobName;
- /**
- * @param jobName
- */
- public SimplePrintJob(String jobName) {
- this .jobName = jobName;
- }
- /**
- * @see java.lang.Runnable#run()
- */
- public void run() {
- System.out.println( "[ " + jobName + " ] start..." );
- int random = 0 ;
- try {
- Random r = new Random();
- random = r.nextInt( 10 );
- Thread.sleep(random * 1000L);
- } catch (Exception e) {
- e.printStackTrace();
- }
- System.out.println( "[ " + jobName + " ] end with sleep:" + random);
- }
- }
package michael.thread.pool; import java.util.Random; /** * @see http://sjsky.iteye.com * @author michael sjsky007@gmail.com */ public class SimplePrintJob implements Runnable { private String jobName; /** * @param jobName */ public SimplePrintJob(String jobName) { this.jobName = jobName; } /** * @see java.lang.Runnable#run() */ public void run() { System.out.println("[ " + jobName + " ] start..."); int random = 0; try { Random r = new Random(); random = r.nextInt(10); Thread.sleep(random * 1000L); } catch (Exception e) { e.printStackTrace(); } System.out.println("[ " + jobName + " ] end with sleep:" + random); } }
运行结果如下:
add job_1 at:Sun Jun 19 00:52:02 CST 2011
add job_2 at:Sun Jun 19 00:52:02 CST 2011
[ job_0 ] start...
add job_3 at:Sun Jun 19 00:52:02 CST 2011
add job_4 at:Sun Jun 19 00:52:02 CST 2011
add job_5 at:Sun Jun 19 00:52:02 CST 2011
add job_6 at:Sun Jun 19 00:52:02 CST 2011
add job_7 at:Sun Jun 19 00:52:02 CST 2011
add job_8 at:Sun Jun 19 00:52:02 CST 2011
add job_9 at:Sun Jun 19 00:52:02 CST 2011
execute all job
[ job_2 ] start...
[ job_0 ] end with sleep:0
[ job_3 ] start...
[ job_2 ] end with sleep:0
[ job_4 ] start...
main program end-----------
[ job_1 ] start...
[ job_4 ] end with sleep:1
[ job_5 ] start...
[ job_5 ] end with sleep:6
[ job_6 ] start...
[ job_1 ] end with sleep:7
[ job_7 ] start...
[ job_3 ] end with sleep:8
[ job_8 ] start...
[ job_6 ] end with sleep:4
[ job_9 ] start...
[ job_7 ] end with sleep:6
[ job_8 ] end with sleep:7
[ job_9 ] end with sleep:5
我们可以自己更改
RejectedExecutionHandler handler这个参数,来观察运行结果会有什么不同。
==============转载http://sjsky.iteye.com/blog/1100208==================
==============转载: 讨论帖 http://www.iteye.com/topic/1118660==================
背景
前段时间一个项目中因为涉及大量的线程开发,把jdk cocurrent的代码重新再过了一遍。这篇文章中主要是记录一下学习ThreadPoolExecutor过程中容易被人忽略的点,Doug Lea的整个类设计还是非常nice的
正文
先看一副图,描述了ThreadPoolExecutor的工作机制:
整个ThreadPoolExecutor的任务处理有4步操作:
- 第一步,初始的poolSize < corePoolSize,提交的runnable任务,会直接做为new一个Thread的参数,立马执行
- 第二步,当提交的任务数超过了corePoolSize,就进入了第二步操作。会将当前的runable提交到一个block queue中
- 第三步,如果block queue是个有界队列,当队列满了之后就进入了第三步。如果poolSize < maximumPoolsize时,会尝试new 一个Thread的进行救急处理,立马执行对应的runnable任务
- 第四步,如果第三步救急方案也无法处理了,就会走到第四步执行reject操作。
- block queue有以下几种实现:
1. ArrayBlockingQueue : 有界的数组队列
2. LinkedBlockingQueue : 可支持有界/无界的队列,使用链表实现
3. PriorityBlockingQueue : 优先队列,可以针对任务排序
4. SynchronousQueue : 队列长度为1的队列,和Array有点区别就是:client thread提交到block queue会是一个阻塞过程,直到有一个worker thread连接上来poll task。 - RejectExecutionHandler是针对任务无法处理时的一些自保护处理:
1. Reject 直接抛出Reject exception
2. Discard 直接忽略该runnable,不可取
3. DiscardOldest 丢弃最早入队列的的任务
4. CallsRun 直接让原先的client thread做为worker线程,进行执行
Btrace容量规划
再提供一个btrace脚本,分析线上的thread pool容量规划是否合理,可以运行时输出poolSize等一些数据。
- import static com.sun.btrace.BTraceUtils.addToAggregation;
- import static com.sun.btrace.BTraceUtils.field;
- import static com.sun.btrace.BTraceUtils.get;
- import static com.sun.btrace.BTraceUtils.newAggregation;
- import static com.sun.btrace.BTraceUtils.newAggregationKey;
- import static com.sun.btrace.BTraceUtils.printAggregation;
- import static com.sun.btrace.BTraceUtils.println;
- import static com.sun.btrace.BTraceUtils.str;
- import static com.sun.btrace.BTraceUtils.strcat;
- import java.lang.reflect.Field;
- import java.util.concurrent.atomic.AtomicInteger;
- import com.sun.btrace.BTraceUtils;
- import com.sun.btrace.aggregation.Aggregation;
- import com.sun.btrace.aggregation.AggregationFunction;
- import com.sun.btrace.aggregation.AggregationKey;
- import com.sun.btrace.annotations.BTrace;
- import com.sun.btrace.annotations.Kind;
- import com.sun.btrace.annotations.Location;
- import com.sun.btrace.annotations.OnEvent;
- import com.sun.btrace.annotations.OnMethod;
- import com.sun.btrace.annotations.OnTimer;
- import com.sun.btrace.annotations.Self;
- /**
- * 并行加载监控
- *
- * @author jianghang 2011-4-7 下午10:59:53
- */
- @BTrace
- public class AsyncLoadTracer {
- private static AtomicInteger rejecctCount = BTraceUtils.newAtomicInteger( 0 );
- private static Aggregation histogram = newAggregation(AggregationFunction.QUANTIZE);
- private static Aggregation average = newAggregation(AggregationFunction.AVERAGE);
- private static Aggregation max = newAggregation(AggregationFunction.MAXIMUM);
- private static Aggregation min = newAggregation(AggregationFunction.MINIMUM);
- private static Aggregation sum = newAggregation(AggregationFunction.SUM);
- private static Aggregation count = newAggregation(AggregationFunction.COUNT);
- @OnMethod (clazz = "java.util.concurrent.ThreadPoolExecutor" , method = "execute" , location = @Location (value = Kind.ENTRY))
- public static void executeMonitor( @Self Object self) {
- Field poolSizeField = field( "java.util.concurrent.ThreadPoolExecutor" , "poolSize" );
- Field largestPoolSizeField = field( "java.util.concurrent.ThreadPoolExecutor" , "largestPoolSize" );
- Field workQueueField = field( "java.util.concurrent.ThreadPoolExecutor" , "workQueue" );
- Field countField = field( "java.util.concurrent.ArrayBlockingQueue" , "count" );
- int poolSize = (Integer) get(poolSizeField, self);
- int largestPoolSize = (Integer) get(largestPoolSizeField, self);
- int queueSize = (Integer) get(countField, get(workQueueField, self));
- println(strcat(strcat(strcat(strcat(strcat( "poolSize : " , str(poolSize)), " largestPoolSize : " ),
- str(largestPoolSize)), " queueSize : " ), str(queueSize)));
- }
- @OnMethod (clazz = "java.util.concurrent.ThreadPoolExecutor" , method = "reject" , location = @Location (value = Kind.ENTRY))
- public static void rejectMonitor( @Self Object self) {
- String name = str(self);
- if (BTraceUtils.startsWith(name, "com.alibaba.pivot.common.asyncload.impl.pool.AsyncLoadThreadPool" )) {
- BTraceUtils.incrementAndGet(rejecctCount);
- }
- }
- @OnTimer ( 1000 )
- public static void rejectPrintln() {
- int reject = BTraceUtils.getAndSet(rejecctCount, 0 );
- println(strcat( "reject count in 1000 msec: " , str(reject)));
- AggregationKey key = newAggregationKey( "rejectCount" );
- addToAggregation(histogram, key, reject);
- addToAggregation(average, key, reject);
- addToAggregation(max, key, reject);
- addToAggregation(min, key, reject);
- addToAggregation(sum, key, reject);
- addToAggregation(count, key, reject);
- }
- @OnEvent
- public static void onEvent() {
- BTraceUtils.truncateAggregation(histogram, 10 );
- println( "---------------------------------------------" );
- printAggregation( "Count" , count);
- printAggregation( "Min" , min);
- printAggregation( "Max" , max);
- printAggregation( "Average" , average);
- printAggregation( "Sum" , sum);
- printAggregation( "Histogram" , histogram);
- println( "---------------------------------------------" );
- }
- }
import static com.sun.btrace.BTraceUtils.addToAggregation; import static com.sun.btrace.BTraceUtils.field; import static com.sun.btrace.BTraceUtils.get; import static com.sun.btrace.BTraceUtils.newAggregation; import static com.sun.btrace.BTraceUtils.newAggregationKey; import static com.sun.btrace.BTraceUtils.printAggregation; import static com.sun.btrace.BTraceUtils.println; import static com.sun.btrace.BTraceUtils.str; import static com.sun.btrace.BTraceUtils.strcat; import java.lang.reflect.Field; import java.util.concurrent.atomic.AtomicInteger; import com.sun.btrace.BTraceUtils; import com.sun.btrace.aggregation.Aggregation; import com.sun.btrace.aggregation.AggregationFunction; import com.sun.btrace.aggregation.AggregationKey; import com.sun.btrace.annotations.BTrace; import com.sun.btrace.annotations.Kind; import com.sun.btrace.annotations.Location; import com.sun.btrace.annotations.OnEvent; import com.sun.btrace.annotations.OnMethod; import com.sun.btrace.annotations.OnTimer; import com.sun.btrace.annotations.Self; /** * 并行加载监控 * * @author jianghang 2011-4-7 下午10:59:53 */ @BTrace public class AsyncLoadTracer { private static AtomicInteger rejecctCount = BTraceUtils.newAtomicInteger(0); private static Aggregation histogram = newAggregation(AggregationFunction.QUANTIZE); private static Aggregation average = newAggregation(AggregationFunction.AVERAGE); private static Aggregation max = newAggregation(AggregationFunction.MAXIMUM); private static Aggregation min = newAggregation(AggregationFunction.MINIMUM); private static Aggregation sum = newAggregation(AggregationFunction.SUM); private static Aggregation count = newAggregation(AggregationFunction.COUNT); @OnMethod(clazz = "java.util.concurrent.ThreadPoolExecutor", method = "execute", location = @Location(value = Kind.ENTRY)) public static void executeMonitor(@Self Object self) { Field poolSizeField = field("java.util.concurrent.ThreadPoolExecutor", "poolSize"); Field largestPoolSizeField = field("java.util.concurrent.ThreadPoolExecutor", "largestPoolSize"); Field workQueueField = field("java.util.concurrent.ThreadPoolExecutor", "workQueue"); Field countField = field("java.util.concurrent.ArrayBlockingQueue", "count"); int poolSize = (Integer) get(poolSizeField, self); int largestPoolSize = (Integer) get(largestPoolSizeField, self); int queueSize = (Integer) get(countField, get(workQueueField, self)); println(strcat(strcat(strcat(strcat(strcat("poolSize : ", str(poolSize)), " largestPoolSize : "), str(largestPoolSize)), " queueSize : "), str(queueSize))); } @OnMethod(clazz = "java.util.concurrent.ThreadPoolExecutor", method = "reject", location = @Location(value = Kind.ENTRY)) public static void rejectMonitor(@Self Object self) { String name = str(self); if (BTraceUtils.startsWith(name, "com.alibaba.pivot.common.asyncload.impl.pool.AsyncLoadThreadPool")) { BTraceUtils.incrementAndGet(rejecctCount); } } @OnTimer(1000) public static void rejectPrintln() { int reject = BTraceUtils.getAndSet(rejecctCount, 0); println(strcat("reject count in 1000 msec: ", str(reject))); AggregationKey key = newAggregationKey("rejectCount"); addToAggregation(histogram, key, reject); addToAggregation(average, key, reject); addToAggregation(max, key, reject); addToAggregation(min, key, reject); addToAggregation(sum, key, reject); addToAggregation(count, key, reject); } @OnEvent public static void onEvent() { BTraceUtils.truncateAggregation(histogram, 10); println("---------------------------------------------"); printAggregation("Count", count); printAggregation("Min", min); printAggregation("Max", max); printAggregation("Average", average); printAggregation("Sum", sum); printAggregation("Histogram", histogram); println("---------------------------------------------"); } }
运行结果:
poolSize : 1 , largestPoolSize = 10 , queueSize = 10 reject count in 1000 msec: 0
说明:
1. poolSize 代表为当前的线程数
2. largestPoolSize 代表为历史最大的线程数
3. queueSize 代表blockqueue的当前堆积的size
4. reject count 代表在1000ms内的被reject的数量
1.ThreadPoolExecutor
Spring中的ThreadPoolTaskExecutor是借助于JDK并发包中的java.util.concurrent.ThreadPoolExecutor来实现的.下面先学习下ThreadPoolExecutor中的相关信息.ThreadPoolExecutor构造函数如下:
- public ThreadPoolExecutor( int corePoolSize,
- int maximumPoolSize,
- long keepAliveTime,
- TimeUnit unit,
- BlockingQueue<Runnable> workQueue,
- ThreadFactory threadFactory,
- RejectedExecutionHandler handler) {
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { }
下面分别说下各项代表的具体意义:
int corePoolSize:线程池维护线程的最小数量.
int
maximumPoolSize:线程池维护线程的最大数量.
long keepAliveTime:空闲线程的存活时间.
TimeUnit
unit: 时间单位,现有纳秒,微秒,毫秒,秒枚举值.
BlockingQueue<Runnable>
workQueue:持有等待执行的任务队列.
RejectedExecutionHandler handler:
用来拒绝一个任务的执行,有两种情况会发生这种情况。
一是在execute方法中若addIfUnderMaximumPoolSize(command)为false,即线程池已经饱和;
二是在execute方法中, 发现runState!=RUNNING || poolSize ==
0,即已经shutdown,就调用ensureQueuedTaskHandled(Runnable command),在该方法中有可能调用reject。
Reject策略预定义有四种:
(1)ThreadPoolExecutor.AbortPolicy策略,是默认的策略,处理程序遭到拒绝将抛出运行时
RejectedExecutionException。
(2)ThreadPoolExecutor.CallerRunsPolicy策略
,调用者的线程会执行该任务,如果执行器已关闭,则丢弃.
(3)ThreadPoolExecutor.DiscardPolicy策略,不能执行的任务将被丢弃.
(4)ThreadPoolExecutor.DiscardOldestPolicy策略,如果执行程序尚未关闭,则位于工作队列头部的任务将被删除,然后重试执行程序(如果再次失败,则重复此过程).
2.
Spring中ThreadPoolTaskExecutor的使用
最常用方式就是做为BEAN注入到容器中,如下代码:
- <bean id= "threadPoolTaskExecutor"
- class = "org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor" >
- <property name= "corePoolSize" value= "10" />
- <property name= "maxPoolSize" value= "15" />
- <property name= "queueCapacity" value= "1000" />
- </bean>
<bean id="threadPoolTaskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor"> <property name="corePoolSize" value="10" /> <property name="maxPoolSize" value="15" /> <property name="queueCapacity" value="1000" /> </bean>
ThreadPoolExecutor执行器的处理流程:
(1)当线程池大小小于corePoolSize就新建线程,并处理请求.
(2)当线程池大小等于corePoolSize,把请求放入workQueue中,池子里的空闲线程就去从workQueue中取任务并处理.
(3)当workQueue放不下新入的任务时,新建线程加入线程池,并处理请求,如果池子大小撑到了maximumPoolSize就用RejectedExecutionHandler来做拒绝处理.
(4)另外,当线程池的线程数大于corePoolSize的时候,多余的线程会等待keepAliveTime长的时间,如果无请求可处理就自行销毁.
了解清楚了ThreadPoolExecutor的执行流程,开头提到的org.springframework.core.task.TaskRejectedException异常也就好理解和解决了.ThreadPoolTaskExecutor类中使用的
就是ThreadPoolExecutor.AbortPolicy()策略,直接抛出异常.
相关推荐
在《阿里巴巴java开发手册》中指出了线程资源必须通过线程池提供,不允许在应用中自行显示的创建线程,这样一方面是线程的创建更加规范,可以合理控制开辟线程的数量;另一方面线程的细节管理交给线程池处理,优化了...
线程池ThreadPoolExecutor实战及其原理分析(下)线程池ThreadPoolExecutor实战及其原理分析(下)线程池ThreadPoolExecutor实战及其原理分析(下)线程池ThreadPoolExecutor实战及其原理分析(下)线程池ThreadPoolExecutor...
Java线程池ThreadPoolExecutor原理及使用实例 Java线程池ThreadPoolExecutor是Java并发编程中的一种基本机制,主要用于管理和执行任务的线程池。下面对其原理和使用实例进行详细介绍。 线程池概述 线程池是一个...
Java线程池是Java并发编程中...总结来说,理解并正确使用Java线程池和ThreadPoolExecutor对于优化Java应用程序的并发性能至关重要。通过调整线程池的参数,可以平衡资源利用率和系统响应时间,从而提高整体的系统效率。
java线程池使用后到底要关闭吗 java线程池是一种高效的并发编程技术,可以帮助开发者更好地管理线程资源,提高系统的性能和可靠性。然而,在使用java线程池时,一个常见的问题是:使用完线程池后到底要不要关闭?...
Java 线程池例子 ThreadPoolExecutor Java 中的线程池是指一个容器,里面包含了多个线程,这些线程可以重复使用,以避免频繁创建和销毁线程的开销。ThreadPoolExecutor 是 Java 中一个非常重要的线程池实现类,它...
线程池ThreadPoolExecutor实战及其原理分析(上)
这个类是Spring对Java内置的`java.util.concurrent.ThreadPoolExecutor`的封装,允许开发者在Spring应用上下文中声明式地定义线程池。在本篇文章中,我们将深入探讨`ThreadPoolTaskExecutor`的配置及其使用,并结合`...
线程池是多线程编程中一种高效管理线程资源的方式,主要由Java的`ThreadPoolExecutor`类实现。线程池的工作机制在于控制线程数量,它会将任务放入队列,然后根据线程池的设定创建并启动线程执行这些任务。如果线程...
### 线程池 `ThreadPoolExecutor` 原理源码分析 #### 一、概述 线程池作为 Java 并发编程中的重要组件,在实际应用中被广泛使用。其核心类 `ThreadPoolExecutor` 实现了对线程的管理、调度等功能。本文将围绕 `...
`ThreadPoolExecutor`是Java并发包`java.util.concurrent`中的核心类,用于实现线程池服务。通过设置线程池参数,我们可以定制线程池的行为,比如核心线程数、最大线程数、线程空闲时间等。 创建一个`...
线程池ThreadPoolExecutor是Java并发编程中一个非常重要的概念,它允许开发者将任务提交给线程池,并由线程池来管理这些任务的执行。今天,我们将对线程池ThreadPoolExecutor进行详细的介绍,并提供一些实际的使用...
Java线程池是一种高效管理线程的技术,它允许开发者预定义一组线程,根据任务的需要灵活调度,而不是每次需要执行任务时都创建新的线程。这种设计模式大大提高了系统的性能,减少了系统资源的消耗,特别是在高并发...
通过上述对`ThreadPoolExecutor`线程池底层实现原理的解析,我们可以看到Java线程池的强大之处在于其高效的状态管理和任务调度机制。通过对`ctl`变量的巧妙利用,线程池能够有效地管理线程状态和数量,从而实现高...
`ThreadPoolExecutor`是Java中非常重要的一个类,它用于创建线程池。线程池可以重用一组预先创建好的线程来执行任务,从而减少了创建和销毁线程所带来的开销,同时也能够控制运行的线程数量,避免由于过多的线程导致...
Java线程池的实现主要有`ThreadPoolExecutor`类,它提供了丰富的构造参数来定制线程池的行为: - `corePoolSize`:线程池的基本大小,即当线程池创建后和运行过程中,即使没有任务,也会保持这个数量的线程存活。 -...
### 自定义实现Java线程池 #### 一、概述 在深入探讨自定义Java线程池之前,我们先简要回顾一下线程池的基本概念及其重要性。线程池是一种多线程处理形式,处理过程中将任务添加到队列,然后在创建线程后自动启动...
要理解`java线程池threadpool简单使用源码`,你需要查看`src`目录下的Java文件,了解如何实例化`ThreadPoolExecutor`,设置相关参数,以及如何提交任务到线程池。同时,查看源码中对`ThreadGroup`的使用,理解它如何...
首先,我们要理解Java线程池的核心类`java.util.concurrent.ThreadPoolExecutor`,它是所有自定义线程池的基类。线程池通过维护一个工作线程集合来管理任务的执行,避免了频繁创建和销毁线程的开销,提高了系统的...
Java线程池是一种高效管理线程的机制,它允许开发者预先设定线程的数量,并通过池化的方式重用已创建的线程,以提高系统性能,减少线程的创建和销毁开销。线程池在Java中是通过`java.util.concurrent`包下的`...