Java线程2-线程池

1，简介

1）线程池是运用最多的并发框架，几乎所有的异步操作或者并发执行任务都可以使用线程池。
2）优势。
降低资源消耗 创建线程需要分配线程私有数据 如栈、pc程序计数器等
提高响应速度 任务到达时，不需要重新创建线程，可立即执行
提高线程可管理性 统一分配、调优、监控线程

image.png

2，线程池实现原理

1）线程池构造函数
corePoolSize：核心线程数 the number of threads to keep in the pool even if they are idle
maximumPoolSize: 线程池允许的最大线程数the maximum number of threads to allow in the pool(如果使用了无界队列，此参数无效)
keepAliveTime：大于core线程数的线程，空闲的最大时间
unit: keepAliveTime的单位
workQueue：工作队列，保存将要被执行的Runnable tasks。the queue to use for holding tasks before they are executed.
threadFactory: 线程工厂，线程池创建线程时使用。the factory to use when the executor creates a new thread
handler：RejectedExecutionHandler拒绝执行处理器。当workQueue满并且线程数达到maximunPoolSize时，执行拒绝执行的处理策略

image.png
2）线程池执行任务流程
execute方法创建一个线程并执行当前task，之后循环从BlockingQueue取task来执行。
step1：线程池当前运行的线程数<corePoolSize，则创建新的Worker执行线程。需要获取全局锁
step2：运行线程数>= corePoolSize，将任务加入BlockingQueue中(无界队列会导致JVM频繁发生minor gc，然后old区域满进而发生full gc，当多次full gc old区域仍然没有内存，则发生OOM。)。不需要获取全局锁，ThreadPoolExecutor在execute任务时，尽可能避免获取全局锁。
step3：BlockingQueue满，线程数未达到maxPoolSize时，创建新的Worker执行任务。需要获取全局锁
step4：当BlockingQueue满，并且线程数达到maxPoolSize时，则执行RejectedExecutionHandler的rejectedExecution方法。拒绝策略
image.png
image.png
execute(Runnable r)源码
image.png
3）线程池创建线程执行任务时，将Runnable封装成Worker，workers.add(w)Worker执行完任务后，还会循环获取workQuene中的任务来执行。
workerAdded后，启动worker线程
image.png
image.png
创建Worker时，通过ThreadFactory创建Thread线程 image.png
Worker的run()方法。runWorker(this);
image.png
线程循环从workQueue中获取task
image.png
取不到task时，将线程放回线程池
image.png

3，线程池使用

1）BlockingQueue任务队列主要分为有界队列和无界队列，建议使用有界队列，方便系统稳定性和预警能力
ArrayBlockingQueue final Object[] items;：基于数组的有界队列。A bounded BlockingQueue backed by an array
LinkedBlockingDeque transient Node<E> first;：基于链表的无界队列。An optionally-bounded BlockingDeque based on linked nodes.

image.png
SynchronousQueue ：不存元素的阻塞队列，每次插入必须等到另一个线程调用移除消费操作，否则插入阻塞。
image.png
PriorityBlockingQueue：具有优先级的无界阻塞队列。
2）RejectedExecutionHandler饱和策略
AbortPolicy: 拒绝任务，抛出异常。rejected tasks that throws a RejectedExecutionException
DiscardPolicy：丢弃该任务。silently discards the rejected task.
DiscardOldestPolicy：丢弃最老的未执行的任务。discards the oldest unhandled task
CallerRunsPolicy：使用调用者线程来执行task。
3）向线程池提交任务
void execute(Runnable command); Executor接口定义，提交不需要返回值的任务，所以无法判断是否被线程池执行成功
Future<?> submit(Runnable task); ExecutorService接口定义，提交需要返回值的任务，会返回Future类型的对象。可以使用future.get()阻塞直到返回，或者get(long timeout, TimeUnit unit)超时抛出异常
4）关闭线程池。原理是遍历所有Worker，逐个调用线程的interrupt方法来中断线程
shutdown()：设置线程池状态为SHUTDOWN，然后中断idle的线程
image.png
image.png
shutdownNow(): 设置线程池状态为SHUTDOWN，然后尝试中断所有的线程
image.png
image.png
5）线程池的监控。如果系统中大量使用线程池，则需要对线程池进行监控
扩展ThreadPoolExecutor重写beforeExecute、afterExecute、terminated方法。
beforeExecute：任务执行前 Method invoked prior to executing the given Runnable in the given thread.
afterExecute: 任务执行后 Method invoked upon completion of execution of the given Runnable.
terminated：线程池关闭 Method invoked when the Executor has terminated.
image.png