JUC的线程池架构
1.Executor
Executor是Java异步任务的执行者接口,目标是执行目标任务。Executor作为执行者角色,目的是提供一种将“任务提交者”与“任务执行者”分离的机制。它只有一个函数式方法:
public interface Executor {
void execute(Runnable command);
}
2.ExecutorService
ExecutorService继承于Executor。它对外提供异步任务的接收服务。ExecutorService提供了“接受异步任务并转交给执行者”的方法,比如submit、invoke方法等。具体如下:
public interface ExecutorService extends Executor {
void shutdown();
List<Runnable> shutdownNow();
boolean isShutdown();
boolean isTerminated();
boolean awaitTermination(long timeout, TimeUnit unit)
throws InterruptedException;
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException;
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
3.AbstractExecutorService
AbstractExecutorService是一个抽象类,它实现了ExecutorService接口。AbstractExecutorService存在的目的是为ExecutorService中的接口提供默认实现。(模板模式)
4.ThreadPoolExecutor
大名鼎鼎的线程池实现类,继承于AbstractExecutorService。它是核心实现类,它可以预先提供指定数量的可重用线程,可以对线程进行管理和监控。
5.ScheduledExecutorService
她继承于ExecutorService。是一个完成延时和周期性任务的接口。
6.Executors
是一个静态工厂类,内置的静态工厂方法可以理解为快捷创建线程池的方法。
Executors的4种快捷创建线程池的方法
newSingleThreadExecutor 创建只有一个线程的线程池
newFixedThreadPool 创建固定大小的线程池
newCachedThreadPool 创建一个不限制线程数量的线程池,任何提交的任务都立即执行,空闲线程会及时回收
newScheduledThreadPool 创建一个可定期或延时执行任务的线程池
- newSingleThreadExecutor
public static void main(String[] args) {
final AtomicInteger integer = new AtomicInteger(0);
ExecutorService pool = Executors.newSingleThreadExecutor();
for (int i = 0; i < 5; i++) {
pool.execute(() -> {
System.out.println(Thread.currentThread() + " :doing" + "-" + integer.incrementAndGet());
try {
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
pool.shutdown();
}
Thread[pool-1-thread-1,5,main] :doing-1
Thread[pool-1-thread-1,5,main] :doing-2
Thread[pool-1-thread-1,5,main] :doing-3
Thread[pool-1-thread-1,5,main] :doing-4
Thread[pool-1-thread-1,5,main] :doing-5
场景:任务按照提交顺序,一个任务一个任务逐个执行。
以上代码最后调用shutdown来关闭线程池。执行shutdown方法后,线程池状态变为shutdown,线程池将拒绝新任务,不能再往线程池中添加新任务。此时,线程池不会立刻退出,直到线程池中的任务处理完成后才会退出。还有一个shutdownNow方法,执行这个后,线程状态变为stop,试图停止所有正在执行的线程,并且不再处理阻塞队列中等待的任务,会返回那些未执行的任务。
- newFixedThreadPool
ExecutorService pool = Executors.newFixedThreadPool(3);
Thread[pool-1-thread-1,5,main] :doing-1
Thread[pool-1-thread-3,5,main] :doing-2
Thread[pool-1-thread-2,5,main] :doing-3
Thread[pool-1-thread-3,5,main] :doing-4
Thread[pool-1-thread-1,5,main] :doing-5
适用场景:需要任务长期执行的场景。“固定数量的线程池”能稳定的保证一个数,避免频繁 回收和创建线程,适用于CPU密集型的任务,在CPU被线程长期占用的情况下,能确保少分配线程。
弊端:内部使用*队列存放任务,当有大量任务,队列无限增大,服务器资源迅速耗尽。
newFixedThreadPool工厂方法返回一个ThreadPoolExecutor实例,该线程池实例的corePoolSize数量为参数nThread,其maximumPoolSize数量也为参数nThread,其workQueue属性的值为LinkedBlockingQueue
- newCachedThreadPool
线程池内的某些线程无事可干成为空闲线程,可以灵活回收这些空闲线程。
ExecutorService pool = Executors.newCachedThreadPool();
Thread[pool-1-thread-5,5,main] :doing-5
Thread[pool-1-thread-1,5,main] :doing-1
Thread[pool-1-thread-2,5,main] :doing-2
Thread[pool-1-thread-3,5,main] :doing-3
Thread[pool-1-thread-4,5,main] :doing-4
特点:在执行任务时,如果池内所有线程忙,则会添加新线程来处理。不会限制线程的大小,完全依赖于操作系统能够创建的最大线程大小。如果存量线程超过了处理任务数量,就会回收线程。、
适用场景:快速处理突发性强、耗时短的任务场景,如Netty的NIO处理场景、REST API接口的瞬时削峰场景。
弊端:没有最大线程数量限制,如果大量的异步任务提交,服务器资源可能耗尽。
- newScheduledThreadPool
public static void main(String[] args) {
final AtomicInteger integer = new AtomicInteger(0);
ScheduledExecutorService pool = Executors.newScheduledThreadPool(5);
for (int i = 0; i < 5; i++) {
pool.scheduleAtFixedRate(
() -> {
System.out.println(Thread.currentThread() + " :doing" + "-" + integer.incrementAndGet());
}, 0, 500, TimeUnit.MILLISECONDS);
// 0表示首次执行任务的执行时间,500表示每次执行任务的间隔时间
}
// pool.shutdown();
}
因为可以周期性执行任务,所以不shutdown。
适用场景:周期性执行任务的场景。
线程池的标准创建方式
使用ThreadPoolExecutor构造方法创建,一个比较重要呃构造器如下:
public ThreadPoolExecutor(int corePoolSize,核心线程数
int maximumPoolSize, 最大线程数
long keepAliveTime, TimeUnit unit, 空闲时间
BlockingQueue<Runnable> workQueue, 阻塞队列
ThreadFactory threadFactory, 线程工厂(线程产生方式)
RejectedExecutionHandler handler 拒绝策略) {
...
}
1.核心和最大线程数量
接收新任务时,并且当前工作线程池数少于核心线程数量,即使有工作线程是空闲的,它也会创建新线程处理任务,直到达到核心线程数。
2.BlockingQueue
阻塞队列用于暂时接收任务。
3.KeepAliveTime
设置线程最大空闲时长,如果超过这个时间,非核心线程会被回收。当然,也可以调用allowCoreThreadTimeOut方法将超时策略应用到核心线程。
线程池的任务调度流程
- 工作线程数量小于核心线程数量,执行新任务时会优先创建线程,而不是获取空闲线程。
- 任务数量大于核心线程数量,新任务将被加入阻塞队列中。执行任务时,也是先从阻塞队列中获取任务。
- 在核心线程用完,阻塞队列已满的情况下,会创建非核心线程处理新任务。
- 在如果线程池总数超过maximumPoolSize,线程池会拒绝接收任务,为新任务执行拒绝策略。
ThreadFactory(线程工厂)
创建线程方式
阻塞队列
阻塞队列与普通度列相比:阻塞队列为空时,会阻塞当前线程的元素获取操作。当队列中有元素,被阻塞的线程会被自动唤醒。
BlockingQueue是JUC包的一个超级接口,比较常用的实现类有:
(1)ArrayBlockingQueue:数组队列
(2)LinkedBlockingQueue:链表队列
(3)PriorityBlockingQueue:优先级队列
(4)DelayQueue:延迟队列
(5)SynchronousQueue:同步队列
调度器的钩子方法
ThreadPoolExecutor为每个任务执行前后都提供了钩子方法。
// 任务执行之前的钩子方法(前钩子)
protected void beforeExecute(Thread t, Runnable r) { }
// 之后(后钩子)
protected void afterExecute(Runnable r, Throwable t) { }
// 终止(停止钩子)
protected void terminated() { }
beforeExecute:可用于重新初始化ThreadLocal线程本地变量实例、更新日志记录、计时统计等。
afterExecute:更新日志记录、计时统计等。
terminated:Executor终止时调用。
演示一下前钩子。
public class TestMain {
public static void main(String[] args) {
final ThreadPoolExecutor pool = new ThreadPoolExecutor(
2,
4,
60, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(2)) {
@Override
protected void beforeExecute(Thread t, Runnable r) {
System.out.println("前钩子嗷 ~ ~ ~ ");
}
};
for (int i = 0; i < 5; i++) {
pool.execute(() -> {
System.out.println("你谁啊");
});
}
}
}
线程池拒绝策略
任务被拒绝有两种情况:
- 线程池已经关闭。
- 工作队列已满且最大线程数已满。
拒绝策略有以下实现:
- AbortPolicy:拒绝策略。抛异常。
- DiscardPolicy:抛弃策略。丢弃新来的任务。
- DiscardOldestPolicy:抛弃最老任务策略。因为队列是队尾进对头出,所以每次都是移除队头元素后再入队。
- CallerRunsPolicy:调用者执行策略。提交任务线程自己执行任务,不使用线程池中的线程。
- 自定义策略。实现RejectExecutionHandler接口的rejectedExecution方法。
Re
《Java高并发编程》
(之后补源码)