- 一、线程池优势
- 二、ThreadPoolExecutor继承关系
- 三、ThreadPoolExecutor构造器分析
- 四、如何自定义线程池
- 五、ThreadPoolExecutor处理流程
- 六、核心参数说明
- 线程池状态
- 拒绝策略
- 七、核心源码分析
- 分析流程
- execute(Runnable command)
- addWorker(Runnable firstTask, boolean core)
- runWorker(Worker w)
- processWorkerExit(Worker w, boolean completedAbruptly)
- shutdown() VS shutdownNow()
Java中的线程池是运用场景最多的并发框架，几乎所有需要异步或并发执行任务的程序 都可以使用线程池。合理地使用线程池能够带来3个好处：
- 降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。
- 提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。
- 提高线程的可管理性。线程是稀缺资源，如果无限制地创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一分配、调优和监控。
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.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
自定义线程工厂
class SimpleThreadFactory implements ThreadFactory {
public Thread newThread(Runnable r) {
return new Thread(r);
}
}
创建线程池
ExecutorService es = new ThreadPoolExecutor(
2, 核心
5, 最大
2L, 保活时间
TimeUnit.SECONDS, 保活时间单位
new LinkedBlockingQueue<>(2), 必须指明，默认最大为Integer.MAX_VALUE
new SimpleThreadFactory(), 线程池，可以自定义
new ThreadPoolExecutor.AbortPolicy()); 线程池满之后，处理策略，默认拒绝策略
- 添加任务到线程池中，核心线程执行
- 如果核心线程不够，添加到阻塞队列中
- 如果阻塞队列不够，开启最大线程执行
- 如果最大线程不够，使用拒绝策略
源码认证：
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);//拒绝
}
- RUNNING： 运行状态
- SHUTDOWN ： 不接受任务，但处理已经有的任务
- STOP： 停止所有的任务，包括正在运行的
- TIDYING ： 所有任务停止，待处理任务为0，执行TERMINATED() 方法
- TERMINATED： TERMINATED() 方法执行完成
源码认证：
这里使用一个ctl 是一个32位的int数据，使用高三位标识线程池状态，剩下的标识线程池线程数量。
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; 线程池最大数量
// runState is stored in the high-order bits
//2^3=8 这里可以表示8种状态
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;
// Packing and unpacking ctl
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; }
- RejectedExecutionHandler ：拒绝抛出异常
- CallerRunsPolicy ：本身的调用者调用任务
- DiscardPolicy ：丢弃任务
- DiscardOldestPolicy ：丢弃阻塞队列中第一个任务
源码认证：
public static class CallerRunsPolicy implements RejectedExecutionHandler {
public CallerRunsPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
public static class AbortPolicy implements RejectedExecutionHandler {
public AbortPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}
public static class DiscardPolicy implements RejectedExecutionHandler {
public DiscardPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
public DiscardOldestPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}
- execute执行添加任务，有可能核心，阻塞，最大，拒绝。
- addWorker封装任务执行体，启动工作线程Worker.start()
- Worker.runWorker工作线程启动后，有任务执行，无任务队列中获取任务
- processWorkerExit工作线程退出工作
证明线程的处理流程
- 计算当前线程池数量是否小于核心线程池数量 -> addWorker(command, true)
- 判断线程池状态和添加阻塞队列 ->workQueue.offer(command)
- 判断是否可以添加到最大线程池 ->addWorker(command, false)
- 拒绝策略
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);//拒绝
}
- 判断线程状态 和 增加线程数量
- 新建Worker，添加线程到workers(HashSet)集合中，开启线程执行
- 启动失败 1.移除workers集合中当前线程 2.线程计算减一 3.执行钩子函数
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 以上状态(STOP TIDYING等)
! (rs == SHUTDOWN && //线程池已经关闭，还有没处理的任务
firstTask == null && //此时任务为补充任务，非核心和最大
! workQueue.isEmpty())) //队列是否为空
return false;
for (;;) {
int wc = workerCountOf(c);//目前线程数量
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))//判断是否大于核心或最大
return false;
if (compareAndIncrementWorkerCount(c))//自增1 线程数
break retry; //退出
c = ctl.get(); // Re-read ctl
if (runStateOf(c) != rs)//cas判断runState
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask);//工作者
final Thread t = w.thread;//获取线程
if (t != null) {
final ReentrantLock mainLock = this.mainLock;//防止多线程修改workers
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 ||
(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;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start();//启动线程
workerStarted = true;
}
}
} finally {
if (! workerStarted)
addWorkerFailed(w);//启动失败 workers.remove(w) 移除线程
}
return workerStarted;
}
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts 刚开始抑制中断为-1 现在-1 变 0 允许中断
boolean completedAbruptly = true;
try {
//获取当前任务 或则 阻塞队列任务
while (task != null || (task = getTask()) != null) {
w.lock();
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++;//当前工作线程完成任务+1
w.unlock();
}
}
//the worker died due to user exception == completedAbruptly = true
completedAbruptly = false;//只钩子函数报出异常才为ture
} finally {
processWorkerExit(w, completedAbruptly);
}
}
private void processWorkerExit(Worker w, boolean completedAbruptly) {
//the worker died due to user exception == completedAbruptly = true
//由于用户自定义钩子函数抛出的异常=true
if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted
decrementWorkerCount(); //调整工作线程数量
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;//work使用数量加1
workers.remove(w);//线程集合移除
} finally {
mainLock.unlock();
}
tryTerminate();//执行钩子
int c = ctl.get();
if (runStateLessThan(c, STOP)) {//状态线程池判断
//检查是否用户异常中断，是的话添加对应的工作addWorker
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);
}
}
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(SHUTDOWN);//Cas设置线程池不接受新任务,继续处理队列中的任务
interruptIdleWorkers();//打断空闲的线程
onShutdown(); // hook for ScheduledThreadPoolExecutor 扩展点
} finally {
mainLock.unlock();
}
tryTerminate();//
}
public List<Runnable> shutdownNow() {
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
checkShutdownAccess();
advanceRunState(STOP);//设置线程池不接受新任务,也停止处理队列中的任务
interruptWorkers();//打断所有的线程
tasks = drainQueue();//去拿队列中的剩余返回
} finally {
mainLock.unlock();
}
tryTerminate();
return tasks;
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）