线程池是很常用的并发框架,几乎所有需要异步和并发处理任务的程序都可用到线程池。
使用线程池的好处如下:
- 降低资源消耗:可重复利用已创建的线程池,降低创建和销毁带来的消耗;
- 提高响应速度:任务到达时,可立即执行,无需等待线程创建;
- 提高线程的可管理性:线程池可对线程统一分配、调优和监控。
原理
线程池的原理非常简单,这里用处理流程来概括:
- 线程池判断核心池里的线程是否都在执行任务,如果不是,创建一个新的线程来执行任务;
- 如果核心线程池已满,则将新任务存在工作队列中;
- 如果工作队列满了,线程数量没有达到线程池上限的前提下,新建一个线程来执行任务;
- 线程数量达到上限,则触发饱和策略来处理这个任务;
使用工作队列,是为了尽可能降低线程创建的开销。工作队列用阻塞队列来实现。
阻塞队列
阻塞队列(BlockingQueue)是指支持阻塞的插入和移除元素的队列。
- 阻塞的插入:当队列满时,阻塞插入元素的线程,直到队列不满;
- 阻塞的移除:当队列为空,阻塞移除元素的线层,直到队列不为空;
原理:使用通知者模式实现。当生产者往满的队列中添加元素时,会阻塞生产者。消费者移除元素时,会通知生产者当前队列可用。
阻塞队列有以下三种类型,分别是:
- 有界阻塞队列:ArrayBlockingQueue(数组),LinkedBlockingQueue(链表)
- 无界阻塞队列:LinkedTransferQueue(链表),PriorityBlockingQueue(支持优先级排序),DelayQueue(支持延时获取元素的无界阻塞队列)
- 同步移交队列:SynchronousQueue
有界阻塞队列
主要包括ArrayBlockingQueue(数组),LinkedBlockingQueue(链表)两种。有界队列大小与线程数量大小相互配合,队列容量大线程数量小时,可减少上下文切换降低cpu使用率,但是会降低吞吐量。
无界阻塞队列
比较常用的是LinkedTransferQueue。FixedThreadPool就是用这个实现的。无界阻塞队列要慎重使用,因为在某些情况,可能会导致大量的任务堆积到队列中,导致内存飙升。
同步移交队列
SynchronousQueue。不存储元素的阻塞队列,每一个put操作必须等待一个take操作,否则不能继续添加元素。用于实现CachedThreadPool线程池。
各个线程池所使用的任务队列映射关系如下:
实现类分析
ThreadPoolExecutor是Java线程池的实现类,是Executor接口派生出来的最核心的类。依赖关系图如下:
这里不得不提到Executor框架,该框架包含三大部分,如下:
- 任务。被执行任务需要实现的接口:Runnable和Callable;
- 任务执行。即上述核心接口Executor以及继承而来的ExecutorService。ExecutorService派生出如下两个类:
- ThreadPoolExecutor:线程池核心实现类;
- ScheduledThreadPoolExecutor:用来做定时任务;
- 异步计算的结果。接口Future和实现Future接口的FutureTask类。
线程池创建
new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, milliseconds, runnableTaskQueue, handler)
构造方法如下:
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
参数说明:
- corePoolSize:核心池的线程数量;
- workQueue:用于保存任务的工作队列;
- maximumPoolSize:最大线程池的大小;
- keepAliveTime:当线程数量大于核心池线程数量时,keepAliveTime为多余的空闲线程等待新任务的最长时间,超过这个时间,多余的线程会被终止;
- TimeUnit:keepAliveTime的单位;
- ThreadFactory:线程工厂,可以给线程设置名字;
- handler:饱和策略。当队列和线程池都满了,会触发饱和策略,来处理新提交的任务。饱和策略以下几种:
- AbortPolicy:直接抛出异常;
- CallerRunsPolicy:只用调用者所在线程来运行任务;
- DiscardOldestPolicy:丢弃最近一个任务并执行当前任务;
- DiscardPolicy:不处理,丢弃掉。
使用Executors创建线程池
使用工具类Executors可创建三种类型的线程池:FixedThreadPool、SingleThreadExecutor、CachedThreadPool。本质上也是调用上述构造方法。理解了前文的参数解释,下面三种线程池也就容易理解了。
- FixedThreadPool
可重用固定线程数的线程池。
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
工作流程如下:
- 如果当前运行的线程数少于corePoolSize,则创建新线程来执行任务;
- 线程数等于corePoolSize之后,新任务加入LinkedBlockingQueue(无界阻塞队列)。因为最大线程数maximumPoolSize参数值等于corePoolSize,不会产生多余线程;
- 线程执行完任务之后会反复从LinkedBlockingQueue中获取任务来执行。
- SingleThreadExecutor
单个worker线程的线程池
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
SingleThreadExecutor与FixedThreadPool的区别在于,maximumPoolSize和corePoolSize都设置成了1,其它参数都一样。
- CachedThreadPool
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
CachedThreadPool将corePoolSize设置为0,maximumPoolSize设置为无限大,同时使用了一个没有容量的工作队列SynchronousQueue。这个线程池没有固定的核心线程,而是根据需要创建新线程。
工作流程:
- 有新任务时,主线程执行SynchronousQueue.offer操作,空闲线程执行SynchronousQueue.poll(keepAliveTime,TimeUnit.NANOSECONDS)操作,配对成功则将任务交给空闲线程执行;
- 当没有空闲线程时,上面的配对操作失败,此时会创建一个新线程来执行任务;
- 任务执行完毕后,空闲线程会等待60秒。60秒内如果有新任务,就立即执行,否则时间一过线程就终止。
线程池关闭
调用shutdown或者shutdownNow方法可关闭线程池。原理是遍历线程池中所有工作线程,调用interrupt方法来中断线程。
- shutdown:将线程置为SHUTDOWN状态,不能接受新的任务,等待所有任务执行完毕;
- shutdownNow:将线程置为STOP状态,不能接受新的任务,尝试去终止正在执行的恶任务;
这里涉及到ThreadPoolExecutor中定义的线程的五种状态
// runState is stored in the high-order bits
private static final int RUNNING = -1 << COUNT_BITS;
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;
- RUNNING:接受新任务,处理任务;
- SHUTDOWN:不接受新任务,但会把队列中任务处理完;
- STOP:不接受新任务,不处理队列中的任务,并且终止正在处理的任务;
- TIDYING:正在执行的任务和队列都为空,进入该状态,将要执行terminated();
- TERMINATED:所有terminated()方法执行完毕,线程池彻底终止。
当队列和正在执行的任务都为空时,由SHUTDOWN转化为TIDYING;当正在执行的任务为空,由STOP转化为TIDYING。
本博客从线程池的原理介绍作为切入点,分析了线程池中尤为关键的组件:阻塞队列。同时分析了线程池的核心实现类ThreadPoolExecutor。以线程池的创建和关闭的思路,梳理了相关知识点,包括三种常用线程池介绍以及线程池五种状态。