Java线程池(Executor)详解和用法

背景

面试的时候经常会被三连问。用过吗？如何用的？场景是什么？所以有必要好好的研究下线程池迫在眉睫。

1、讲解之前先了解下 retry: 因为源码中有这个retry标记

先看一个简单的例子

/*** @author shuliangzhao* @Title: RetryTest* @ProjectName design-parent* @Description: TODO* @date 2019/6/1 23:43*/
public class RetryTest {public static void main(String[] args) {testRetry();}public static void testRetry() {//retry:注释1for (int i = 0; i < 10; i++) {retry: //注释2while (i == 5) {continue retry;}System.out.print(i + " ");}}
}

如上如果只保留注释1，循环到 i==5的时候，程序跳到retry的那一行开始执行，此时 i 的值未变，然后又是i==5，程序进入死循环一直执行4到6行；执行结果为0 1 2 3 4

如果直流注释2，循环到 i==5的时候，程序跳到retry的那一行开始执行，注意此时 i 的值还是5，接着 i++（i 不是从0开始了），所以输出 0 1 2 3 4 6 7 8 9

说明：其实retry就是一个标记，标记程序跳出循环的时候从哪里开始执行，功能类似于goto。retry一般都是跟随者for循环出现，第一个retry的下面一行就是for循环，而且第二个retry的前面一般是 continue或是 break。

2、为什么要使用线程池

缺点
a、每次new Thread新建对象，性能差。
b、缺乏统一管理，可能无限制的新建线程，过多占用系统资源导致死机或OOM
优点
a、重用存在的线程，减少对象创建，消亡的开销
b、有效控制最大并发线程数，提高系统资源利用率

3、线程池实现原理

当线程提交一个任务时候，如果处理请看下图

image.png

ThreadPoolExecutor执行execute()分4种情况

a、若当前运行的线程少于corePoolSize,则创建新线程来执行任务(执行这一步需要获取全局锁)
b、若运行的线程多于或等于corePoolSize,则将任务加入BlockingQueue
c、若无法将任务加入BlockingQueue,则创建新的线程来处理任务(执行这一步需要获取全局锁)
d、若创建新线程将使当前运行的线程超出maximumPoolSize,任务将被拒绝,并调用RejectedExecutionHandler.rejectedExecution()
采取上述思路,是为了在执行execute()时,尽可能避免获取全局锁
在ThreadPoolExecutor完成预热之后（当前运行的线程数大于等于corePoolSize),几乎所有的execute()方法调用都是执行步骤b,而步骤b不需要获取全局锁
源码分析execute()

    /*** Executes the given task sometime in the future.  The task* may execute in a new thread or in an existing pooled thread.** If the task cannot be submitted for execution, either because this* executor has been shutdown or because its capacity has been reached,* the task is handled by the current {@code RejectedExecutionHandler}.** @param command the task to execute* @throws RejectedExecutionException at discretion of*         {@code RejectedExecutionHandler}, if the task*         cannot be accepted for execution* @throws NullPointerException if {@code command} is null*/public void execute(Runnable command) {if (command == null)throw new NullPointerException();/** Proceed in 3 steps:** 1. If fewer than corePoolSize threads are running, try to* start a new thread with the given command as its first* task.  The call to addWorker atomically checks runState and* workerCount, and so prevents false alarms that would add* threads when it shouldn't, by returning false.** 2. If a task can be successfully queued, then we still need* to double-check whether we should have added a thread* (because existing ones died since last checking) or that* the pool shut down since entry into this method. So we* recheck state and if necessary roll back the enqueuing if* stopped, or start a new thread if there are none.** 3. If we cannot queue task, then we try to add a new* thread.  If it fails, we know we are shut down or saturated* and so reject the task.*///表示 “线程池状态” 和 “线程数” 的整数int c = ctl.get();// 如果当前线程数少于核心线程数，直接添加一个 worker 执行任务，// 创建一个新的线程，并把当前任务 command 作为这个线程的第一个任务(firstTask)if (workerCountOf(c) < corePoolSize) {// 添加任务成功，即结束// 执行的结果，会包装到 FutureTask // 返回 false 代表线程池不允许提交任务if (addWorker(command, true))return;c = ctl.get();}// 到这说明，要么当前线程数大于等于核心线程数，要么刚刚 addWorker 失败// 如果线程池处于 RUNNING ，把这个任务添加到任务队列 workQueue 中if (isRunning(c) && workQueue.offer(command)) {/* 若任务进入 workQueue，我们是否需要开启新的线程* 线程数在 [0, corePoolSize) 是无条件开启新线程的* 若线程数已经大于等于 corePoolSize，则将任务添加到队列中，然后进到这里*/int recheck = ctl.get();// 若线程池不处于 RUNNING ，则移除已经入队的这个任务，并且执行拒绝策略if (! isRunning(recheck) && remove(command))reject(command);// 若线程池还是 RUNNING ，且线程数为 0，则开启新的线程// 这块代码的真正意图：担心任务提交到队列中了，但是线程都关闭了else if (workerCountOf(recheck) == 0)addWorker(null, false);}// 若 workQueue 满，到该分支// 以 maximumPoolSize 为界创建新 worker，// 若失败，说明当前线程数已经达到 maximumPoolSize，执行拒绝策略else if (!addWorker(command, false))reject(command);}

其他源码暂不贴出来了，自己可以认真阅读下。

4、线程池创建

我们可以通过ThreadPoolExecutor来创建一个线程池
创建一个线程池时需要的参数
corePoolSize(核心线程数量)
线程池中应该保持的主要线程的数量.即使线程处于空闲状态，除非设置了allowCoreThreadTimeOut这个参数,当提交一个任务到线程池时,若线程数量<corePoolSize,线程池会创建一个新线程放入works(一个HashSet)中执行任务,即使其他空闲的基本线程能够执行新任务也还是会创建新线程,等到需要执行的任务数大于线程池基本大小时就不再创建,会尝试放入等待队列workQueue(一个BlockingQueue),如果调用了线程池的prestartAllCoreThreads(),线程池会提前创建并启动所有核心线程

workQueue
存储待执行任务的阻塞队列，这些任务必须是Runnable的对象（如果是Callable对象，会在submit内部转换为Runnable对象）
runnableTaskQueue(任务队列):用于保存等待执行的任务的阻塞队列.可以选择以下几个阻塞队列.
LinkedBlockingQueue:一个基于链表结构的阻塞队列,此队列按FIFO排序元素,吞吐量通常要高于ArrayBlockingQueue.静态工厂方法Executors.newFixedThreadPool()使用了这个队列
SynchronousQueue:一个不存储元素的阻塞队列.每个插入操作必须等到另一个线程调用移除操作,否则插入操作一直处于阻塞状态,吞吐量通常要高于Linked-BlockingQueue,静态工厂方法Executors.newCachedThreadPool使用了这个队列

maximumPoolSize（线程池最大线程数）
线程池允许创建的最大线程数
若队列满,并且已创建的线程数小于最大线程数,则线程池会再创建新的线程放入works中执行任务,CashedThreadPool的关键,固定线程数的线程池无效
若使用了无界任务队列,这个参数就没什么效果

ThreadFactory:用于设置创建线程的工厂,可以通过线程工厂给每个创建出来的线程设置更有意义的名字.使用开源框架guava提供ThreadFactoryBuilder可以快速给线程池里的线程设置有意义的名字,代码如下
new ThreadFactoryBuilder().setNameFormat("XX-task-%d").build();

RejectedExecutionHandler（饱和策略）:当队列和线程池都满,说明线程池处于饱和,必须采取一种策略处理提交的新任务.策略默认AbortPolicy,表无法处理新任务时抛出异常.在JDK 1.5中Java线程池框架提供了以下4种策略
AbortPolicy：丢弃任务，抛出 RejectedExecutionException
CallerRunsPolicy:只用调用者所在线程来运行任务,有反馈机制，使任务提交的速度变慢）。
DiscardOldestPolicy
若没有发生shutdown,尝试丢弃队列里最近的一个任务,并执行当前任务, 丢弃任务缓存队列中最老的任务，并且尝试重新提交新的任务
DiscardPolicy:不处理,丢弃掉, 拒绝执行，不抛异常
当然,也可以根据应用场景需要来实现RejectedExecutionHandler接口自定义策略.如记录日志或持久化存储不能处理的任务

keepAliveTime（线程活动保持时间）
线程没有任务执行时最多保持多久时间终止
线程池的工作线程空闲后，保持存活的时间。
所以，如果任务很多，并且每个任务执行的时间比较短，可以调大时间，提高线程的利用率

TimeUnit（线程活动保持时间的单位）:指示第三个参数的时间单位；可选的单位有天（DAYS）、小时（HOURS）、分钟（MINUTES）、毫秒（MILLISECONDS）、微秒（MICROSECONDS，千分之一毫秒）和纳秒（NANOSECONDS，千分之一微秒）
可以使用Executors创建线程池

image.png

使用线程池例子

/*** @author shuliangzhao* @Title: ThreadTaskId* @ProjectName design-parent* @Description: TODO* @date 2019/6/1 23:03*/
public class ThreadTaskId implements Runnable {private final int id;public ThreadTaskId(int id) {this.id = id;}@Overridepublic void run() {for (int i = 0;i < 5;i++) {System.out.println("TaskInPool-["+id+"] is running phase-"+i);try {TimeUnit.SECONDS.sleep(1);System.out.println("TaskInPool-["+id+"] is over");} catch (InterruptedException e) {e.printStackTrace();}}}
}

客户端

/*** @author shuliangzhao* @Title: ThreadPoolExample* @ProjectName design-parent* @Description: TODO* @date 2019/6/1 23:03*/
public class ThreadPoolExample {public static void main(String[] args) {ExecutorService executorService = Executors.newSingleThreadExecutor();for (int i = 0; i < 5; i++) {executorService.execute(new ThreadTaskId(i));}executorService.shutdown();}
}

执行结果

image.png

以上就是线程池的简单介绍，这个不是完善版本，会继续补充的。_