`

JAVA线程池ThreadPoolExecutor

 
阅读更多

java.util.concurrent.ThreadPoolExecutor相关基础介绍和使用示例。
[ 一 ]、常用线程池
最常用构造方法为：

Java代码 复制代码  收藏代码
  1. ThreadPoolExecutor( int  corePoolSize,   
  2.                     int  maximumPoolSize,   
  3.                     long  keepAliveTime,   
  4.                    TimeUnit unit,   
  5.                    BlockingQueue<Runnable> workQueue,   
  6.                    RejectedExecutionHandler handler)   
ThreadPoolExecutor(int corePoolSize,
                   int maximumPoolSize,
                   long keepAliveTime,
                   TimeUnit unit,
                   BlockingQueue<Runnable> workQueue,
                   RejectedExecutionHandler handler)



JDK自带的配置好的线程池：

Java代码 复制代码  收藏代码
  1. // 固定工作线程数量的线程池   
  2. ExecutorService executorService1 = Executors.newFixedThreadPool( 3 );   
  3.   
  4. // 一个可缓存的线程池   
  5. ExecutorService executorService2 = Executors.newCachedThreadPool();   
  6.   
  7. // 单线程化的Executor   
  8. ExecutorService executorService3 = Executors.newSingleThreadExecutor();   
  9.   
  10. // 支持定时的以及周期性的任务执行   
  11. 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

Java代码 复制代码  收藏代码
  1. package  michael.thread.pool;   
  2.   
  3. import  java.util.Date;   
  4. import  java.util.concurrent.ArrayBlockingQueue;   
  5. import  java.util.concurrent.ThreadPoolExecutor;   
  6. import  java.util.concurrent.TimeUnit;   
  7.   
  8.   
  9. /**  
  10.  * @see http://sjsky.iteye.com  
  11.  * @author michael sjsky007@gmail.com  
  12.  */   
  13. public   class  JavaThreadPool {   
  14.   
  15.      /**  
  16.      * @param args  
  17.      */   
  18.      public   static   void  main(String[] args) {   
  19.   
  20.         ThreadPoolExecutor threadPool =  new  ThreadPoolExecutor( 3 ,  5 ,  60 ,   
  21.                 TimeUnit.SECONDS,  new  ArrayBlockingQueue<Runnable>( 10 ),   
  22.                  new  ThreadPoolExecutor.CallerRunsPolicy());   
  23.          for  ( int  i =  0 ; i <  10 ; i++) {   
  24.             System.out.println( "add job_"  + i +  " at:"  +  new  Date());   
  25.             SimplePrintJob job =  new  SimplePrintJob( "job_"  + i);   
  26.             threadPool.execute(job);   
  27.         }   
  28.         System.out.println( "execute all job" );   
  29.         threadPool.shutdown();   
  30.         System.out.println( "main program end-----------" );   
  31.     }   
  32.   
  33. }   
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

Java代码 复制代码  收藏代码
  1. package  michael.thread.pool;   
  2.   
  3. import  java.util.Random;   
  4.   
  5.   
  6. /**  
  7.  * @see http://sjsky.iteye.com  
  8.  * @author michael sjsky007@gmail.com  
  9.  */   
  10. public   class  SimplePrintJob  implements  Runnable {   
  11.   
  12.      private  String jobName;   
  13.   
  14.      /**  
  15.      * @param jobName  
  16.      */   
  17.      public  SimplePrintJob(String jobName) {   
  18.          this .jobName = jobName;   
  19.     }   
  20.   
  21.      /**  
  22.      * @see java.lang.Runnable#run()  
  23.      */   
  24.      public   void  run() {   
  25.         System.out.println( "[ "  + jobName +  " ] start..." );   
  26.          int  random =  0 ;   
  27.          try  {   
  28.             Random r =  new  Random();   
  29.             random = r.nextInt( 10 );   
  30.             Thread.sleep(random * 1000L);   
  31.         }  catch  (Exception e) {   
  32.             e.printStackTrace();   
  33.         }   
  34.         System.out.println( "[ "  + jobName +  " ] end with sleep:"  + random);   
  35.     }   
  36. }   
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_0 at:Sun Jun 19 00:52:02 CST 2011
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线程，进行执行

容易被人忽略的点：
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. }   
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("---------------------------------------------");
    }
}
 

运行结果：

 

Java代码 复制代码  收藏代码
  1. poolSize :  1  , largestPoolSize =  10  , queueSize =  10   
  2. reject count in  1000  msec:  0    
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构造函数如下:

Java代码 复制代码  收藏代码
  1. public  ThreadPoolExecutor( int  corePoolSize,   
  2.                            int  maximumPoolSize,   
  3.                            long  keepAliveTime,   
  4.                           TimeUnit unit,   
  5.                           BlockingQueue<Runnable> workQueue,   
  6.                           ThreadFactory threadFactory,   
  7.                           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注入到容器中,如下代码:

Java代码 复制代码  收藏代码
  1. <bean id= "threadPoolTaskExecutor"   
  2.      class = "org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor" >   
  3.     <property name= "corePoolSize"  value= "10"  />   
  4.     <property name= "maxPoolSize"  value= "15"  />   
  5.     <property name= "queueCapacity"  value= "1000"  />   
  6. </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()策略,直接抛出异常.

 

 

分享到：
| ThreadLocal-分析-总结
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

Magicbox
Global site tag (gtag.js) - Google Analytics