posts - 110, comments - 101, trackbacks - 0, articles - 7
  BlogJava :: 首页 :: 新随笔 :: 联系 :: 聚合  :: 管理

ThreadPoolExecutor 很好的一篇文章 路过就留下了

Posted on 2013-01-14 16:08 云云 阅读(373) 评论(0)  编辑  收藏

 原文：http://www.iteye.com/topic/1118660

整个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线程，进行执行

容易被人忽略的点：
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等一些数据。

 

 

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

运行结果：

 

Java代码  
  1. poolSize : 1 , largestPoolSize = 10 , queueSize = 10  
  2. reject count in 1000 msec: 0  

 

说明：

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

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

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

4. reject count 代表在1000ms内的被reject的数量

 

 

最后

  这是我对ThreadPoolExecutor使用过程中的一些经验总结，希望能对大家有所帮助，如有描述不对的地方欢迎拍砖。


只有注册用户登录后才能发表评论。
网站导航: