继承关系
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 11111111111111111111111111111private static final int CAPACITY = (1 << COUNT_BITS) - 1;//运行中 111 00000000000000000000000000000private static final int RUNNING = -1 << COUNT_BITS;//敞开 000 00000000000000000000000000000private static final int SHUTDOWN = 0 << COUNT_BITS;//进行 001 00000000000000000000000000000private static final int STOP = 1 << COUNT_BITS;//整顿 010 00000000000000000000000000000private static final int TIDYING = 2 << COUNT_BITS;//终止 011 00000000000000000000000000000private 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(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办法， 也能够在工作执行前、执行后和线程池敞开前执行一些代码来进行监控。例如，监控工作的均匀执行工夫、最大执行工夫和最小执行工夫等。 这几个办法在线程池里是空办法。
关注微信公众号:【入门小站】,解锁更多知识点