线程池的使用及ThreadPoolExecutor的分析(一)

说明:本作者是文章的原创作者,转载请注明出处:本文地址:http://www.cnblogs.com/qm-article/p/7821602.html

一、线程池的介绍

在开发中,频繁的创建和销毁一个线程,是很耗资源的,为此找出了一个可以循环利用已经存在的线程来达到自己的目的,线程池顾名思义,也就是线程池的集合,通过线程池执行的线程任务,可以很有效的去规划线程的使用。
在java中大致有这几种线程池
      newScheduledThreadPool  创建一个定长线程池,支持定时及周期性任务执行。,可以作一个定时器使用。
      newCachedThreadPool       创建一个可缓存线程池,如果线程池长度超过需要的线程数量,可灵活回收空闲线程,若无可回收,则新建线程。
      newSingleThreadExecutor 创建一个单线程化的线程池, 它只会用唯一的工作线程来执行任务,保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行,可以控制线程的执行顺序
      newFixedThreadPool          创建一个定长线程池,可控制线程最大并发数,超出的线程会在队列中等待,当创建的线程池数量为1的时候。也类似于单线程化的线程池,当为1的时候,也可控制线程的执行顺序

二、线程池的使用

1、newScheduledThreadPool

    /**
* 测试newScheduledThreadPool 创建一个定长线程池,支持定时及周期性任务执行。
* 一般可做定时器使用
*/
public static void test_1(){
//参数是线程的数量
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
/**
* 第二个参数,是首次执行该线程的延迟时间,之后失效
* 第三个参数是,首次执行完之后,再过该段时间再次执行该线程,具有周期性
*/
scheduledExecutorService.scheduleAtFixedRate(new Runnable() { @Override
public void run() {
System.out.println(new Date().getSeconds()); }
}, 10, 3, TimeUnit.SECONDS); }

2、newCachedThreadPool      

     /**
* newCachedThreadPool创建一个可缓存线程池,
* 如果线程池长度超过处理需要,可灵活回收空闲线程,若无可回收,则新建线程。
*/
public static void test_2(){
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int index = i;
try {
Thread.sleep(index * 1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
cachedThreadPool.execute(new Runnable() { @Override
public void run() {
// TODO Auto-generated method stub
System.out.println(index+":"+new Date().getSeconds());
}
});
}
}

3、newSingleThreadExecutor

     /**
* newSingleThreadExecutor 创建一个单线程化的线程池,
* 它只会用唯一的工作线程来执行任务,保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行
*/
public static void test_4(){
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
for(int i = 1; i < 11; i++){
final int index = i;
singleThreadExecutor.execute(new Runnable() {
@Override
public void run() {
// TODO Auto-generated method stub
//会按顺序打印
System.out.println(index);
}
});
}
}

4、newFixedThreadPool

     /**
* newFixedThreadPool 创建一个定长线程池,可控制线程最大并发数,超出的线程会在队列中等待
*/
public static void test_3(){
//当参数为1的时候,可以控制线程的执行顺序,类似join的作用
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(2);
for(int i = 0; i < 2; i++){
final int index = i;
fixedThreadPool.execute(new Runnable() { @Override
public void run() {
// TODO Auto-generated method stub
try {
System.out.println(index);
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
}
}

三、线程池源码分析

以上四种线程都是由一个线程工具类Executors来创造的

线程池的使用及ThreadPoolExecutor的分析(一)

如上图,其中newFixedThreadPool 和newCachedThreadPool 都是由threadPoolExecutor来创建的,只是参数不一致而已,
关于threadPoolExector的构造器的参数

线程池的使用及ThreadPoolExecutor的分析(一)

corePoolSize 代表该线程中允许的核心线程数,要和工作的线程数量区分开来,两者不
                      等价(工作的线程数量一定不大于corePoolSize,即当超过后,会将线程
                      放入队列中),可以理解为一个ArrayList集合中,默认空间是10,但存放的
                     元素的数量 不一定是10, 在这里这个10就寓指corePoolSize ,存放元
                     素的个数是工作线程数量
maximumPoolSize 这个参数的意思就是该线程池所允许的最大线程数量
keepAliveTime 这个参数的意思就是空余线程的存活时间,注意这个值并不会对所有线程起作用,如果线程池中的线程数少于等于核心线程数 corePoolSize,那么这些线程不会因                           为空闲太长时间而被关闭,当然,也可以通过调用allowCoreThreadTimeOut方法使核心线程数内的线程也可以被回收。

unit 时间单位
workQueue 阻塞队列,在此作用就是用来存放线程。
threadFactory 线程工厂
defaultHandler 拒绝策略,即当加入线程失败,采用该handler来处理

3.1、线程池的拒绝策略

AbortPolicy
        为java线程池默认的阻塞策略,不执行此任务,而且直接抛出一个运行时异常
DiscardPolicy
        直接抛弃,任务不执行,空方法
DiscardOldestPolicy
        从队列里面抛弃head的一个任务,并再次execute 此task。
CallerRunsPolicy
        在调用execute的线程里面执行此command,会阻塞入口

在分析该类的execute方法前,先看这几个常量的值和一些方法的作用

    /*
* ctl的默认值为-536870912,
* 作用是将该值传入workerCountOf(int c)的参数c中,
* 则可以返回正在工作的线程数量
* 每当有一个线程加入工作,该值会加1
*/
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3; //32-3=29
private static final int CAPACITY = (1 << COUNT_BITS) - 1;//536870911 // runState is stored in the high-order bits,其中running<shutdown<stop<tidying<terminated
private static final int RUNNING = -1 << COUNT_BITS;// -536870912
private static final int SHUTDOWN = 0 << COUNT_BITS;//
private static final int STOP = 1 << COUNT_BITS;//
private static final int TIDYING = 2 << COUNT_BITS;//
private static final int TERMINATED = 3 << COUNT_BITS;//1610612736 // Packing and unpacking ctl
private static int runStateOf(int c) { return c & ~CAPACITY; }//当c<0时该方法返回的值为-536870912,否则为0
private static int workerCountOf(int c) { return c & CAPACITY; }//获取工作线程数
private static int ctlOf(int rs, int wc) { return rs | wc; }//-536870912

3.2、execute

当线程为null时,直接抛出异常

线程池的使用及ThreadPoolExecutor的分析(一)

第一步、看图,下图所指的将corePoolSize扩充至maxmumPoolSize是一个类比,
因为在addWorker代码中有这么一句wc >= (core ? corePoolSize : maximumPoolSize))成立则返回false,表明core为false时会以maximumPoolSize来当做corePoolSize比较

线程池的使用及ThreadPoolExecutor的分析(一)

         int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
if (! isRunning(recheck) && remove(command))
reject(command);
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
else if (!addWorker(command, false))
reject(command);

3.3、addWorker

         private boolean addWorker(Runnable firstTask, boolean core) {
//外部循环
retry:
for (;;) {
int c = ctl.get();//获取当前工作线程数量,数量为{c-(-536870912)} int rs = runStateOf(c);//若c>=0时,该值才为0,否则该值一直为-536870912 /*
*由上面的一些线程池状态常量值可知,running<shutdown<stop<tidying<terminated
*若rs>=shutdown,则表明线程池处于stop、tidying、terminated三种状态的一种
*若rs>=shutdown成立,则进行后面判断,
*1、线程池处于shutdown状态
* 1.1、firstTask不为null,则返回false,也即是线程池已经处于shutdown状态,还要添加新的线程,被直接驳回(拒绝)
* 1.2、firstTask为null
* 1.2.1、此时意味着线程池状态为shutdown状态,且first为null,若阻塞队列为空,则返回false
*2、线程处于大于shutdown的状态,则直接返回false
*/
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
/*
*进入内循环以下两种情况会跳出该内循环,否则一直会循环
*1、当工作线程数量超过一定阈值,会直接返回false
*2、添加工作线程成功,即ctl的值进行了加一
*/
for (;;) {
int wc = workerCountOf(c);//获取工作线程的数量
//当线程数量>=536870911或者>=corePoolSize或maximumPoolSize的时候,则返回false
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
if (compareAndIncrementWorkerCount(c))//使用unsafe的cas操作对ctl.get()的值进行加一
break retry;//跳出这个外循环
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)//当此时的线程池状态和之前的状态不等时
continue retry;//继续内循环
}
}
//若进行到了此步操作,则表明工作线程数量加了1
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask);
final Thread t = w.thread;//该w.thread为worker内部新创建的thread
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();//开启锁
try {
//获取锁后,再次获取线程池的状态
int rs = runStateOf(ctl.get());
/*
*1、当线程池的状态处于shutdown以上状态,则直接释放锁,不启动线程,且执行addWorkerFailed方法
执行该方法的作用是使工作线程数量-1
*/
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // 创建的线程处于活跃状态,即被启动了,抛出异常
throw new IllegalThreadStateException();
workers.add(w);//workers是一个set集合
int s = workers.size();
if (s > largestPoolSize)//largestPoolSize默认为0,作用是记录set集合中的线程数量
largestPoolSize = s;
workerAdded = true;//改变该值,为了启动线程,且返回一个addWorker执行成功的状态
}
} finally {
mainLock.unlock();//释放锁
}
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}

总结:2017-11-12

上一篇:MyBatis 框架的搭建和配置


下一篇:解决HttpServletResponse输出的中文乱码问题