java - 多线程 07 java多线程
一、线程池的原理与使用
1、线程池的优势
-
第一：降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的损耗
-
第二：提高响应速度。当任务到达时，任务可以不需要等待线程创建就能立即执行
-
第三：提高线程的可管理性。线程时稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性。使用线程池可以统一的分配，调优和监控。但是要做到合理的利用线程池，必须对其原理了如指掌。
2、创建一个线程池并提交线程任务
// 创建一个线程池
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 50, 10, TimeUnit.DAYS, new ArrayBlockingQueue<>(10),new ThreadPoolExecutor.CallerRunsPolicy());
for(int i = 0; i < 100 ;i ++) {
threadPool.execute(new Runnable() { // 用 线程池中的线程执行任务
@Override
public void run() {
}
});
}
threadPool.shutdown();
3、线程池源码解析
3.1、参数认识
-
-
corePoolSize：线程池的基本大小，当提交一个任务到线程池时，线程池会创建一个线程来执行任务，即使其他空闲的基本线程能够执行新任务也会创建线程，等到需要执行的任务数大于线程池的基本大小时就不再创建。如果调用了线程池的prestartAllCoreThreads方法，线程池会提前创建并启动所有线程。
-
maximumPoolSize：线程池最大大小，线程池允许创建的最大线程数。如果队列满了，并且已创建的线程数小于最大线程数，则线程池会再创建新的线程执行任务。值得注意的是如果使用*的任务队列这个参数就没什么效果。
-
keepAliveTime：线程活动保持时间，线程池的工作线程空闲后，保持存活的时间。所以如果任务很多，并且每个任务执行的时间比较短，可以调大这个时间，提高线程的利用率。
-
TimeUnit：线程活动保持时间的单位。可选单位（DAYS、HOURS、MINUTES、MILLISECONDS、MICROSECONDS、NANOSECOUND）
-
runnableTaskQueue：任务队列，用于保存等待执行的任务的阻塞队列。可以选择以下几个阻塞队列
-
ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列，此队列按FIFO原则对元素进行排序。
-
LinkedBlockingQueue：一个基于链表结构的阻塞队列，此队列按FIFO排序元素，吞吐量通常要高于ArrayBlockingQueue。静态工厂方法Executors.newFixedThreadPool() 是用了这个队列
-
SynchronousQueue：一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常高于LinkedBlockingQueue，静态工厂方法Executors.newCachedThreadPool使用了这个队列。
-
PriorityBlockingQueue：一个具有优先级的无限阻塞队列。
-
-
ThreadFactory：（选填）用于设置创建线程的工厂，可以通过线程工厂给每个创建出来的线程设置更有意义的名字，Debug和定位问题时非常有帮助。
-
RejectedExecutionHandler（饱和策略）：当队列和线程池都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。这个策略默认情况下是 AbortPolicy，表示无法处理新任务时抛出的异常。
-
CallerRunsPolicy：只用调用者所在的线程来运行任务
-
DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务。
-
DiscardPolicy：不处理，丢弃掉
-
当然也可以根据应用场景需要来实现RejectExecutionHandler接口自定义策略。如记录日志或持久化不能处理的任务。
-
-
-
-
-
线程运行状态：
-
RUNNING：接受新任务并且处理已经进入阻塞队列的任务
-
SHUTDOWN：不接受新任务，但是处理已进入阻塞队列的任务
-
STOP：不接受新任务，不处理已经进入阻塞队列的任务并且中断正在运行的任务。
-
TIDYING：所有任务都已经终止，workerCount为0，线程转化为TIDYING状态并且调用terminated钩子函数。
-
TERMINATED：terminated钩子函数已经运行完成
-
-
3.2、线程池提交任务源码详解
3.2.1、execute方法
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
/*
* 进行下面三步
*
* 1. 如果运行的线程小于 corePoolSize，则尝试使用用户自定义的Runnable对象创建一个新的线程
* 调用addWorker函数会原子性的检查runState和workCount，通过返回false来防止在不应该添加
* 线程时添加了线程。
*
* 2. 如果一个任务能成功入队列，在添加一个线程时仍需要进行双重检查（因为在前一此检查后该
* 线程死亡了，或当进入到此方法时，线程池已经shutdown了，所以需要再次检查状态）若有必要，
* 当停止时还需要增加一个新线程，如果此操作失败，那么就意味着线程池已经shutdown或者已经
* 饱和了，所以拒绝任务。
*
* 3. 如果无法加入队列，那么需要增加一个新线程，如果此操作失败，那么就意味着线程池已经
* shutdown或者已经饱和了，所以拒绝任务
*/
// 获取线程池控制状态
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) { // worker数量小于corePoolSize
if (addWorker(command, true)) // 添加worker
// 成功则返回
return;
// 不成功则再次获取线程池控制状态
c = ctl.get();
}
// 线程池处于RUNNING 状态，将用户自定义的Runnable对象添加进workQueue队列
if (isRunning(c) && workQueue.offer(command)) {
// 再次检查，获取线程池控制状态
int recheck = ctl.get();
// 线程池不处于RUNNING 状态，将自定义任务从 workQueue队列中移除
if (! isRunning(recheck) && remove(command))
// 拒绝执行命令
reject(command);
// worker数量等于0
else if (workerCountOf(recheck) == 0)
// 添加 worker
addWorker(null, false);
}
// 添加 worker 失败
else if (!addWorker(command, false))
// 拒绝执行命令
reject(command);
}
3.2.2、 addWorker() 解析
/**
* addWorker
* 1、原子性增加 workCount
* 2、将用户给定的任务封装成为一个 worker，并将此worker添加进 workers集合中
* 3、启动worker对应的线程，运行worker的run方法
* 4、回滚worker的创建动作，即将worker从workers集合中删除，并原子性的减少workerCount
*/
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) { // 外层无限循环
// 获取线程池控制状态
int c = ctl.get();
// 获取状态
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && // 状态大于SHUTDOWN，初始的ctl为RUNNING，小于SHUTDOWN
! (rs == SHUTDOWN && // 状态为SHUTDOWN
firstTask == null && // 第一个任务为null
! workQueue.isEmpty())) // worker队列不为空
return false;
for (;;) {
// 获取worker的数量
int wc = workerCountOf(c);
if (wc >= CAPACITY || // worker数量大于等于最大容量
// worker数量大于等于核心线程池大小或者最大线程池大小
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 比较并增加worker的数量
if (compareAndIncrementWorkerCount(c))
break retry;
// 获取线程池的控制状态
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs) // 此次的状态与上次获取的状态不相同
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
// worker开始标识
boolean workerStarted = false;
// worker被添加标识
boolean workerAdded = false;
//
Worker w = null;
try {
// 初始化worker
w = new Worker(firstTask);
// 获取worker对应的线程
final Thread t = w.thread;
if (t != null) { // 线程不为null
// 线程池锁
final ReentrantLock mainLock = this.mainLock;
// 获取锁
mainLock.lock();
try {
// Recheck while holding lock.
// Back out on ThreadFactory failure or if
// shut down before lock acquired.
// 线程池的运行状态
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN || // 小于SHUTDOWN
(rs == SHUTDOWN && firstTask == null)) { // 等于SHUTDOWN并且firstTask为null
if (t.isAlive()) // precheck that t is startable // 线程刚添加进来，还未启动就存活
// 抛出线程状态异常
throw new IllegalThreadStateException();
// 将worker添加到worker集合
workers.add(w);
// 获取worker集合的大小
int s = workers.size();
if (s > largestPoolSize) // 队列大小大于largestPoolSize
// 重新设置largestPoolSize
largestPoolSize = s;
// 设置worker已被添加标识
workerAdded = true;
}
} finally {
// 释放锁
mainLock.unlock();
}
if (workerAdded) { // worker被添加
// 开始执行worker的run方法
t.start();
// 设置worker已开始标识
workerStarted = true;
}
}
} finally {
if (! workerStarted) // worker没有开始
// 添加worker失败
addWorkerFailed(w);
}
return workerStarted;
}
3.3、执行任务源码详解
3.3.1、runWorker
-
runWorker函数中会实际执行给定任务（即调用用户重写的run方法），并且当给定任务完成后，会继续从阻塞队列中取任务，直到阻塞队列为空（即任务全部完成）。在执行给定任务时，会调用钩子函数，利用钩子函数可以完成用户自定义的一些逻辑。在runWorker中会调用到getTask函数和processWorkerExit钩子函数
final void runWorker(Worker w) {
// 获取当前线程
Thread wt = Thread.currentThread();
// 获取w的firstTask
Runnable task = w.firstTask;
// 设置w的firstTask为null
w.firstTask = null;
// 释放锁（设置state为0，允许中断）
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) { // 任务不为null或者阻塞队列还存在任务
// 获取锁
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
if ((runStateAtLeast(ctl.get(), STOP) || // 线程池的运行状态至少应该高于STOP
(Thread.interrupted() && // 线程被中断
runStateAtLeast(ctl.get(), STOP))) && // 再次检查，线程池的运行状态至少应该高于STOP
!wt.isInterrupted()) // wt线程（当前线程）没有被中断
wt.interrupt(); // 中断wt线程（当前线程）
try {
// 在执行之前调用钩子函数
beforeExecute(wt, task);
Throwable thrown = null;
try {
// 运行给定的任务
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
// 执行完后调用钩子函数
afterExecute(task, thrown);
}
} finally {
task = null;
// 增加给worker完成的任务数量
w.completedTasks++;
// 释放锁
w.unlock();
}
}
completedAbruptly = false;
} finally {
// 处理完成后，调用钩子函数
processWorkerExit(w, completedAbruptly);
}
}
3.3.2、getTask方法
-
此函数用于从workerQueue阻塞队列中获取Runnable对象，由于是阻塞队列，所以支持有限时间等待（poll）和无限时间等待（take）。在该函数中还会响应shutDown和、shutDownNow函数的操作，若检测到线程池处于SHUTDOWN或STOP状态，则会返回null，而不再返回阻塞队列中的Runnalbe对象。
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) { // 无限循环，确保操作成功
// 获取线程池控制状态
int c = ctl.get();
// 运行的状态
int rs = runStateOf(c);
// Check if queue empty only if necessary.
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { // 大于等于SHUTDOWN（表示调用了shutDown）并且（大于等于STOP（调用了shutDownNow）或者worker阻塞队列为空）
// 减少worker的数量
decrementWorkerCount();
// 返回null，不执行任务
return null;
}
// 获取worker数量
int wc = workerCountOf(c);
// Are workers subject to culling?
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; // 是否允许coreThread超时或者workerCount大于核心大小
if ((wc > maximumPoolSize || (timed && timedOut)) // worker数量大于maximumPoolSize
&& (wc > 1 || workQueue.isEmpty())) { // workerCount大于1或者worker阻塞队列为空（在阻塞队列不为空时，需要保证至少有一个wc）
if (compareAndDecrementWorkerCount(c)) // 比较并减少workerCount
// 返回null，不执行任务，该worker会退出
return null;
// 跳过剩余部分，继续循环
continue;
}
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : // 等待指定时间
workQueue.take(); // 一直等待，直到有元素
if (r != null)
return r;
// 等待指定时间后，没有获取元素，则超时
timedOut = true;
} catch (InterruptedException retry) {
// 抛出了被中断异常，重试，没有超时
timedOut = false;
}
}
}
3.3.3、processWorkerExit
-
processWorkerExit函数是在worker退出时调用到的钩子函数，而引起worker退出的主要因素如下
-
阻塞队列已经为空，即没有任务可以运行了。
-
调用了shutDown或shutDownNow函数
-
-
此函数会根据是否中断了空闲线程来确定是否减少workerCount的值，并且将worker从workers集合中移除并且会尝试终止线程池。
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) // 如果被中断，则需要减少workCount // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
// 获取可重入锁
final ReentrantLock mainLock = this.mainLock;
// 获取锁
mainLock.lock();
try {
// 将worker完成的任务添加到总的完成任务中
completedTaskCount += w.completedTasks;
// 从workers集合中移除该worker
workers.remove(w);
} finally {
// 释放锁
mainLock.unlock();
}
// 尝试终止
tryTerminate();
// 获取线程池控制状态
int c = ctl.get();
if (runStateLessThan(c, STOP)) { // 小于STOP的运行状态
if (!completedAbruptly) {
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty()) // 允许核心超时并且workQueue阻塞队列不为空
min = 1;
if (workerCountOf(c) >= min) // workerCount大于等于min
// 直接返回
return; // replacement not needed
}
// 添加worker
addWorker(null, false);
}
}
3.4、关闭线程池源码详解
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 检查shutdown权限
checkShutdownAccess();
// 设置线程池控制状态为SHUTDOWN
advanceRunState(SHUTDOWN);
// 中断空闲worker
interruptIdleWorkers();
// 调用shutdown钩子函数
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
// 尝试终止
tryTerminate();
}
final void tryTerminate() {
for (;;) { // 无限循环，确保操作成功
// 获取线程池控制状态
int c = ctl.get();
if (isRunning(c) || // 线程池的运行状态为RUNNING
runStateAtLeast(c, TIDYING) || // 线程池的运行状态最小要大于TIDYING
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) // 线程池的运行状态为SHUTDOWN并且workQueue队列不为null
// 不能终止，直接返回
return;
if (workerCountOf(c) != 0) { // 线程池正在运行的worker数量不为0 // Eligible to terminate
// 仅仅中断一个空闲的worker
interruptIdleWorkers(ONLY_ONE);
return;
}
// 获取线程池的锁
final ReentrantLock mainLock = this.mainLock;
// 获取锁
mainLock.lock();
try {
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { // 比较并设置线程池控制状态为TIDYING
try {
// 终止，钩子函数
terminated();
} finally {
// 设置线程池控制状态为TERMINATED
ctl.set(ctlOf(TERMINATED, 0));
// 释放在termination条件上等待的所有线程
termination.signalAll();
}
return;
}
} finally {
// 释放锁
mainLock.unlock();
}
// else retry on failed CAS
}
}
private void interruptIdleWorkers(boolean onlyOne) {
// 线程池的锁
final ReentrantLock mainLock = this.mainLock;
// 获取锁
mainLock.lock();
try {
for (Worker w : workers) { // 遍历workers队列
// worker对应的线程
Thread t = w.thread;
if (!t.isInterrupted() && w.tryLock()) { // 线程未被中断并且成功获得锁
try {
// 中断线程
t.interrupt();
} catch (SecurityException ignore) {
} finally {
// 释放锁
w.unlock();
}
}
if (onlyOne) // 若只中断一个，则跳出循环
break;
}
} finally {
// 释放锁
mainLock.unlock();
}
}
二、Executor框架
1、Executors可创建的线程池
ExecutorService pool = Executors.newFixedThreadPool(int ThreadSize); // 固定大小线程池
ExecutorService pool = Executors.newCacheThreadPool(); // 缓冲线程池
ExecutorService pool = Executors.newSingleThreadPool(); // 单个线程的线程池
ScheduledExecutorService pool = Executors.newScheduledThreadPool(int ThreadSize); // 任务线程池
ExecutorService pool = Executors.newWorkStealingPool(); // ForkJoin类线程池
2、execute 方法与submit 方法
public void execute(Runnable command) // 无返回值
public <T> ForkJoinTask<T> submit(Callable<T> task) // 返回一个ForkJoinTask
上一篇： Java多线程（一）认识多线程
下一篇： 喜欢你