Executor框架最核心的类是ThreadPoolExecutor,它是线程池的实现类,主要由下列4个组件构成。
- corePool: 核心线程池的大小
- maximunPoolSize: 最大线程池的大小
- BlockingQueue: 用来暂时保存任务的工作队列
- RejectedExecutionHandler: 当ThreadPoolExecutor已经关闭或ThreadPoolExecutor已经饱和时(达到了最大线程池大小且工作队列已满),execute()方法将要调用的Handler
- 通过Executor框架的工具类Executors,可以创建3种类型的ThreadPoolExecutor
- FixedThreadPool
- SingleThreadPool
- CachedTheadPool
FixedThreadPool 详解
FixedThreadPool被称为可重用固定线程数的线程池。下面是FixedThreadPool的源代码
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
FixedThreadPool的corePoolSize
和maximunPoolSize
都被设置为创建FixedThreadPool时指定的参数nThreads.
当线程池中的线程数大于corePoolSize
时,keepAliveTime
为多余的空闲线程等待新任务的最长时间,超过这个时间后多余的线程将被终止。这里把keepAliveTime
设置为0L,意味着多余的空闲线程会被立即终止。
FixedThreadPool的execute()方法的运行示意图
- 如果当前运行的线程数少于corePoolSize,则创建新线程来执行任务
- 在线程池完成预热之后(当前运行的线程数等于corePoolSize),将任务加入LinkedBlockingQueue。
- 线程执行完1中的任务后,会在循环中反复从LinkedBlockingQueue获取任务来执行。
FixedThreadPool使用无界队列LinkedBlockingQueue作为线程池的工作队列(队列的容量为Integer.MAX_VALUE)。使用无界队列作为工作队列会对线程池带来如下影响。 - 当线程池中的线程数达到
corePoolSize
后,新任务将在无界队列中等待,因此线程池中的线程数不会超过corePoolSize
。 - 由于1,使用无界队列时
maximumPoolSize
将是一个无效参数。 - 由于1和2,使用误解队列时
keepAliveTime
将是一个无效参数 - 由于使用无界队列,运行中的
FixedThreadPool
(未执行方法shutdown()或shutdownNow())不会拒绝任务(不会调用RejectedExecutionHandler.rejectedExecution
方法)。
SingleThreadExecutor详解
SingleThreadExecutor是使用单个worker线程的Executor。下面是SingleThreadExecutor的源代码实现。
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
SingleThreadExecutor的 corePoolSize
和maximumPoolSize
被设置为1。其他参数与FixedThreadPool相同。SingleThreadExecutor
使用无界队列LinkedBlockingQueue
作为线程池的工作队列(队列的容量为Integer.MAX_VALUE).SingleThreadExecutor使用无界队列对线程池带来的影响与FixedThreadPool
相同。
singleThreadExecutor的运行示例图
参数说明
- 如果当前运行的线程数少于
corePoolSize
(即线程池中无运行的线程),则创建一个新线程来执行任务。 - 在线程池完成预热之后(当前线程池中有一个运行的线程),将任务加入LinkedBlockingQueue.
- 线程执行完1中的任务后,会在一个无限循环中反复从LinkedBlockingQueue获取任务来执行。
CachedThreadPool详解
CachedThreadPool是一个会根据需要创建新线程的线程池。下面是创建CachedThreadPool
的源码
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
CachedThreadPool的corePoolSize被设置为0,即corePool为空;maximumPoolSize被设置为Integer.MAX_VALUE,即maximumPool是无界的。这里把keepAliveTime
设置为60L,意味着CachedThreadPool中的空闲线程等待新任务的最长时间为60s,空闲线程超过60s后将会被终止。
FixedThreadPool和SingleThreadExecutor使用无界队列LinkedBlockingQueue作为线程池的工作队列。CachedThreadPool使用没有容量的SynchronousQueue作为线程池的工作队列,但CachedThreadPool的maximunPool是无界的。这意味着,如果主线程提交任务的速度高于maximunPool中线程处理任务的速度时,CachedThreadPool会不断创建新线程。极端情况下,CachedThreadPool会因为创建过多线程而耗尽CPU和内存资源。
CachedThreadPool的executor()方法的执行示意图
- 首先执行SynchronousQueue.offer(Runnable task)。如果当前maximumPool中有空闲线程正在执行SynchronousQueue.poll(keepAliveTime,TimeUnit.NANOSECONDS),那么主线程执行offer操作与空闲线程执行的poll操作配对成功,主线程把任务交给空闲线程执行,execute()方法执行完成;否则执行下面的步骤2
- 当初始maximumPool为空,或者maximumPool中当前没有空闲线程时,将没有线程执行SynchronousQueue.poll(KeepAliveTime,TimeUnit.NANOSECONDS).这种情况下,步骤1将失败。此时CachedThreadPool会创建一个新线程执行任务,execute()方法执行完成。
-
在步骤2中新创建的线程将新任务执行完后,会执行SynchronousQueue.poll(keepAliveTime,TimeUnit.NANOSECONDS).这个poll操作会让空闲线程最多在SynchronousQueue中等待60s,如果60s内主线程提交一个新任务(主线程执行步骤1),那么这个空闲线程将执行主线程提交的新任务;否则,这个空闲线程将终止。由于空闲60s的空闲线程会被终止,因此长时间保持空闲的CachedThreadPool不会使用任何资源。
SynchronousQueue是一个没有容量的阻塞队列。每个插入操作必须等待另一个线程的对应移除操作,反之亦然。CachedThreadPool使用SynchronousQueue,把主线程提交的任务传递给空闲线程执行。CachedThreadPool中任务传递的示意图如图
14.png
ScheduledThreadPoolExecutor详解
ScheduledThreadPoolExecutor继承自ThreadPoolExecutor。它主要用来在给定的延迟之后运行任务,或者定期执行任务。SheduledThreadPoolExecutor的功能与Timer类似,但ScheduledTheadPoolExecutor功能更强大、更灵活。Timer对应的是单个后台线程,而ScheduledThreadPoolExecutor可以在构造函数中指定多个对应的后台线程数。
ScheduledThreadPoolExecutor的执行示意图
15.png
DelayQueue是一个无界队列,所以ThreadPoolExecutor的maximumPoolSize在ScheduledThreadPoolExecutor中没有什么意义(设置maximumPoolSize的大小没有什么效果).
ScheduledThreadPoolExecutor的执行主要分为两大部分。
- 当调用ScheduledThreadPoolExecutor的scheduleAtFixedRate()方法或者scheduleWithFixedDelay()方法时,会向ScheduledThreadPoolExecutor的DelayQueue添加一个实现了RunnableScheduledFuture接口的ScheduledFutureTask。
- 线程池中的线程从DelayQueue中获取ScheduledFutrueTask,然后执行任务。
ScheduledThreadPoolExecutor会把待调度的任务(ScheduledFutureTask)放在一个DelayQueue中。
ScheduledFutureTask主要包含3个成员变量,如下: - long型成员变量time,表示这个任务将要被执行的具体时间。
- long型成员变量sequenceNumber,表示这个任务被添加到ScheduledThreadPoolExecutor中的序号
-
long型成员变量period,表示任务执行的间隔周期
DelayQueue封装了一个PriorityQueue,这个PriorityQueue会对队列中的ScheduledFutureTask进行排序。排序时,time小的排在前面(时间早的任务将被先执行)。如果两个ScheduledFutureTask的time相同,就比较sequenceNumber,sequenceNumber小的排在前面(也就是说,如果两个任务的执行时间相同,那么先提交的任务将被先执行)
首先,让我们看看ScheduledThreadPoolExecutor中的线程执行周期任务的过程
16.png - 线程1从DelayQueue中获取已到期的ScheduledFutureTask(DelayQueue.take())。到期任务是指ScheduleFutureTask的time大于等于当前时间。
- 线程1执行这个ScheduledFutureTask.
- 线程1修改ScheduledFutureTask的time变量为下次将要被执行的时间
- 线程1把这个修改time之后的ScheduledFutureTask放回DelayQueue中
网友评论