关于线程池的源码分析，在这里也没认真说明，后面单独起一片文章进行研究。
1.大纲
线程池介绍
创建与停止线程池
常见的线程池特点与用法
任务太多，怎么拒绝
钩子方法
实现原理，源码分析
使用线程池的主要点
一：介绍
1.重要性
使用中重要
面试中重要
2.池
线程可以复用
可以控制资源的总量
3.不使用线程池些的程序
这里有两个程序，只粘贴进行循环对每个任务进行创建线程，并执行
package com.jun.juc.threadpool;
/**
* for循环执行每一个任务的线程
* 可以正常的执行，但是有些问题
* 开销大，反复的操作系统进行创建与销毁
*/
public class ForLoop {
public static void main(String[] args) {
for (int i = 0; i< 10000; i++){
Thread thread = new Thread(new Task());
thread.start();
}
}
static class Task implements Runnable{
@Override
public void run() {
System.out.println("执行了任务");
}
}
}
4.为什么使用线程池
反复的创建，开销大
让一部分的线程保持工作，反复的执行任务
过多的线程会占用太多的内存
使用少量的线程
5.线程池的好处
加快响应速度
更好的利用CPU，与内存。选择合适的线程数
统一管理
6.使用场景
服务器接收大量的请求
多个线程的创建
二：创建与停止线程池
1.线程池的构造函数的参数
corePoolSize：核心线程数，int
maxPoolSize：最大的线程数，int
keepAliveTime：存活时间，long
workQueue：任务存储队列，BlockingQueue
threadFactory：工厂类，ThreadFactory
Handler：拒绝策略，RejectedExecutionHandler
2.corePoolSize
线程池进行初始化的时候，线程池里没有任何的线程，线程池会等待有任务到来的时候，再进行创建新线程执行任务
3.macPoolSize
线程池有可能子啊核心线程数的基础上，额外增加一些线程，但是这些新增加的线程数有一定的上限，这个就是最大量
如果超过了corePoolSize的时候，先将任务放到队列中。
队列中满了，才会去看
4.添加线程的规则
如果线程小于corePoolSize的时候，即使线程有处于空闲状态，也会继续创建新的线程运行新的任务
如果等于大于corePoolSize，但是小于maxPoolSize，放入队列
如果队列已满，并且线程小于maxPoolSize，创建新的线程
5.keepAliveTime
主要看是控制的是谁。
如果线程池的当前的线程数多于了corePoolSize，那么多于的线程空闲时间超过keepAliveTime，将会被终止
减少资源消耗
6.ThreadFactory
新的线程默认使用Exectors.defaultThreadFactory()，创建的线程都在一个线程组，拥有相同的NORM_PRIORITY优先级，并且都不是守护线程
7.workQueue
工作队列
最常见的队列类型：
直接交换：SynchronousQueue，内部没有容量
无界队列：LinkedBlockingQueue
有界队列：ArrayBlockingQueue
三：常见的线程池
1.FixedThreadPool创建
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
可以看见，corePoolSize与maxPoolSize是相等的
然后使用的是无界队列。
由于传递进去的任务是没有容量上限的，可能占用大量的内存，出现OOM
2.演示溢出
package com.jun.juc.threadpool;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* FixedThreadPool的使用场景
*/
public class FixedThreadPoolTest {
public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(5);
for (int i=0; i<Integer.MAX_VALUE;i++){
executorService.execute(new Task());
}
}
static class Task implements Runnable{
@Override
public void run() {
try {
// 让这个任务执行的很慢，表示队列中会一直增加
Thread.sleep(500000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("执行了任务");
}
}
}
效果：
D:jdk1.8.0_144injava -agentlib:jdwp=transport=dt_socket,address=127.0.0.1:63206,suspend=y,server=n -Xmx8m -Xms8m -Dfile.encoding=UTF-8 -classpath "D:jdk1.8.0_144jrelibcharsets.jar;D:jdk1.8.0_144j Exception in thread "main" java.lang.OutOfMemoryError: GC overhead limit exceeded at java.util.concurrent.LinkedBlockingQueue.offer(LinkedBlockingQueue.java:416) at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1371) at com.jun.juc.threadpool.FixedThreadPoolTest.main(FixedThreadPoolTest.java:13)
3.SingleThreadExector的使用
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
说明：
相比于FixedThreadPool，只是corePoolSize与macPoolSize都是1，其他不变
4.也会出现OOM
package com.jun.juc.threadpool;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class SingleThreadExectos {
public static void main(String[] args) {
ExecutorService executorService = Executors.newSingleThreadExecutor();
for (int i=0; i<Integer.MAX_VALUE;i++){
executorService.execute(new FixedThreadPoolTest.Task());
}
}
static class Task implements Runnable{
@Override
public void run() {
try {
// 让这个任务执行的很慢，表示队列中会一直增加
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("执行了任务");
}
}
}
5.CachedThreadPool
可缓存线程池，无界的线程池，可以自动回收多于线程的功能
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
其中，最大的线程数没有限制，也是一个大的弊端。如果任务量过大，一样会出现的是OOM
6.测试
package com.jun.juc.threadpool;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class CachedThreadPoolTest {
public static void main(String[] args) {
ExecutorService executorService = Executors.newCachedThreadPool();
for (int i=0; i< 1000; i++){
executorService.execute(new Task());
}
}
static class Task implements Runnable{
@Override
public void run() {
try {
// 让这个任务执行的很慢，表示队列中会一直增加
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread()+"执行了任务");
}
}
}
效果：
Thread[pool-1-thread-495,5,main]执行了任务 Thread[pool-1-thread-491,5,main]执行了任务 Thread[pool-1-thread-511,5,main]执行了任务 Thread[pool-1-thread-508,5,main]执行了任务 Thread[pool-1-thread-507,5,main]执行了任务 Thread[pool-1-thread-499,5,main]执行了任务 Thread[pool-1-thread-519,5,main]执行了任务 Thread[pool-1-thread-522,5,main]执行了任务 Thread[pool-1-thread-518,5,main]执行了任务 Thread[pool-1-thread-510,5,main]执行了任务
创建了很多的线程
7.ScheduledThreadPool
支持定时与周期性的执行的线程池
核心线程是传递过去的，但是最大的核心线程数是INTEGER.MAX_VALUE
8.延迟一定时间之后运行
定时
package com.jun.juc.threadpool;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class ScheledThreadPoolTest {
public static void main(String[] args) {
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(10);
scheduledExecutorService.schedule(new Task(),6, TimeUnit.SECONDS);
}
static class Task implements Runnable{
@Override
public void run() {
try {
// 让这个任务执行的很慢，表示队列中会一直增加
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread()+"执行了任务");
}
}
}
9.周期性的运行
周期
package com.jun.juc.threadpool;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class ScheledThreadPoolTest {
public static void main(String[] args) {
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(10);
scheduledExecutorService.schedule(new Task(),6, TimeUnit.SECONDS);
// 周期的运行
scheduledExecutorService.scheduleAtFixedRate(new Task(), 1, 3, TimeUnit.SECONDS);
}
static class Task implements Runnable{
@Override
public void run() {
try {
// 让这个任务执行的很慢，表示队列中会一直增加
Thread.sleep(50);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread()+"执行了任务");
}
}
}
10.总结
主要注意的是CacheDThreadPool与ScheduleThreadPool的对比。
CachedThreadPool为啥使用SynchronousQueue，因为有任务不需要进行存储，直接交给线程执行就行了。
ScheduledThreadPool使用的是延迟队列DelayedWorkQueue
四：线程池中的线程数量设定
1.计算密集型的
为cpu核心数的1~2倍
2.耗时IO型的
最佳线程数一般大于cpu很多倍。
以jvm线程监控显示繁忙情况为依据，参考brain goetz推荐的计算方法
3.计算方法
cpu核心数 * （1+平均等待时间/平均工作时间）
五：停止线程池
1.shutdown
要线程中，会队列中的线程任务都执行完成后，再进行停止
对拒绝新的任务
2.测试
package com.jun.juc.threadpool;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* shutdown关闭
*/
public class ShutDown {
public static void main(String[] args) throws Exception{
ExecutorService executorService = Executors.newSingleThreadExecutor();
for (int i=0; i< 1000; i++){
executorService.execute(new ShutDownTask());
}
Thread.sleep(2000);
executorService.shutdown();
executorService.execute(new ShutDownTask());
}
static class ShutDownTask implements Runnable {
@Override
public void run(){
try {
Thread.sleep(50);
System.out.println(Thread.currentThread().getName());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
效果：
可以发现，在执行一段时间后，就可以发现，真的不再进行接收任务了
pool-1-thread-1 pool-1-thread-1 Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task com.jun.juc.threadpool.ShutDown$ShutDownTask@ed17bee rejected from java.util.concurrent.ThreadPoolExecutor@2a33fae0[Shutting down, pool size = 1, active threads = 1, queued tasks = 960, completed tasks = 39] at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063) at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830) at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379) at java.util.concurrent.Executors$DelegatedExecutorService.execute(Executors.java:668) at com.jun.juc.threadpool.ShutDown.main(ShutDown.java:17) pool-1-thread-1 pool-1-thread-1
3.isShutdown
可以知道线程被停止过了
package com.jun.juc.threadpool;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* shutdown关闭
*/
public class ShutDown {
public static void main(String[] args) throws Exception{
ExecutorService executorService = Executors.newSingleThreadExecutor();
for (int i=0; i< 100; i++){
executorService.execute(new ShutDownTask());
}
Thread.sleep(2000);
System.out.println(executorService.isShutdown());
executorService.shutdown();
System.out.println(executorService.isShutdown());
executorService.execute(new ShutDownTask());
}
static class ShutDownTask implements Runnable {
@Override
public void run(){
try {
Thread.sleep(50);
System.out.println(Thread.currentThread().getName());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
效果：
先false，然后true
4.isTerminated
返回是否真正的结束
package com.jun.juc.threadpool;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* shutdown关闭
*/
public class ShutDown {
public static void main(String[] args) throws Exception{
ExecutorService executorService = Executors.newSingleThreadExecutor();
for (int i=0; i< 10; i++){
executorService.execute(new ShutDownTask());
}
System.out.println(executorService.isShutdown());
executorService.shutdown();
System.out.println(executorService.isShutdown());
Thread.sleep(2000);
System.out.println(executorService.isTerminated());
}
static class ShutDownTask implements Runnable {
@Override
public void run(){
try {
Thread.sleep(50);
System.out.println(Thread.currentThread().getName());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
效果：
D:jdk1.8.0_144injava -agentlib:jdwp=transport=dt_socket,address=127.0.0.1:62572,suspend=y,server=n -Dfile.encoding=UTF-8 -classpath "D:jdk1.8.0_144jrelibcharsets.jar;D:jdk1.8.0_144jrelibdeploy.jarConnected to the target VM, addre true pool-1-thread-1 pool-1-thread-1 pool-1-thread-1 pool-1-thread-1 pool-1-thread-1 pool-1-thread-1 pool-1-thread-1 pool-1-thread-1 pool-1-thread-1 pool-1-thread-1 Disconnected from the target VM, address: '127.0.0.1:62572', transport: 'socket' true Process finished with exit code 0
5.awaitTermination
所有的任务都执行完毕，等待的时间到了，等待过程中被打断都会返回，否则阻塞。
说明：8秒内，如果关闭了线程，并且都执行完成返回true，否则是false
package com.jun.juc.threadpool;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* shutdown关闭
*/
public class ShutDown {
public static void main(String[] args) throws Exception{
ExecutorService executorService = Executors.newSingleThreadExecutor();
for (int i=0; i< 10; i++){
executorService.execute(new ShutDownTask());
}
executorService.shutdown();
boolean b = executorService.awaitTermination(8L, TimeUnit.SECONDS);
System.out.println("b="+b);
}
static class ShutDownTask implements Runnable {
@Override
public void run(){
try {
Thread.sleep(50);
System.out.println(Thread.currentThread().getName());
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
6.shutdownNow
立刻关闭线程
存在返回未执行的任务。
package com.jun.juc.threadpool;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* shutdown关闭
*/
public class ShutDown {
public static void main(String[] args) throws Exception{
ExecutorService executorService = Executors.newSingleThreadExecutor();
for (int i=0; i< 10; i++){
executorService.execute(new ShutDownTask());
}
// shutdownNow
List<Runnable> runnables = executorService.shutdownNow();
ExecutorService executorService2 = Executors.newSingleThreadExecutor();
runnables.forEach(item -> executorService2.execute(item));
}
static class ShutDownTask implements Runnable {
@Override
public void run(){
try {
Thread.sleep(50);
System.out.println(Thread.currentThread().getName());
} catch (InterruptedException e) {
System.out.println("线程终端了");
}
}
}
}
六：拒绝策略
1.拒绝时机
Executor关闭时
最大线程和队列已满
2.拒绝策略
AbsortPolicy：直接抛出异常
DiscradPolicy：默默丢弃
DiscardOldestPolicy：丢弃最老的任务
CallerRunsPolicy：谁提交的任务，则有谁进行运行，这样可以降低提交速度
七：钩子方法
1.说明
在任务的前后
日志，统计
2.暂停线程池
package com.jun.juc.threadpool; import java.util.concurrent.*; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; public class PauseableThreadPool extends ThreadPoolExecutor { /** * 并发加锁 */ private final ReentrantLock lock = new ReentrantLock(); private Condition unpaused = lock.newCondition(); /** * 是否暂停 */ private boolean isPaused; public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue); } public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory); } public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler); } public PauseableThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler); } /** * 执行之前，暂停 * * @param t * @param r */ @Override protected void beforeExecute(Thread t, Runnable r) { super.beforeExecute(t, r); lock.lock(); try { while (isPaused) { unpaused.await(); } } catch (InterruptedException e) { e.printStackTrace(); }finally { lock.unlock(); } } /** * 暂停 */ private void pause() { lock.lock(); try { isPaused = true; } finally { lock.unlock(); } } /** * 恢复 */ public void resume(){ lock.lock(); try { isPaused = false; unpaused.signalAll(); }finally { lock.unlock(); } } public static void main(String[] args) throws InterruptedException { PauseableThreadPool pauseableThreadPool = new PauseableThreadPool(10, 20, 10L, TimeUnit.SECONDS, new LinkedBlockingDeque<>()); Runnable task = new Runnable(){ @Override public void run() { System.out.println("开始执行了"); try { try { Thread.sleep(10); } catch (InterruptedException e) { e.printStackTrace(); } }catch (Exception e){ e.printStackTrace(); } } }; for(int i=0; i<1000; i++){ pauseableThreadPool.execute(task); } Thread.sleep(1000); pauseableThreadPool.pause(); System.out.println("线程池被暂停了"); Thread.sleep(1000); pauseableThreadPool.resume(); System.out.println("线程池被再次执行了"); } }
效果：
开始执行了
开始执行了
线程池被暂停了
线程池被再次执行了
开始执行了
开始执行了
八：源码
1.组成部分
线程池管理器
工作线程
任务队列
任务接口
2.Exector家族
3.Exector
顶层接口，只有一个方法
* @since 1.5
* @author Doug Lea
*/
public interface Executor {
/**
* Executes the given command at some time in the future. The command
* may execute in a new thread, in a pooled thread, or in the calling
* thread, at the discretion of the {@code Executor} implementation.
*
* @param command the runnable task
* @throws RejectedExecutionException if this task cannot be
* accepted for execution
* @throws NullPointerException if command is null
*/
void execute(Runnable command);
}
4.ExecuorService
继承了Excetor，然后增加了几个新的方法
初步的有了管理线程池的方法
5.Excetors
这是一个工具类
进入可以发现是使用ThreadPoolExector进行创建的线程
6.线程池实现任务复用的原理
execute：
添加到worker
進行运行：
九：线程池状态
1.线程池状态
Running：接收新任务并处理排队任务
SHUTDOWN：不接受新任务，但是处理排队任务
stop：不接受新任务，也不处理排队任务，并中断正在进行的任务
tidying：所有的任务都已经终结，workerCount为零时，线程就会转为这个状态，并且运行terminate（）方法
TERMIMATED：运行完成
2.状态值
