Java的ThreadPoolExecutor使用几点建议

IT程序员开发必备-各类资源下载清单,史上最全IT资源,个人收藏总结!

背景

前段时间一个项目中因为涉及大量的线程开发,把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 exception2. Discard 直接忽略该runnable,不可取3. DiscardOldest 丢弃最早入队列的的任务4. CallsRun 直接让原先的client thread做为worker线程,进行执行容易被人忽略的点: 1. pool threads启动后,以后的任务获取都会通过block queue中,获取堆积的runnable task. 所以建议: block size >= corePoolSize ,不然线程池就没任何意义2. corePoolSize 和maximumPoolSize的区别, 和大家正常理解的数据库连接池不太一样。 * 据dbcp pool为例,会有minIdle , maxActive配置。minIdle代表是常驻内存中的threads数量,maxActive代表是工作的最大线程数。* 这里的corePoolSize就是连接池的maxActive的概念,它没有minIdle的概念(每个线程可以设置keepAliveTime,超过多少时间多有任务后销毁线程,但不会固定保持一定数量的threads)。* 这里的maximumPoolSize,是一种救急措施的第一层。当threadPoolExecutor的工作threads存在满负荷,并且block queue队列也满了,这时代表接近崩溃边缘。这时允许临时起一批threads,用来处理runnable,处理完后立马退出。所以建议: maximumPoolSize >=corePoolSize =期望的最大线程数。 (我曾经配置了corePoolSize=1,maximumPoolSize=20, blockqueue为无界队列,最后就成了单线程工作的pool。典型的配置错误)3. 善用blockqueue和reject组合. 这里要重点推荐下CallsRun的Rejected Handler,从字面意思就是让调用者自己来运行。我们经常会在线上使用一些线程池做异步处理,比如我前面做的 (业务层)异步并行加载技术分析和设计,将原本串行的请求都变为了并行操作,但过多的并行会增加系统的负载(比如软中断,上下文切换)。所以肯定需要对线程池做一个size限制。但是为了引入异步操作后,避免因在block queue的等待时间过长,所以需要在队列满的时,执行一个callsRun的策略,并行的操作又转为一个串行处理,这样就可以保证尽量少的延迟影响。所以建议: RejectExecutionHandler =CallsRun , blockqueue size = 2 *poolSize (为啥是2倍poolSize,主要一个考虑就是瞬间高峰处理,允许一个thread等待一个runnable任务)Btrace容量规划

再提供一个btrace脚本,分析线上的thread pool容量规划是否合理,可以运行时输出poolSize等一些数据。

01import staticcom.sun.btrace.BTraceUtils.addToAggregation; 02import staticcom.sun.btrace.BTraceUtils.field; 03import staticcom.sun.btrace.BTraceUtils.get; 04import staticcom.sun.btrace.BTraceUtils.newAggregation; 05import staticcom.sun.btrace.BTraceUtils.newAggregationKey; 06import staticcom.sun.btrace.BTraceUtils.printAggregation; 07import staticcom.sun.btrace.BTraceUtils.println; 08import staticcom.sun.btrace.BTraceUtils.str; 09import staticcom.sun.btrace.BTraceUtils.strcat; 10 11import java.lang.reflect.Field; 12import java.util.concurrent.atomic.AtomicInteger; 13 14import com.sun.btrace.BTraceUtils; 15import com.sun.btrace.aggregation.Aggregation; 16import com.sun.btrace.aggregation.AggregationFunction; 17import com.sun.btrace.aggregation.AggregationKey; 18import com.sun.btrace.annotations.BTrace; 19import com.sun.btrace.annotations.Kind; 20import com.sun.btrace.annotations.Location; 21import com.sun.btrace.annotations.OnEvent; 22import com.sun.btrace.annotations.OnMethod; 23import com.sun.btrace.annotations.OnTimer; 24import com.sun.btrace.annotations.Self; 25 26/** 27* 并行加载监控 28* 29* @author jianghang 2011-4-7 下午10:59:53 30*/ 31@BTrace 32public classAsyncLoadTracer { 33 34privatestaticAtomicInteger rejecctCount = BTraceUtils.newAtomicInteger(0); 35privatestaticAggregation histogram = newAggregation(AggregationFunction.QUANTIZE); 36privatestaticAggregation average = newAggregation(AggregationFunction.AVERAGE); 37privatestaticAggregation max = newAggregation(AggregationFunction.MAXIMUM); 38privatestaticAggregation min = newAggregation(AggregationFunction.MINIMUM); 39privatestaticAggregation sum = newAggregation(AggregationFunction.SUM); 40privatestaticAggregation count = newAggregation(AggregationFunction.COUNT); 41 42@OnMethod(clazz ="java.util.concurrent.ThreadPoolExecutor", method ="execute", location = @Location(value = Kind.ENTRY)) 43publicstaticvoid executeMonitor(@SelfObject self) { 44Field poolSizeField = field("java.util.concurrent.ThreadPoolExecutor","poolSize"); 45Field largestPoolSizeField = field("java.util.concurrent.ThreadPoolExecutor","largestPoolSize"); 46Field workQueueField = field("java.util.concurrent.ThreadPoolExecutor","workQueue"); 47 48Field countField = field("java.util.concurrent.ArrayBlockingQueue","count"); 49intpoolSize = (Integer) get(poolSizeField, self); 50intlargestPoolSize = (Integer) get(largestPoolSizeField, self); 51intqueueSize = (Integer) get(countField, get(workQueueField, self)); 52 53println(strcat(strcat(strcat(strcat(strcat("poolSize : ", str(poolSize))," largestPoolSize : "), 54str(largestPoolSize))," queueSize : "), str(queueSize))); 55} 56 57@OnMethod(clazz ="java.util.concurrent.ThreadPoolExecutor", method ="reject", location = @Location(value = Kind.ENTRY)) 58publicstaticvoid rejectMonitor(@SelfObject self) { 59String name = str(self); 60if(BTraceUtils.startsWith(name,"com.alibaba.pivot.common.asyncload.impl.pool.AsyncLoadThreadPool")) { 61BTraceUtils.incrementAndGet(rejecctCount); 62} 63} 64 65@OnTimer(1000) 66publicstaticvoid rejectPrintln() { 67intreject = BTraceUtils.getAndSet(rejecctCount,0); 68println(strcat("reject count in 1000 msec: ", str(reject))); 69AggregationKey key = newAggregationKey("rejectCount"); 70addToAggregation(histogram, key, reject); 71addToAggregation(average, key, reject); 72addToAggregation(max, key, reject); 73addToAggregation(min, key, reject); 74addToAggregation(sum, key, reject); 75addToAggregation(count, key, reject); 76} 77 78@OnEvent 79publicstaticvoid onEvent() { 80BTraceUtils.truncateAggregation(histogram,10); 81println("---------------------------------------------"); 82printAggregation("Count", count); 83printAggregation("Min", min); 84printAggregation("Max", max); 85printAggregation("Average", average); 86printAggregation("Sum", sum); 87printAggregation("Histogram", histogram); 88println("---------------------------------------------"); 89} 90}

运行结果:

1poolSize : 1, largestPoolSize =10 , queueSize = 10 2reject count in 1000msec:0

说明:

1. poolSize 代表为当前的线程数

2. largestPoolSize 代表为历史最大的线程数

3. queueSize 代表blockqueue的当前堆积的size

4. reject count 代表在1000ms内的被reject的数量。一个今天胜过两个明天

Java的ThreadPoolExecutor使用几点建议

相关文章:

你感兴趣的文章:

标签云: