线程池:管理一组工作线程的资源池。
1.避免反复创建回收线程,降低资源消耗。 2.提供线程的可管理性。 3.提高响应速度
ThreadPoolExecutor是jdk提供的线程池的服务类,基于ThreadPoolExecutor可以很容易将一个实现Runnable接口的任务放入线程池中执行,下面是ThreadPoolExecutor实现:
//构造函数:
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
创建一个线程池需要以下几个参数:
当客户端提交一个新任务时,如果此时线程池中线程的个数小于corePoolSize,则线程池会新建一个线程来处理该任务,即时此时有空闲线程。
客户端提交新任务,此时线程池的任务队列也已经满了,那么如果maximumPoolSize < corePoolSize,就新建线程执行该任务。
如果线程池用的是无界队列,那么这个参数也就不起任何作用了。
线程池中超过corePoolSize的线程会在空闲keepAliveTime时间后被关闭,keepAliveTime单位由TimeUnit指定,如果allowCoreThreadTimeOut=true,即核心线程如果空闲时间超过keepAliveTime时间后同样会被关闭。
当核心线程池已满,即当前worker数=corePoolSize时,新提交的任务会被放入工作队列中。此时一旦有worker完成手头的任务就会到workQueue中领取一个新任务继续执行。 工作队列可以有以下几种选择:
每当线程池需要创建一个线程时,都是通过线程工厂方法来完成。默认的线程工厂方法将创建一个新的、非守护的线程,并且不包含特殊配置信息。通过指定一个线程工厂方法,可以定制线程池的配置信息。 我们可以通过实现ThreadFactory接口,来定制线程工厂。样例如下:
public interface ThreadFactory {
Thread newThread(Runnable r);
}
public class MyThreadFactory implements ThreadFactory {
private final String poolName;
public MyThreadFactory(String name){
this.poolName = name;
}
@Override
public Thread newThread(Runnable r) {
// TODO Auto-generated method stub
return new MyAppThread(r,poolName);
}
}
public class MyAppThread extends Thread {
public static final String DEFAULT_NAME = "MyAppThread";
private static volatile boolean debugLifecycle = false;
private static final AtomicInteger created = new AtomicInteger();
private static final AtomicInteger alive = new AtomicInteger();
public MyAppThread(Runnable r){
this(r,DEFAULT_NAME);
}
public MyAppThread(Runnable r,String name){
super(r,name + "-" + created.incrementAndGet());
setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
@Override
public void uncaughtException(Thread t, Throwable e) {
System.out.println("UNCATGHT in Thread " + t.getName());
}
});
}
public void run(){
try {
alive.incrementAndGet();
super.run();
} finally{
alive.decrementAndGet();
}
}
public static int getThreadsCreated(){return created.get();}
public static int getThreadAlive(){return alive.get();}
public static boolean getDebug(){return debugLifecycle;}
}
当线程池达到最大容量,饱和策略就发挥作用,ThreadPoolExecutor的饱和策略可以通过setRejectedExecutionHandler方法来修改。 JDK提供几种不同的RejectedExecutionHandler的实现:
饱和策略实现类都需要实现RejectedExecutionHandler接口,下面是四种jdk内置的四种饱和策略的源码:
public static class AbortPolicy implements RejectedExecutionHandler {
public AbortPolicy() { }
// 直接抛出RejectedExecutionException异常
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}
public static class CallerRunsPolicy implements RejectedExecutionHandler {
public CallerRunsPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
public static class DiscardPolicy implements RejectedExecutionHandler {
public DiscardPolicy() { }
//不做操作,直接丢弃了
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
public DiscardOldestPolicy() { }
//抛弃队头的任务,即最早提交且未执行的任务
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}
创建线程池
jdk为我们提供了一个工厂类Executors,其中提供了几个静态工厂方法用于新建不同特性的线程池。如下:
public class Executors {
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
public static ScheduledExecutorService newScheduledThreadPool(
int corePoolSize, ThreadFactory threadFactory) {
return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}
注意事项
1.newFixedThreadPool和newSingleThreadExecutor都是用了LinkedBlockingQueue(),默认capacity=Integer.MAX_VALUE,线程池工作队列可以认为是无限大的,所以线程池中的线程数不会超过CorePoolSize,maximumPoolSize可以认为是一个无效参数,且饱和策略不可能执行,这几点需要注意。
2.newFixedThreadPool(1)和newSingleThreadPool区别?
newSingleThreadPool返回的是一个代理对象,屏蔽了ThreadPoolExecutor的一些set方法,即newSingleThreadPool一旦返回,就无法在重新配置线程池的参数了。
3.CachedThreadPool的corePoolSize=0,即核心线程池默认为空,maximumPoolSize=Integer.MAX_VALUE,最大线程池为无限大的。且空闲线程等待新任务超过60秒即被终止。
由于Executor以异步方式来执行任务,因此在任意时刻,之前提交的任务的状态是无法立刻得到的。有些任务可能已经完成,有些可能正在运行,而其他的任务可能在队列中等待执行。
为了解决执行任务的生命周期问题,ExecutorService扩展了Executor接口,添加了一些生命周期管理的方法。如下:
void shutdown();
List<Runnable> shutdownNow();
boolean isShutdown();
boolean isTerminated();
boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException;
// ...还有用于提交任务的一些方法
线程池的生命周期有以下几种状态
线程池如何存储自身状态的?
线程池的状态信息是用一个AtomicInteger类型的变量ctl存储的,定义如下:
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
ctl中除了存放状态信息,还存放了线程池当前工作线程的个数信息。下图展示这两个信息在ctl中的存储形式:
下面是状态相关信息的源码,结合上图应该就不难理解了
private static final int COUNT_BITS = Integer.SIZE - 3;//为啥减3
//0-28位全为1
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// RUNNING : 111。。。
private static final int RUNNING = -1 << COUNT_BITS;
// SHUTDOWN : 000。。。
private static final int SHUTDOWN = 0 << COUNT_BITS;
// STOP : 001。。。
private static final int STOP = 1 << COUNT_BITS;
// TIDYING : 010。。。
private static final int TIDYING = 2 << COUNT_BITS;
//TERMINATED :110。。。
private static final int TERMINATED = 3 << COUNT_BITS;
// 通过简单的位运算获取ctl中的信息
private static int runStateOf(int c) { return c & ~CAPACITY; }
private static int workerCountOf(int c) { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }
runState有5种状态,所以最少需要3bit来表示。这就是为什么COUNT_BITS=32-3
各种状态之间的转换时机:
关闭线程池方法
线程池处理新提交任务的流程如下:
ThreadPoolExecutor 中execute源码:如下:
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
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);
}
//线程池已完成任务数量
private long completedTaskCount;
//当前运行的线程数
public int getActiveCount()
此外,ThreadPoolExecutor还提供了以下几个钩子函数用于扩展它的行为,我们可以在子类中实现自己的逻辑,在每个任务执行的前、后以及worker退出时进行定制处理。
protected void beforeExecute(Thread t, Runnable r) { }
protected void afterExecute(Runnable r, Throwable t) { }
protected void terminated() { }
下面从线程池中的某个线程的角度出发,分析一下线程从被创建一直到被销毁,整个生命周期里的工作流程
public int prestartAllCoreThreads() {
int n = 0;
while (addWorker(null, true))//firstTask为null
++n;
return n;
}
线程池中的工作线程是被封装到一个Worker类中,部分源码如下:
private final class Worker extends AbstractQueuedSynchronizer
implements Runnable
{
/** worker关联的线程 */
final Thread thread;
/** worker的第一个任务,可能为null */
Runnable firstTask;
/** worker完成的任务数量 */
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
/** 将任务代理给runWorker方法 */
public void run() {
runWorker(this);
}
。。。。
}
//下面是ThreadPoolExecutor中的runWorker方法源码:
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) {
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);//钩子函数
Throwable thrown = null;
try {
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);//钩子函数
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
当firstTassk为null的情况下,线程的执行流程如下图
对于Executor框架,需要明白以下两点
Executor框架基于生产者-消费者模式:提交任务的执行者是生成者,执行任务的线程是消费者。
Executor是异步执行任务,这是通过队列来实现的。