文章目录[隐藏]
继承关系
Executor接口
public interface Executor {
void execute(Runnable command);
}
ExecutorService接口
public interface ExecutorService extends Executor {
void shutdown();
List<Runnable> shutdownNow();
boolean isShutdown();
boolean isTerminated();
boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException;
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException;
<T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}
ExecutorService接口继承Executor接口，并减少了submit、shutdown、invokeAll等等一系列办法。
AbstractExecutorService抽象类
public abstract class AbstractExecutorService implements ExecutorService {
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks, boolean timed, long nanos)
throws InterruptedException, ExecutionException, TimeoutException {...}
public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
throws InterruptedException, ExecutionException {... }
public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException {...}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
throws InterruptedException {...}
public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
long timeout, TimeUnit unit)
throws InterruptedException {...}
}
AbstractExecutorService抽象类实现ExecutorService接口，并且提供了一些办法的默认实现，例如submit办法、invokeAny办法、invokeAll办法。
像execute办法、线程池的敞开办法（shutdown、shutdownNow等等）就没有提供默认的实现。
构造函数与线程池状态
public ThreadPoolExecutor(int corePoolSize, //外围线程数
int maximumPoolSize, //最大线程数
long keepAliveTime, //线程存活工夫
TimeUnit unit, //keepAliveTime的单位
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;
}
线程池状态
//记录线程池状态和线程数量（总共32位，前三位示意线程池状态，后29位示意线程数量）
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
//线程数量统计位数29 Integer.SIZE=32
private static final int COUNT_BITS = Integer.SIZE - 3;
//容量 000 11111111111111111111111111111
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
//运行中 111 00000000000000000000000000000
private static final int RUNNING = -1 << COUNT_BITS;
//敞开 000 00000000000000000000000000000
private static final int SHUTDOWN = 0 << COUNT_BITS;
//进行 001 00000000000000000000000000000
private static final int STOP = 1 << COUNT_BITS;
//整顿 010 00000000000000000000000000000
private static final int TIDYING = 2 << COUNT_BITS;
//终止 011 00000000000000000000000000000
private static final int TERMINATED = 3 << COUNT_BITS;
//获取运行状态（获取前3位）
private static int runStateOf(int c) { return c & ~CAPACITY; }
//获取线程个数（获取后29位）
private static int workerCountOf(int c) { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }
int是4个字节,32位
RUNNING：承受新工作并且解决阻塞队列里的工作 SHUTDOWN：回绝新工作然而解决阻塞队列里的工作 STOP：回绝新工作并且摈弃阻塞队列里的工作同时会中断正在解决的工作 TIDYING：所有工作都执行完（蕴含阻塞队列外面工作），以后线程池流动线程为0，将要调用terminated办法 TERMINATED：终止状态。terminated办法调用实现当前的状态 线程池状态转换： RUNNING -> SHUTDOWN：显式调用shutdown()办法, 或者隐式调用了finalize()办法 (RUNNING or SHUTDOWN) -> STOP：显式调用shutdownNow()办法 SHUTDOWN -> TIDYING：当线程池和工作队列都为空的时候 STOP -> TIDYING：当线程池为空的时候 TIDYING -> TERMINATED：当 terminated() hook 办法执行实现时候
submit办法和execute办法的区别
submit办法
- 调用submit办法，传入Runnable或者Callable对象
- 判断传入的对象是否为null，为null则抛出异样，不为null持续流程
- 将传入的对象转换为RunnableFuture对象
- 执行execute办法，传入RunnableFuture对象
- 返回RunnableFuture对象
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
execute办法
public void execute(Runnable command) {
//传进来的线程为null，则抛出空指针异样
if (command == null)
throw new NullPointerException();
//获取以后线程池的状态+线程个数变量
int c = ctl.get();
/**
* 3个步骤
*/
//1.判断以后线程池线程个数是否小于corePoolSize,小于则调用addWorker办法创立新线程运行,
//且传进来的Runnable当做第一个工作执行。
//如果调用addWorker办法返回false，则间接返回
if (workerCountOf(c) < corePoolSize) {
//增加一个core线程(外围线程)。此处参数的true，示意增加的线程是core容量下的线程
if (addWorker(command, true))
return;
//刷新数据，乐观锁就是没有锁
c = ctl.get();
}
/* isRunning办法的定义：
private static boolean isRunning(int c)
{return c < SHUTDOWN;}
2.SHUTDOWN值为0，即如果c小于0，示意在运行；offer用来判断工作是否胜利入队*/
if (isRunning(c) && workQueue.offer(command)) {
//二次查看
int recheck = ctl.get();
//如果以后线程池状态不是RUNNING则从队列删除工作，并执行回绝策略
if (! isRunning<strong style="color:transparent">来源gaodai#ma#com搞@代~码网</strong>(recheck) && remove(command))
//执行回绝策略
reject(command);
//否则如果以后线程池线程空，则增加一个线程
else if (workerCountOf(recheck) == 0)
//增加一个空线程进线程池，应用非core容量线程
//仅有一种状况，会走这步，core线程数为0，max线程数>0,队列容量>0
//创立一个非core容量的线程，线程池会将队列的command执行
addWorker(null, false);
}
//线程池进行了或者队列已满，增加maximumPoolSize容量工作线程，如果失败，执行回绝策略
else if (!addWorker(command, false))
reject(command);
}
ThreadPoolExecutor.addWorker()
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get(); //获取运行状态和工作数量
int rs = runStateOf(c); //获取以后线程池运行的状态
// Check if queue empty only if necessary.
//条件代表着以下几个场景，间接返回false阐明当前工作线程创立失败
//1.rs>SHUTDOWN 此时不再接管新工作，且所有的工作曾经执行结束
//2.rs=SHUTDOWN 此时不再接管新工作，然而会执行队列中的工作
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
//先判断以后流动的线程数是否大于最大值，如果超过了就间接返回false阐明线程创立失败
//如果没有超过再依据core的值再进行以下判断
//1. core为true，则判断以后流动的线程数是否大于corePoolSize
//2. core为false，则判断以后流动线程数是否大于maximumPoolSize
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
//比拟以后值是否和c雷同，如果雷同，则改为c+1，并且跳出大循环，间接执行Worker进行线程创立
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get(); // Re-read ctl
//查看下以后线程池的状态是否曾经产生扭转
//如果曾经扭转了，则进行外层retry大循环，否则只进行内层的循环
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
//Worker的也是Runnable的实现类
w = new Worker(firstTask);
//因为不能够间接在Worker的构造方法中进行线程创立
//所以要把它的援用赋给t不便前面进行线程创立
final Thread t = w.thread;
if (t != 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 ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive()) // precheck that t is startable
throw new IllegalThreadStateException();
workers.add(w);//将创立的线程增加到workers容器中
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);
}
return workerStarted;
}
Worker办法
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable{
/** Thread this worker is running in. Null if factory fails. */
final Thread thread;
/** Initial task to run. Possibly null. */
Runnable firstTask;
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
}
Worker在ThreadPoolExecutor为一个外部类实现了Runnable接口。只有一个构造方法，在下面的addWorker()中final Thread t = w.thread;晓得其实是获取了线程的对象，因为在构造方法中，线程的援用即是它本人。
因而在调用t.start()执行的是（Worker类中的办法）：
/** Delegates main run loop to outer runWorker */
public void run() {
//这里执行的是ThreadPoolExecutor中的runWorker
runWorker(this);
}
ThreadPoolExecutor.runWorker()
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;//获取Worker中的工作
w.firstTask = null; //将Woeker中的工作置空
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
//如果当前任务为空 那么就从getTask中取得工作
/**
* 如果task不为空，执行完task后则将task置空
* 持续进入循环，则从getTask中获取工作
*/
while (task != null || (task = getTask()) != 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) ||
(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() 。
beforeExecute(Thread t, Runnable r)与afterExecute(Runnable r, Throwable t)并未在类中有解决业务的逻辑，即能够通过继承线程池的形式来重写这两个办法，这样就可能对工作的执行进行监控。
processWorkerExit
- 从While循环体中能够晓得，当线程运行时出现异常，那么都会退出循环，进入到processWorkerExit()
- 从getTask()取得后果为null，则也会进到processWorkerExit()
getTask()
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())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
// Are workers subject to culling?
//如果设置了allowCoreThreadTimeOut(true)
//或者以后运行的工作数大于设置的外围线程数
// timed = true
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
/** ------------------------以上的操作跟之前相似----------------------- */
/** ------------------------关键在于上面的代码------------------------- */
/** ------------------------从阻塞队列中获取工作----------------------- */
try {
Runnable r = timed ?
//对于阻塞队列，poll(long timeout, TimeUnit unit) 将会在规定的工夫内去工作
//如果没取到就返回null
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
//take会始终阻塞，期待工作的增加
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
线程池可能保障始终期待工作而不被销毁，其实就是进入了阻塞状态
ThreadPoolExecutor.processWorkerExit()
/**
* @param completedAbruptly
*/
private void processWorkerExit(Worker w, boolean completedAbruptly) {
if (completedAbruptly) //如果忽然被打断，工作线程数不会被缩小
decrementWorkerCount();
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
completedTaskCount += w.completedTasks;
workers.remove(w);
} finally {
mainLock.unlock();
}
tryTerminate();
int c = ctl.get();
//判断运行状态是否在STOP之前
if (runStateLessThan(c, STOP)) {
if (!completedAbruptly) {//失常退出，也就是task == null
int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
if (min == 0 && ! workQueue.isEmpty())
min = 1;
if (workerCountOf(c) >= min)
return; // replacement not needed
}
//新增一个工作线程，代替原来的工作线程
addWorker(null, false);
}
}
线程池敞开
能够通过调用线程池的shutdown或shutdownNow办法来敞开线程池。它们的原理是遍历线程池中的工作线程， 而后一一调用线程的interrupt办法来中断线程，所以无奈响应中断的工作可能永远无奈终止。然而它们存在肯定的区别， shutdownNow首先将线程池的状态设置成STOP，而后尝试进行所有的正在执行或暂停工作的线程，并返回期待执行工作的列表，而 shutdown只是将线程池的状态设置成SHUTDOWN状态，而后中断所有没有正在执行工作的线程。
只有调用了这两个敞开办法中的任意一个，isShutdown办法就会返回true。当所有的工作都已敞开后，才示意线程池敞开胜利， 这时调用isTerminaed办法会返回true。至于应该调用哪一种办法来敞开线程池，应该由提交到线程池的工作个性决定， 通常调用shutdown办法来敞开线程池，如果工作不肯定要执行完，则能够调用shutdownNow办法。
shutdown
当调用shutdown办法时，线程池将不会再接管新的工作，而后将先前放在队列中的工作执行实现。
public void shutdown() {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
//查看权限
checkShutdownAccess();
//CAS 更新线程池状态
advanceRunState(SHUTDOWN);
//中断所有闲暇的线程
interruptIdleWorkers();
//敞开，此处是do nothing
onShutdown();
} finally {
mainLock.unlock();
}
//尝试完结，下面代码已剖析
tryTerminate();
}
shutdownNow
立刻进行所有的执行工作，并将队列中的工作返回
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;
}
总结
- 线程池优先应用corePoolSize的数量执行工作工作
- 如果超过corePoolSize，队列入队
- 超过队列，应用maximumPoolSize-corePoolSize的线程解决，这部分线程超时不干活就销毁掉。
- 每个线程执行完结的时候，会判断以后的工作线程和工作数，如果工作数多，就会创立空线程从队列拿工作。
- 线程池执行实现，不会主动销毁，须要手工shutdown，批改线程池状态，中断所有线程。
调配线程池大小的根据
从以下几个角度思考
- 工作的性质：CPU密集型工作、IO密集型工作和混合型工作。
- 工作的优先级：高、中和低。
- 工作的执行工夫：长、中和短。
- 工作的依赖性：是否依赖其余系统资源，如数据库连贯。
性质不同的工作能够用不同规模的线程池离开解决。CPU密集型工作应配置尽可能小的线程，如配置cpu个数 +1个线程的线程池。 因为IO密集型工作线程并不是始终在执行工作，则应配置尽可能多的线程，如2*cpu个数 。混合型的工作，如果能够拆分， 将其拆分成一个CPU密集型工作和一个IO密集型工作，只有这两个工作执行的工夫相差不是太大，那么合成后执行的吞吐量 将高于串行执行的吞吐量。如果这两个工作执行工夫相差太大，则没必要进行合成。能够通过 Runtime.getRuntime().availableProcessors()办法取得以后设施的CPU个数。 优先级不同的工作能够应用优先级队列PriorityBlockingQueue来解决。它能够让优先级高的工作先执行。
执行工夫不同的工作能够交给不同规模的线程池来解决，或者能够应用优先级队列，让执行工夫短的工作先执行。
依赖数据库连接池的工作，因为线程提交SQL后须要期待数据库返回后果，期待的工夫越长，则CPU闲暇工夫就越长，那么线程数应该设置得越大， 这样能力更好地利用CPU。
应用有界队列
有界队列能减少零碎的稳定性和预警能力，能够依据须要设大一点儿，比方几千。有一次，咱们零碎里后台任务线程池的队列和线程池全满了， 一直抛出摈弃工作的异样，通过排查发现是数据库呈现了问题，导致执行SQL变得十分迟缓， 因为后台任务线程池里的工作全是须要向数据库查问和插入数据的，所以导致线程池里的工作线程全副阻塞，工作积压在线程池里。 如果过后咱们设置成无界队列，那么线程池的队列就会越来越多，有可能会撑满内存，导致整个零碎不可用，而不只是后台任务呈现问题。 当然，咱们的零碎所有的工作是用独自的服务器部署的，咱们应用不同规模的线程池实现不同类型的工作，然而呈现这样问题时也会影响到其余工作。
线程池监控
如果在零碎中大量应用线程池，则有必要对线程池进行监控，不便在呈现问题时，能够依据线程池的应用情况疾速定位问题。 能够通过线程池提供的参数进行监控，在监控线程池的时候能够应用以下属性。
- taskCount：线程池须要执行的工作数量。
- completedTaskCount：线程池在运行过程中已实现的工作数量，小于或等于taskCount。
- largestPoolSize：线程池里已经创立过的最大线程数量。通过这个数据能够晓得线程池是否已经满过。如该数值等于线程池的最大大小， 则示意线程池已经满过。
- getPoolSize：线程池的线程数量。如果线程池不销毁的话，线程池里的线程不会主动销毁，所以这个大小只增不减。
- getActiveCount：获取流动的线程数。
- 通过扩大线程池进行监控。能够通过继承线程池来自定义线程池，重写线程池的beforeExecute、afterExecute和terminated办法， 也能够在工作执行前、执行后和线程池敞开前执行一些代码来进行监控。例如，监控工作的均匀执行工夫、最大执行工夫和最小执行工夫等。 这几个办法在线程池里是空办法。
关注微信公众号:【入门小站】,解锁更多知识点
搞代码网（gaodaima.com）提供的所有资源部分来自互联网，如果有侵犯您的版权或其他权益，请说明详细缘由并提供版权或权益证明然后发送到邮箱[email protected]，我们会在看到邮件的第一时间内为您处理，或直接联系QQ：872152909。本网站采用BY-NC-SA协议进行授权
转载请注明原文链接：关于java:Java线程池ThreadPoolExecutor源码分析
转载请注明原文链接：关于java:Java线程池ThreadPoolExecutor源码分析
