`

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()策略,直接抛出异常.

 

 

分享到:
评论

相关推荐

    java线程池ThreadPoolExecutor类使用详解.docx

    在《阿里巴巴java开发手册》中指出了线程资源必须通过线程池提供,不允许在应用中自行显示的创建线程,这样一方面是线程的创建更加规范,可以合理控制开辟线程的数量;另一方面线程的细节管理交给线程池处理,优化了...

    12、线程池ThreadPoolExecutor实战及其原理分析(下)

    线程池ThreadPoolExecutor实战及其原理分析(下)线程池ThreadPoolExecutor实战及其原理分析(下)线程池ThreadPoolExecutor实战及其原理分析(下)线程池ThreadPoolExecutor实战及其原理分析(下)线程池ThreadPoolExecutor...

    Java线程池ThreadPoolExecutor原理及使用实例

    Java线程池ThreadPoolExecutor原理及使用实例 Java线程池ThreadPoolExecutor是Java并发编程中的一种基本机制,主要用于管理和执行任务的线程池。下面对其原理和使用实例进行详细介绍。 线程池概述 线程池是一个...

    Java线程池与ThreadPoolExecutor.pdf

    Java线程池是Java并发编程中...总结来说,理解并正确使用Java线程池和ThreadPoolExecutor对于优化Java应用程序的并发性能至关重要。通过调整线程池的参数,可以平衡资源利用率和系统响应时间,从而提高整体的系统效率。

    java线程池使用后到底要关闭吗

    java线程池使用后到底要关闭吗 java线程池是一种高效的并发编程技术,可以帮助开发者更好地管理线程资源,提高系统的性能和可靠性。然而,在使用java线程池时,一个常见的问题是:使用完线程池后到底要不要关闭?...

    java 线程池例子ThreadPoolExecutor

    Java 线程池例子 ThreadPoolExecutor Java 中的线程池是指一个容器,里面包含了多个线程,这些线程可以重复使用,以避免频繁创建和销毁线程的开销。ThreadPoolExecutor 是 Java 中一个非常重要的线程池实现类,它...

    11、线程池ThreadPoolExecutor实战及其原理分析(上)

    线程池ThreadPoolExecutor实战及其原理分析(上)

    spring线程池ThreadPoolExecutor配置以及FutureTask的使用

    这个类是Spring对Java内置的`java.util.concurrent.ThreadPoolExecutor`的封装,允许开发者在Spring应用上下文中声明式地定义线程池。在本篇文章中,我们将深入探讨`ThreadPoolTaskExecutor`的配置及其使用,并结合`...

    线程池之ThreadPoolExecutor.docx

    线程池是多线程编程中一种高效管理线程资源的方式,主要由Java的`ThreadPoolExecutor`类实现。线程池的工作机制在于控制线程数量,它会将任务放入队列,然后根据线程池的设定创建并启动线程执行这些任务。如果线程...

    线程池ThreadPoolExecutor原理源码分析.md

    ### 线程池 `ThreadPoolExecutor` 原理源码分析 #### 一、概述 线程池作为 Java 并发编程中的重要组件,在实际应用中被广泛使用。其核心类 `ThreadPoolExecutor` 实现了对线程的管理、调度等功能。本文将围绕 `...

    使用线程池ThreadPoolExecutor 抓取论坛帖子列表

    `ThreadPoolExecutor`是Java并发包`java.util.concurrent`中的核心类,用于实现线程池服务。通过设置线程池参数,我们可以定制线程池的行为,比如核心线程数、最大线程数、线程空闲时间等。 创建一个`...

    线程池ThreadPoolExecutor使用简介与方法实例

    线程池ThreadPoolExecutor是Java并发编程中一个非常重要的概念,它允许开发者将任务提交给线程池,并由线程池来管理这些任务的执行。今天,我们将对线程池ThreadPoolExecutor进行详细的介绍,并提供一些实际的使用...

    java线程池封装j

    Java线程池是一种高效管理线程的技术,它允许开发者预定义一组线程,根据任务的需要灵活调度,而不是每次需要执行任务时都创建新的线程。这种设计模式大大提高了系统的性能,减少了系统资源的消耗,特别是在高并发...

    12-线程池ThreadPoolExecutor底层原理源码分析(下)-周瑜.pdf

    通过上述对`ThreadPoolExecutor`线程池底层实现原理的解析,我们可以看到Java线程池的强大之处在于其高效的状态管理和任务调度机制。通过对`ctl`变量的巧妙利用,线程池能够有效地管理线程状态和数量,从而实现高...

    11-线程池ThreadPoolExecutor底层原理源码分析(上)-周瑜.pdf

    `ThreadPoolExecutor`是Java中非常重要的一个类,它用于创建线程池。线程池可以重用一组预先创建好的线程来执行任务,从而减少了创建和销毁线程所带来的开销,同时也能够控制运行的线程数量,避免由于过多的线程导致...

    java线程池实例详细讲解

    Java线程池的实现主要有`ThreadPoolExecutor`类,它提供了丰富的构造参数来定制线程池的行为: - `corePoolSize`:线程池的基本大小,即当线程池创建后和运行过程中,即使没有任务,也会保持这个数量的线程存活。 -...

    自定义实现Java线程池

    ### 自定义实现Java线程池 #### 一、概述 在深入探讨自定义Java线程池之前,我们先简要回顾一下线程池的基本概念及其重要性。线程池是一种多线程处理形式,处理过程中将任务添加到队列,然后在创建线程后自动启动...

    java线程池threadpool简单使用源码

    要理解`java线程池threadpool简单使用源码`,你需要查看`src`目录下的Java文件,了解如何实例化`ThreadPoolExecutor`,设置相关参数,以及如何提交任务到线程池。同时,查看源码中对`ThreadGroup`的使用,理解它如何...

    java线程池的源码分析.zip

    首先,我们要理解Java线程池的核心类`java.util.concurrent.ThreadPoolExecutor`,它是所有自定义线程池的基类。线程池通过维护一个工作线程集合来管理任务的执行,避免了频繁创建和销毁线程的开销,提高了系统的...

    Java线程池文档

    Java线程池是一种高效管理线程的机制,它允许开发者预先设定线程的数量,并通过池化的方式重用已创建的线程,以提高系统性能,减少线程的创建和销毁开销。线程池在Java中是通过`java.util.concurrent`包下的`...

Global site tag (gtag.js) - Google Analytics