线程数量和队列的关系
流程的示意图
线程池的优势
小刘老实讲源码
线程池的优势
线程池是 Java 中对线程进行统一管理，复用和监控的组件。它有几个重要的参数，sizeCtl 它的高3位表示线程池的运行状态，剩余27位表示线程的运行数量。它将执行的线程封装成了 Worker 对象，Worker 对象本身也是一个锁，是独占锁模式，主要是为了去确认线程运行的状态，lock 状态时，表示 worker 在执行，unlock 状态时 worker 没有在执行。所有的 Worker 对象都由一个 HashSet 管理。在添加一个线程时，会采用自旋和 CAS 的方式检查线程的运行状态和运行数量，只有在状态是 Runing 且线程数量符合要求时，才可以将任务交给线程池。
线程池中核心线程和普通线程都是 Worker 对象，区别是他们是否会执行退出逻辑。当设置超时时间，并且允许核心线程退出时，每一个 Worker 都会通过 BlockingQueue 的带超时的 poll 去取任务，当没有任务时，就会执行退出逻辑，此时就是一个普通线程。
当不允许核心线程退出时，如果当前线程池中的线程没有达到核心线程的上限，Worker 在取消息时就使用 take 方法，阻塞到它取到任务，这样 Worker 就不会执行退出逻辑，它就是一个核心线程。
在 Worker 退出时，如果当前线程池是 Running 或者 ShutDown 状态，线程池没有工作线程了，但是队列中仍然有任务，就再创建一个临时的 Worker 去执行这些任务。
线程池的状态更新是通过自旋+ CAS 的方式更新的
线程池的原理
1.将一个任务交给线程池时，线程池会经过以下判断
[图片上传失败...(image-302947-1597385094797)]
2.创建新的工作线程后，工作线程会首先执行传递进来的任务，然后再从任务队列中不断取新的任务执行
- shutDown 不在接收新的任务，但是可以继续执行已经存在的任务
- shutDownNow 方法会停止所有线程，并且停止线程入队
一、线程池的主要参数
corePoolSize:线程中的核心线程数，代表可以工作的线程数
maximumPoolSize:线程池中允许存在的最大线程数(coreThread + idle thread)，当等待队列满后，将使用 maximumPoolSize 作为边界条件
keeplive + unit : 允许空闲线程存活的时间
workQueue: 等待队列，当核心线程满后，新的任务加入到等待队列
threadFactory 线程工厂，用来创建线程
handler 拒绝策略
线程池的创建策略
[图片上传失败...(image-7c730d-1597385094797)]
- 判断核心线程池是否已满，如果不是，则创建线程执行任务
- 如果核心线程池满了，判断队列是否满了，如果队列没满，将任务放在队列中
- 如果队列满了，则判断线程池（maximumPoolSize 是否达到上限）是否已满，如果没满，创建线程执行任务
- 如果线程池也满了，则按照拒绝策略对任务进行处理
[图片上传失败...(image-d33eef-1597385094797)]
队列和最大
使用提交队列 SynchronousQueue 时，它无法设置任务数，当任务来临时立即提交执行，当任务数大于最大线程数时会抛出异常
public class ThreadPool {
private static ExecutorService pool;
public static void main( String[] args )
{
//maximumPoolSize设置为2 ，拒绝策略为AbortPolic策略，直接抛出异常
pool = new ThreadPoolExecutor(1, 2, 1000, TimeUnit.MILLISECONDS, new SynchronousQueue<Runnable>(),Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());
for(int i=0;i<3;i++) {
pool.execute(new ThreadTask());
}
}
}
public class ThreadTask implements Runnable{
public void run() {
System.out.println(Thread.currentThread().getName());
}
}
output
pool-1-thread-1
pool-1-thread-2
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task com.hhxx.test.ThreadTask@55f96302 rejected from java.util.concurrent.ThreadPoolExecutor@3d4eac69[Running, pool size = 2, active threads = 0, queued tasks = 0, completed tasks = 2]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(Unknown Source)
使用有界队列 ArrayBlockingQueue 时，当有任务来临时会创建线程执行任务，当达到 coreThreadSize 上限时会将任务加入到任务队列中。如果等待队列也满，会继续创建闲置线程直到达到 maximumPoolSize 的上限。如果达到上限后仍然有任务到来，会执行拒绝的策略
线程池的生命状态
Running:运行的状态
ShutDown:不接受新任务，可以处理已经添加的请求
Stop:不接受新的任务，停止正在运行的任务，不处理队列中的任务
Tidying:所有任务都已经终止
Terminated:线程池彻底终止
线程池复用线程的原理
将一个 Runnable 任务对象交给线程池后，如果线程池的核心线程数还没有满，它会创建一个 Worker 对象，并将这个 Runnable 交给这个 Worker 对象。Worker 的 Run 方法会启动一个循环，先执行我们提交的任务，当我们提交的任务执行完后，会继续从队列中获取新的任务去执行。
线程池源码
// ctl 表示当前线程池的运行状态，它的低27位表示当前运行的线程数量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// 线程运行的状态
private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0 << COUNT_BITS;
private static final int STOP = 1 << COUNT_BITS;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;
// ctl 的状态是否比指定的 s 状态小
private static boolean runStateLessThan(int c, int s) {
return c < s;
}
// ctl 的状态是否不小于指定的 s 状态
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
// 任务队列，当核心线程数已经满时，任务会加入到 workQueue 中
private final BlockingQueue<Runnable> workQueue;
// 线程池的全局锁，向线程池添加或者减少线程，修改线程池状态都需要 lock，保证在同一时刻是能有一个线程向线程池加入线程
private final ReentrantLock mainLock = new ReentrantLock();
// 存放所有线程的地方
private final HashSet<Worker> workers = new HashSet<>();
// termination.await 阻塞线程 termination.signalAll 唤醒所有阻塞线程
private final Condition termination = mainLock.newCondition();
// 记录线程池生命周期内运行的线程的最大值
private int largestPoolSize;
// 完成过的所有任务，当线程被回收时计数
private long completedTaskCount;
// 线程空闲的时间，如果超出空闲时间 allowCoreThreadTimeOut = false，核心线程处于闲置状态，也不会被回收
// allowCoreThreadTimeOut = true，核心线程处于闲置状态，会被回收
// 非核心线程超时后都会被回收
private volatile long keepAliveTime;
private volatile boolean allowCoreThreadTimeOut;
// 核心线程数量
private volatile int corePoolSize;
// 线程池中最大的线程数量，超过后加入的任务，会被执行拒绝策略
private volatile int maximumPoolSize;
// 采用了 AQS 的独占模式
// 独占模式的两个重要属性 state 和 ExclusiveOwnerThread
// state 0 表示可以占用 >0 表示已经被占用 <0 表示初始化中
// ExclusiveOwnerThread 表示独占锁的线程
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
// Worker 中封装的工作线程
final Thread thread;
// 工作线程要执行的第一个任务
Runnable firstTask;
volatile long completedTasks;
Worker(Runnable firstTask) {
// 设置初始化状态
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
// 将 Worker 交给 Thread，当 Thread start 时，会执行 Worker 的 run 方法
this.thread = getThreadFactory().newThread(this);
}
public void run() {
runWorker(this);
}
// 是否被独占 0 没有 1 被独占
protected boolean isHeldExclusively() {
return getState() != 0;
}
protected boolean tryAcquire(int unused) {
// 申请占用锁 设置 state 为 1
if (compareAndSetState(0, 1)) {
// 设置占有线程
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
// 外部不会直接调用，当外部调用 unlock 时，tryRelease 被调用
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
execute方法
public void execute(Runnable command) {
if (command == null)
throw new NullPointerException();
// ctl 高3位线程运行状态 低 27 位表示线程数量
int c = ctl.get();
// 工作的线程数量是否小于核心线程数
if (workerCountOf(c) < corePoolSize) {
// 添加一个核心线程，并启动线程
if (addWorker(command, true))
return;
// 执行到这里表示发生了并发调用，导致 addWorker 失败
// 当前线程状态发生了改变
c = ctl.get();
}
// 核心线程数已经满了
// addWorker 失败了
// 如果线程是 running 状态，将任务加入到队列中
if (isRunning(c) && workQueue.offer(command)) {
// 再次获取 ctl
int recheck = ctl.get();
// 如果是非 running 状态，需要把刚提交的任务移除掉
if (! isRunning(recheck) && remove(command))
// 拒绝这个任务
reject(command);
// 如果线程池是运行状态，为了保证当前有线程工作，加入了一层检查
// 如果工作线程数是 0 ，启动一个线程
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 如果线程入队失败，尝试 add 一个非工作线程，如果达到最大线程数，则会失败
// 如果线程处于非运行状态也会失败
else if (!addWorker(command, false))
reject(command);
}
addWorker 操作
// firstTask 代表要执行的任务，可以为 null，如果为空 Worker 会自动从消息队列中取消息
// core 代表启动的是否为核心线程，true 使用核心线程数量限制 false 使用最大线程数限制
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
// 使用自旋
for (;;) {
// 获取线程池的状态
int c = ctl.get();
int rs = runStateOf(c);
// Check if queue empty only if necessary.
// 如果当前线程池不处于运行状态，任务队列没有任务可以执行，addWorker 失败，执行拒绝策略
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
// 内部自旋
for (;;) {
// 获取当前线程运行的数量
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
// 根据 core 来判断是选用核心线程数还是最大线程数限制
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
// 通过 CAS 的方式更新执行的线程数
// CAS 失败，可能是其他线程修改了这个值
if (compareAndIncrementWorkerCount(c))
// 如果成功，直接跳出外部循环 因为 retry 标记的是外部循环
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 的过程
// 创建 worker 是否成功
boolean workerStarted = false;
// 添加 Worker 到 HashSet 是否成功
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask);
final Thread t = w.thread;
// 防止 ThreadFactory 类有 bug，因为 ThreadFactory 是可以通过外部传入的
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 获取线程状态
int rs = runStateOf(ctl.get());
// 如果当前线程是运行状态 可以将 Worker 加入到 HashSet 队列中
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
// 当线程已经被启动过了，抛出异常
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
// add 成功
workerAdded = true;
}
} finally {
mainLock.unlock();
}
// 如果 add 成功 启动这个线程
if (workerAdded) {
t.start();
// 执行线程成功
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
Worker 的 runWorker 方法
final void runWorker(Worker w) {
// 获取 Worker 内部的 thread 的对象
Thread wt = Thread.currentThread();
// 获取我们传入的任务
Runnable task = w.firstTask;
w.firstTask = null;
// 启动 worker 之前，调用 unlock 将锁的状态和独占线程 release 掉，进行初始化
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
// Worker 有外部传入的任务，或者工作队列中有任务可以执行
while (task != null || (task = getTask()) != null) {
// 给 worker 加独占锁，主要是为了确认 Worker 是否有任务在执行
w.lock();
// 如果当前线程状态大于 stop 给线程一个中断信号
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
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;
w.completedTasks++;
// 任务执行完，释放锁
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
** getTask 获取待执行的任务**
getTask 使用一个自旋的操作
private Runnable getTask() {
boolean timedOut = false; // Did the last poll() time out?
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 如果线程池不处于运行状态，或者队列中没有任务
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
使用 CAS 的方式将计数减 1
decrementWorkerCount();
return null;
}
// 当前运行线程的数量
int wc = workerCountOf(c);
// Are workers subject to culling?
// 是否允许核心线程被回收
// true 表示核心线程也会被回收
// false 表示核心不会被回收
// 1. 我们设置了允许核心线程被回收
// 2. 当前核心线程数量已经超过了上限
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
// 核心线程是否回收取决于执行 poll 超时方法，还是 take 阻塞方法。
// 如果使用 poll 在超时时间内没有获取到任务，Worker 会执行退出方法
// 如果使用 take 会一直阻塞在这里，直到获取到任务，所以 Worker 不会退出，这时 Worker 就是一个核心线程
try {
// 如果执行 poll 操作，会有可能到时 worker 取不到任务，执行退出逻辑
// take 会一直阻塞在这里，直到取出任务
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
processWorkerExit 执行 Worker 的退出
processWorkerExit 分为正常突出和异常退出
completedAbruptly true 表示异常退出 false 表示正常退出
private void processWorkerExit(Worker w, boolean completedAbruptly) {
// 如果是异常退出的，让工作线程计数减 1
if (completedAbruptly)
decrementWorkerCount();
// 需要操作线程池，所以要加锁
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 从队列中移除
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
// 尝试终止线程
tryTerminate();
// 获取当前线程的状态
// 如果是 Running 或者 ShutDown 状态
int c = ctl.get();
if (runStateLessThan(c, STOP)) {
// 如果不是异常退出的
if (!completedAbruptly) {
// 队列中还有任务要执行，就在创建一个 Worker 去执行任务
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
// 如果仍然有工作线程，不需要补充线程了
if (workerCountOf(c) >= min)
return; // replacement not needed
}
// 补充一个线程
addWorker(null, false);
}
}
shuDown 函数
根据锁状态判断 Worker 是否闲置，闲置的线程直接发送中断信号
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
// 使用自旋和 CAS 的方式更新线程池状态
advanceRunState(SHUTDOWN);
// 中断所有空闲线程
interruptIdleWorkers();
onShutdown(); // hook for ScheduledThreadPoolExecutor
} finally {
mainLock.unlock();
}
tryTerminate();
}
private void interruptIdleWorkers(boolean onlyOne) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
// 迭代所有 Worker
for (Worker w : workers) {
Thread t = w.thread;
// !t.isInterrupted() 表示当前线程处于非中断状态
// tryLock 成功代表 Worker 处于空闲状态
if (!t.isInterrupted() && w.tryLock()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
} finally {
w.unlock();
}
}
if (onlyOne)
break;
}
} finally {
mainLock.unlock();
}
}
网友评论