Java的ThreadPoolExecutor使用几点建议

fmms 13年前
     <h1 style="font-size:2em;">背景</h1>    <p style="margin:0px;">前段时间一个项目中因为涉及大量的线程开发,把jdk cocurrent的代码重新再过了一遍。这篇文章中主要是记录一下学习ThreadPoolExecutor过程中容易被人忽略的点,Doug Lea的整个类设计还是非常nice的</p>    <p style="margin:0px;"> </p>    <h1 style="font-size:2em;">正文</h1>    <p style="margin:0px;">先看一副图,描述了ThreadPoolExecutor的工作机制: </p>    <p style="margin:0px;"><img style="width:734px;height:369px;" class="magplus" title="点击查看原始大小图片" alt="ThreadPoolExecutor几点使用建议" src="https://simg.open-open.com/show/c1ef6265d236e90c361e6c943c288746.jpg" /></p>    <p style="margin:0px;"> </p>    <p style="margin:0px;">整个ThreadPoolExecutor的任务处理有4步操作:</p>    <p style="margin:0px;"> </p>    <ul>     <li>第一步,初始的poolSize < corePoolSize,提交的runnable任务,会直接做为new一个Thread的参数,立马执行</li>     <li>第二步,当提交的任务数超过了corePoolSize,就进入了第二步操作。会将当前的runable提交到一个block queue中<br /> </li>     <li>第三步,如果block queue是个有界队列,当队列满了之后就进入了第三步。如果poolSize < maximumPoolsize时,会尝试new 一个Thread的进行救急处理,立马执行对应的runnable任务</li>     <li>第四步,如果第三步救急方案也无法处理了,就会走到第四步执行reject操作。</li>    </ul>    <div>     几点说明:(相信这些网上一搜一大把,我这里简单介绍下,为后面做一下铺垫)    </div>    <div>     <ul>      <li>block queue有以下几种实现:<br /> 1. ArrayBlockingQueue :  有界的数组队列<br /> 2. LinkedBlockingQueue : 可支持有界/无界的队列,使用链表实现<br /> 3. PriorityBlockingQueue : 优先队列,可以针对任务排序<br /> 4. SynchronousQueue : 队列长度为1的队列,和Array有点区别就是:client thread提交到block queue会是一个阻塞过程,直到有一个worker thread连接上来poll task。</li>      <li>RejectExecutionHandler是针对任务无法处理时的一些自保护处理:<br /> 1. Reject 直接抛出Reject exception<br /> 2. Discard 直接忽略该runnable,不可取<br /> 3. DiscardOldest 丢弃最早入队列的的任务<br /> 4. CallsRun 直接让原先的client thread做为worker线程,进行执行</li>     </ul>    </div>    <div>     <br />    </div>    <div>     容易被人忽略的点:    </div>    <div>     1.  pool threads启动后,以后的任务获取都会通过block queue中,获取堆积的runnable task.    </div>    <div>     <br />    </div>    <div>     所以建议:      <span style="color:#ff0000;">block size >= corePoolSize ,不然线程池就没任何意义</span>    </div>    <div>     2.  corePoolSize 和 maximumPoolSize的区别, 和大家正常理解的数据库连接池不太一样。    </div>    <div>       *  据dbcp pool为例,会有minIdle , maxActive配置。minIdle代表是常驻内存中的threads数量,maxActive代表是工作的最大线程数。    </div>    <div>       *  这里的corePoolSize就是连接池的maxActive的概念,它没有minIdle的概念(每个线程可以设置keepAliveTime,超过多少时间多有任务后销毁线程,但不会固定保持一定数量的threads)。     </div>    <div>       * 这里的maximumPoolSize,是一种救急措施的第一层。当threadPoolExecutor的工作threads存在满负荷,并且block queue队列也满了,这时代表接近崩溃边缘。这时允许临时起一批threads,用来处理runnable,处理完后立马退出。    </div>    <div>     <br />    </div>    <div>     所以建议:      <span style="color:#ff0000;"> maximumPoolSize >= corePoolSize =期望的最大线程数。 (我曾经配置了corePoolSize=1, maximumPoolSize=20, blockqueue为无界队列,最后就成了单线程工作的pool。典型的配置错误)</span>    </div>    <div>     <br />    </div>    <div>     3. 善用blockqueue和reject组合. 这里要重点推荐下CallsRun的Rejected Handler,从字面意思就是让调用者自己来运行。    </div>    <div>     我们经常会在线上使用一些线程池做异步处理,比如我前面做的     <span style="line-height:24px;font-family:Helvetica, Tahoma, Arial, sans-serif;font-size:16px;font-weight:bold;"><a style="color:#006699;text-decoration:underline;" href="/misc/goto?guid=4959498383989483162">(业务层)异步并行加载技术分析和设计</a>, </span>将原本串行的请求都变为了并行操作,但过多的并行会增加系统的负载(比如软中断,上下文切换)。所以肯定需要对线程池做一个size限制。但是为了引入异步操作后,避免因在block queue的等待时间过长,所以需要在队列满的时,执行一个callsRun的策略,并行的操作又转为一个串行处理,这样就可以保证尽量少的延迟影响。    </div>    <div>     <br />    </div>    <div>     所以建议:      <span style="color:#ff0000;"> RejectExecutionHandler = CallsRun ,  blockqueue size = 2 * poolSize (为啥是2倍poolSize,主要一个考虑就是瞬间高峰处理,允许一个thread等待一个runnable任务)</span>    </div>    <h1 style="font-size:2em;">Btrace容量规划</h1>    <p style="margin:0px;">再提供一个btrace脚本,分析线上的thread pool容量规划是否合理,可以运行时输出poolSize等一些数据。</p>    <pre class="brush:java; toolbar: true; auto-links: false;">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("---------------------------------------------");     } }</pre>    <p></p>    <p style="margin:0px;">运行结果:</p>    <pre class="brush:java; toolbar: true; auto-links: false;">poolSize : 1 , largestPoolSize = 10 , queueSize = 10 reject count in 1000 msec: 0</pre>    <p></p>    <p style="margin:0px;">说明:</p>    <p style="margin:0px;">1. poolSize 代表为当前的线程数</p>    <p style="margin:0px;">2. largestPoolSize 代表为历史最大的线程数</p>    <p style="margin:0px;">3. queueSize 代表blockqueue的当前堆积的size</p>    <p style="margin:0px;">4. reject count 代表在1000ms内的被reject的数量。<br /> <br /> 转自:<a href="/misc/goto?guid=4959498384076520772" target="_blank">http://www.iteye.com/topic/1118660</a></p>    <p style="margin:0px;"></p>    <p style="margin:0px;"></p>