Java教程
Java线程池源码分析
本文主要是介绍Java线程池源码分析，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！
Java线程池，基于jdk1.8
一些属性
//线程数量和线程池状态 高三位是状态 低29位是数量
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//位移位数 29
private static final int COUNT_BITS = Integer.SIZE - 3;
//容量 2的29次方-1 00011111 11111111 11111111 11111111
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
// 线程池状态标记
// 11100000 00000000 00000000 00000000
private static final int RUNNING = -1 << COUNT_BITS;
// 00000000 00000000 00000000 00000000
private static final int SHUTDOWN = 0 << COUNT_BITS;
// 00100000 00000000 00000000 00000000
private static final int STOP = 1 << COUNT_BITS;
// 01000000 00000000 00000000 00000000
private static final int TIDYING = 2 << COUNT_BITS;
// 01100000 00000000 00000000 00000000
private static final int TERMINATED = 3 << COUNT_BITS;
// 位运算 获取当前线程池的状态
private static int runStateOf(int c) { return c & ~CAPACITY; }
// 位运算 获取当前线程池的线程数
private static int workerCountOf(int c) { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }
private final BlockingQueue<Runnable> workQueue;
private final ReentrantLock mainLock = new ReentrantLock();
private final HashSet<Worker> workers = new HashSet<Worker>();
//是否 允许核心线程超时
private volatile boolean allowCoreThreadTimeOut;
private int largestPoolSize;
构造方法
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
threadFactory, defaultHandler);
}
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
RejectedExecutionHandler handler) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), handler);
}
//参数依次是 核心线程数 最大线程数 线程存活时间 线程存活时间的单位 阻塞队列 线程工厂 拒绝策略
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
提交任务
public void execute(Runnable command) {
//任务为空，抛异常
if (command == null)
throw new NullPointerException();
//线程计数
int c = ctl.get();
//当前线程池的线程数 小于 核心线程数
if (workerCountOf(c) < corePoolSize) {
// 添加线程
if (addWorker(command, true))
return;
//获取最新的线程数
c = ctl.get();
}
//在运行中并且可以将任务添加到阻塞队列(阻塞队列未满)
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// 线程池不是运行中 并且 将阻塞队列 里面的线程 移除
if (! isRunning(recheck) && remove(command))
//执行拒绝
reject(command);
//如果线程池中没有线程了，则创建一个线程执行任务
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
// 阻塞队列满了，添加线程失败(达到最大线程数) 执行拒绝
else if (!addWorker(command, false))
reject(command);
}
private static boolean isRunning(int c) {
return c < SHUTDOWN;
}
//添加线程方法 当前任务 是否核心线程
private boolean addWorker(Runnable firstTask, boolean core) {
retry://跳出标志
for (;;) {
int c = ctl.get();
//获取线程池状态
int rs = runStateOf(c);
// 检查阻塞队列是否为空 阻塞队列不为空，说明里面有任务，说明线程数已经达到核心线程数的限制
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
// 线程数 大于等于 最大线程容量 或者 线程数大于等于 核心线程数/最大线程数 不再添加线程
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//增加线程数 跳出循环，执行flag处的语句
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
}
}
// flag 添加的线程 启动 和 添加成功标记
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//将当前任务封装成一个Worker线程
w = new Worker(firstTask);
//工作线程
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();//加锁
try {
//
int rs = runStateOf(ctl.get());
// 当前线程池是运行状态 || 是SHUTDOWN并且当前任务为空
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
// 新创建的线程 已经存活 抛异常 刚创建还没启动怎么就启动了
if (t.isAlive())
throw new IllegalThreadStateException();
//添加到工作线程集合
workers.add(w);
// 判断工作线程数是否大于醉倒线程数 出现过的最大线程数
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
//标记添加线程成功
workerAdded = true;
}
} finally {
mainLock.unlock();
}
// 添加线程成功 启动 设置启动标志
if (workerAdded) {
t.start();
workerStarted = true;
}
}
} finally {
//添加工作线程失败 将Worker从Worker集合中删除，并且减少工作线程数 因为上面对工作线程数+1了
if (! workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
private boolean compareAndIncrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect + 1);
}
private void addWorkerFailed(Worker w) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (w != null)
workers.remove(w);
decrementWorkerCount();
tryTerminate();
} finally {
mainLock.unlock();
}
}
private void decrementWorkerCount() {
do {} while (! compareAndDecrementWorkerCount(ctl.get()));
}
//执行任务
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // 允许中断
boolean completedAbruptly = true;
try {
//当前任务不为空或者可以从阻塞队列里面获取到任务
while (task != null || (task = getTask()) != null) {
w.lock();
//(线程池状态为stop || (线程中断了 并且 线程池准备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 {
//置空任务 任务完成数加1
task = null;
w.completedTasks++;
w.unlock();
}
}
//获取不到任务了
completedAbruptly = false;
} finally {
//销毁工作线程
processWorkerExit(w, completedAbruptly);
}
}
//获取任务方法
private Runnable getTask() {
boolean timedOut = false;
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
// 阻塞队列为空
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// 允许核心线程超时 或者 当前线程数大于核心线程数(有非核心线程)
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
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;
}
}
}
private static boolean runStateAtLeast(int c, int s) {
return c >= s;
}
private boolean compareAndDecrementWorkerCount(int expect) {
return ctl.compareAndSet(expect, expect - 1);
}
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//增加完成任务数量
completedTaskCount += w.completedTasks;
//移除Worker
workers.remove(w);
} finally {
mainLock.unlock();
}
//终止线程
tryTerminate();
int c = ctl.get();
if (runStateLessThan(c, STOP)) {
//检查是不是因为获取不到任务 进入该方法的
if (!completedAbruptly) {
//线程获取不到任务，就是该消亡线程了
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
//给线程池保存一个工作线程
addWorker(null, false);
}
}
final void tryTerminate() {
for (;;) {
int c = ctl.get();
if (isRunning(c) ||
runStateAtLeast(c, TIDYING) ||
(runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
return;
if (workerCountOf(c) != 0) { // Eligible to terminate
interruptIdleWorkers(ONLY_ONE);
return;
}
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
try {
terminated();
} finally {
ctl.set(ctlOf(TERMINATED, 0));
termination.signalAll();
}
return;
}
} finally {
mainLock.unlock();
}
// else retry on failed CAS
}
}
//内部类
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
// 工作线程
final Thread thread;
// 工作任务
Runnable firstTask;
// 完成的任务数量
volatile long completedTasks;
Worker(Runnable firstTask) {
setState(-1); // aqs的状态 防止中断
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
// 执行任务 活跃线程执行的任务在这里
public void run() {
runWorker(this);
}
// 0 未上锁， 1 上锁了
protected boolean isHeldExclusively() {
return getState() != 0;
}
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
public void lock() { acquire(1); }
public boolean tryLock() { return tryAcquire(1); }
public void unlock() { release(1); }
public boolean isLocked() { return isHeldExclusively(); }
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}
这篇关于Java线程池源码分析的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！
您可能喜欢
-
怎么修改Kafka的JVM参数？-icode9专业技术文章分享12-24
-
线下车企门店如何实现线上线下融合？12-23
-
鸿蒙Next ArkTS编程规范总结12-23
-
物流团队冬至高效运转，哪款办公软件可助力风险评估？12-23
-
优化库存，提升效率：医药企业如何借助看板软件实现仓库智能化12-23
-
项目管理零负担！轻量化看板工具如何助力团队协作12-23
-
电商活动复盘，为何是团队成长的核心环节？12-23
-
鸿蒙Next ArkTS高性能编程实战12-23
-
数据驱动：电商复盘从基础到进阶！12-23
-
从数据到客户：跨境电商如何通过销售跟踪工具提升营销精准度？12-23
-
汽车4S店运营效率提升的核心工具12-23
-
ERP 系统开发案例：用 PERT 图打造高效项目管理12-23
-
初创团队必备！全面解析支持甘特图的项目管理工具与功能特点12-23
-
任务进度跟踪不再难！可视化项目协作工具精选12-23
-
看板工具为汽车销售门店协作赋能：关键方法12-23
栏目导航
