JAVA 线程池源码分析

java5之后为我们提供了线程池，只需要使用API，不用去考虑线程池里特殊的处理机制。

Java5的线程池分好多种：固定尺寸的线程池、可变尺寸连接池等。

常用的是ThreadPoolExecutor，它的继承关系如下：

 

下面详细说明一下自定义线程的使用：

TreadPoolExecutor的常用构造方法如下：

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              RejectedExecutionHandler handler) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), handler);
    }

 

下面是参数说明：

corePoolSize：线程池维护线程的最少数量，有可能是空闲的线程。
maximunPoolSize：线程池维护线程的最大数量。
keepAliveTime：线程池维护线程所允许的空闲时间。
TimeUnit：程池维护线程所允许的空闲时间的单位。
workQueue：线程池所使用的缓冲队列，改缓冲队列的长度决定了能够缓冲的最大数量。
RejectedExecutionHandler ：拒绝任务的处理方式。
拒绝任务：是指当线程池里面的线程数量达到 maximumPoolSize 且 workQueue 队列已满的情况下被尝试添加进来的任务。

 

在 ThreadPoolExecutor 里面定义了 4 种 handler 策略，分别是

CallerRunsPolicy ：这个策略重试添加当前的任务，他会自动重复调用 execute() 方法，直到成功。
AbortPolicy ：对拒绝任务抛弃处理，并且抛出异常。
DiscardPolicy ：对拒绝任务直接无声抛弃，没有异常信息。
DiscardOldestPolicy ：对拒绝任务不抛弃，而是抛弃队列里面等待最久的一个线程，然后把拒绝任务加到队列。
一个任务通过 execute(Runnable) 方法被添加到线程池，任务就是一个 Runnable 类型的对象，任务的执行方法就是 Runnable 类型对象的 run() 方法。

当一个任务通过 execute(Runnable) 方法欲添加到线程池时，线程池采用的策略如下：

1. 如果此时线程池中的数量小于 corePoolSize ，即使线程池中的线程都处于空闲状态，也要创建新的线程   来处理被添加的任务。

2. 如果此时线程池中的数量等于 corePoolSize ，但是缓冲队列 workQueue 未满，那么任务被放入缓冲队列。

3. 如果此时线程池中的数量大于 corePoolSize ，缓冲队列 workQueue 满，并且线程池中的数量小于maximumPoolSize ，建新的线程来处理被添加的任务。

4. 如果此时线程池中的数量大于 corePoolSize ，缓冲队列 workQueue 满，并且线程池中的数量等于maximumPoolSize ，那么通过 handler 所指定的策略来处理此任务。

 

处理任务的优先级为：

核心线程 corePoolSize 、任务队列 workQueue 、最大线程 maximumPoolSize ，如果三者都满了，使用handler 处理被拒绝的任务。当线程池中的线程数量大于 corePoolSize 时，如果某线程空闲时间超过keepAliveTime ，线程将被终止。这样，线程池可以动态的调整池中的线程数。

 

我应用中的代码如下：

 

public class ThreadExecute{
 
    private static Logger   logger  = LoggerFactory.getLogger(ThreadExecute.class);
    private final int   corePoolSize    = 2;      // 线程池维护线程的最少数量
     private final int  maximumPoolSize   = 4;    // 线程池维护线程的最大数量
     private final long   keepAliveTime  = 3;    // 线程池维护线程所允许的空闲时间
     private static int   produceTaskMaxNumber = 4;
 
 
    public void processMessageTask() {
        // 创建等待队列
        BlockingQueue<Runnable> bqueue = new ArrayBlockingQueue<Runnable>(20);
        // 构造一个线程池
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime,
                                                               TimeUnit.SECONDS, bqueue,
                                                               new ThreadPoolExecutor.CallerRunsPolicy());
 
        for (int i = 1; i <= produceTaskMaxNumber; i++) {
            try {
                threadPool.execute(new MyThread());
            } catch (Exception e) {
                logger.error("thread pool is error, content::" + e);
            }
        }
    }
 
}
 
class MyThread implements Runnable {
 
    public void run() {
            while(true){
                try {
                    Thread.sleep(2*1000);
                    System.out.println(Thread.currentThread().getName() + "正在执行。。。");
                } catch (InterruptedException e) {
 
                }
            }
    }
}