1. 初识线程池
线程池解决了如下两个问题
当执行大量的异步任务时，线程池可以减少每个任务的调用切换开销从而提高应用性能
对执行的线程，和要被执行的任务，提供了管理的方法
此外每个线程池还维护了一些基本统计信息，比如已完成任务的数量
2. ThreadPoolExecutor的简单使用
我们创建一个线程池对象ThreadPoolExecutor，让线程池执行10个打印任务，输出当前任务名称以及线程的名称
public class ThreadPoolExecutorTest { public static void main(String[] args) { ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2,5,0L, TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(10)); for(int i=0;i<10;i++){ final int num = i; threadPoolExecutor.execute(new Runnable() { @Override public void run() { try { TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("execute task "+num+" in "+Thread.currentThread().getName()); } }); } } }
代码还是蛮简单的。首先创建一个ThreadPoolExecutor对象，然后调用ThreadPoolExecutor.execute(Runnable runnable)方法。输出结果如下。我们可以观察到线程池启动了两个线程pool-1-thread-1和pool-1-thread-2来执行任务。那么这里提两个问题
该线程池最多能启动5个线程，为什么线程池只启动了2个线程来执行任务？
如果把for循环次数由10改成100，线程池会启动几个线程呢？
如果你还不是很有把握回答这两个问题那么请接着看下文分析吧。
execute task 0 in pool-1-thread-1 execute task 1 in pool-1-thread-2 execute task 2 in pool-1-thread-1 execute task 3 in pool-1-thread-2 execute task 5 in pool-1-thread-2 execute task 4 in pool-1-thread-1 execute task 6 in pool-1-thread-2 execute task 7 in pool-1-thread-1 execute task 8 in pool-1-thread-2 execute task 9 in pool-1-thread-1
3. ThreadPoolExecutor的成员变量和构造函数
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)
线程池中的线程被逻辑分为两类(核心线程和非核心线程)。核心线程如果被启动了，一般情况下是会一直活着的(除非allowCoreThreadTimeOut被设置成true)，非核心线程在keepAlivTime时间范围内，如果没有任务执行会被系统中断回收掉。
那么核心线程是在什么时候被创建并启动的呢？非核心线程又是在什么时候被创建并启动的呢？(稍安勿躁，后面会给出答案)
workQueue是指任务的集合。我们可以把ThreadPoolExecutor简单的认为它只有两个比较重要的属性 线程的集合和任务的集合。
private final HashSet<Worker> workers = new HashSet<>(); private final BlockingQueue<Runnable> workQueue;
workers对象正是线程的集合，后面我们会对Worker对象做一个详细的讲解
workQueue是一个阻塞队列。阻塞队列的概念就是如果队列满了那么put操作会阻塞，如果队列为空那么take操作会阻塞。
线程池的工作原理我们可以简单地认为是往workQueue里面put任务，系统会合理的调度workers中的线程来处理workQueue中的任务。
下面我们来对构造函数的各个参数做一个详细的介绍
corePoolSize:核心线程个数，当新的任务通过execute(java.lang.Runnable)方法提交到线程池中，如果线程池的线程个数小于corePoolSize设定的值时，不管线程池中的线程是否空闲，都会新建线程来处理该任务
maximumPoolSize:线程池最多允许的线程数。如果线程池中的线程数量大于等于corePoolSize而且小于等于maximumPoolSize，如果此时workQueue已满，则会创建新的线程来处理任务(如果线程数等于maximumPoolSize则会执行相应的拒绝策略)，反之如果workQueue未满，则将该任务put到workQueue中。
keepAliveTime:允许线程空闲的时间，如果空闲时间超过，而且线程池中的线程数比corePoolSize数量多，则会中断和回收空闲线程
unit:keepAliveTime的时间单位，秒或者毫秒等
workQueue:工作队列可以用来交付和保存提交到线程池中的工作任务，它和corePoolSize maximumPoolSize相互作用
处理任务的流程如下
如果线程池中的线程数量小于corePoolSize，线程池将新建新的线程处理该任务，而不是将任务入队列
如果线程池中的线程数量大于等于corePoolSize，线程池将优先选择将任务入队列
如果入队列失败(队列已满)，将创建新的线程处理任务，除非线程超过了maximumPoolSize，任务将被拒绝
三种不同的入队策略
直接交付，使用 SynchronousQueue直接交付任务，而不是把任务保存在队列中。
无界的队列，比如LinkedBlockingQueue，用无界队列构建的线程池，线程数永远不会超过corePoolSize
有界的队列，比如ArrayBlockingQueue，这种情况比较复杂，线程池同时受corePoolSize,maximumPoolSize,workQueue大小约束
threadFactory:主要是给线程池中的线程命名，以便查找问题
handler:线程池无法处理任务时的拒绝策略
4. execute(java.lang.Runnable)方法
public void execute(Runnable command) { if (command == null) throw new NullPointerException(); /* * Proceed in 3 steps: * * 1. If fewer than corePoolSize threads are running, try to * start a new thread with the given command as its first * task. The call to addWorker atomically checks runState and * workerCount, and so prevents false alarms that would add * threads when it shouldn't, by returning false. * * 2. If a task can be successfully queued, then we still need * to double-check whether we should have added a thread * (because existing ones died since last checking) or that * the pool shut down since entry into this method. So we * recheck state and if necessary roll back the enqueuing if * stopped, or start a new thread if there are none. * * 3. If we cannot queue task, then we try to add a new * thread. If it fails, we know we are shut down or saturated * and so reject the task. */ int c = ctl.get(); //1 如果线程池中的线程数小于corePoolSize,新建线程处理command if (workerCountOf(c) < corePoolSize) { if (addWorker(command, true)) return; c = ctl.get(); } //2 如果线程池中的线程数量大于等于corePoolSize，将任务入队列 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); } //3 如果任务入队列失败，创建非核心线程处理任务 else if (!addWorker(command, false)) reject(command);//4.如果创建非核心线程失败，拒绝该任务 }
如果线程池中的线程数小于corePoolSize,新建线程处理command
如果线程池中的线程数量大于等于corePoolSize，将任务入队列
如果任务入队列失败，创建非核心线程处理任务
如果创建非核心线程失败，拒绝该任务
5.Worker类（工作线程类）
private final class Worker extends AbstractQueuedSynchronizer implements Runnable { /** * This class will never be serialized, but we provide a * serialVersionUID to suppress a javac warning. */ private static final long serialVersionUID = 6138294804551838833L; /** Thread this worker is running in. Null if factory fails. */ final Thread thread; /** Initial task to run. Possibly null. */ Runnable firstTask; /** Per-thread task counter */ volatile long completedTasks; /** * Creates with given first task and thread from ThreadFactory. * @param firstTask the first task (null if none) */ Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } /** Delegates main run loop to outer runWorker. */ public void run() { runWorker(this); } final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { 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); } } }
Worker实现了Runnable接口，构造函数中this.thread = getThreadFactory().newThread(this)，启动线程将执行run方法
Worker是AbstractQueuedSynchronizer的子类，说明了Worker本身是一把锁。如何判断工作线程是否空闲？就是通过work.tryLock()来判断不信可以看下interruptIdleWorkers(boolean onlyOne)方法，该方法的功能是中断空闲的工作线程
private void interruptIdleWorkers(boolean onlyOne) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (Worker w : workers) { Thread t = w.thread; //如果w.tryLock()返回true表示该工作线程处于空闲状态 if (!t.isInterrupted() && w.tryLock()) { try { t.interrupt(); } catch (SecurityException ignore) { } finally { w.unlock(); } } if (onlyOne) break; } } finally { mainLock.unlock(); } }
我们来看下w.lock()调用的地方。在runWorker(Worker worker)方法中，在获取到任务去处理时，会调用w.lock()
final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); // allow interrupts boolean completedAbruptly = true; try { //从任务队列中获取任务，可能会阻塞，因为BlockingQueue是阻塞队列 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); } }
6. 工作线程什么时候启动的呢
在讲execute(Runnable runnable)的方法的时候，创建工作线程是通过addWorker(Runnable firstTask, boolean core)方法实现的
private boolean addWorker(Runnable firstTask, boolean core) { retry: for (;;) { int c = ctl.get(); int rs = runStateOf(c); //如果线程池被shutDown,直接返回 if (rs >= SHUTDOWN && ! (rs == SHUTDOWN && firstTask == null && ! workQueue.isEmpty())) return false; for (;;) { int wc = workerCountOf(c); if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) //如果线程数超过了限制直接返回false return false; 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 } } 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; 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); } return workerStarted; }
- 判断线程池是否满足条件，如果不满足返回false
- 创建Worker对象，并启动工作线程
7. keepAliveTime功能是如何实现的
如何实现在keepAliveTime时间内空闲，中断线程的呢，答案在getTask()中
private Runnable getTask() { boolean timedOut = false; // Did the last poll() time out? for (;;) { int c = ctl.get(); int rs = runStateOf(c); //1.如果调用了shutDownNow返回null //2.如果调用了shutDown而且workQueue没有任务返回null if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } int wc = workerCountOf(c); //如果允许杀死空闲的核心线程，或者线程数超过corePoolSize boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; } try { //如果指定时间内获取不到任务，返回null，线程将会结束 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } }
1.判断是否需要杀死空闲的线程
允许杀死核心线程 返回true
不允许杀死核心线程 判断线程数是否大于corePoolSize
2.通过workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)方法获取任务，如果超时，返回null
3.runWorker中getTask返回null，跳出循环，调用processWorkerExit(w, completedAbruptly)
8. shutdown和shutdownNow的区别
shutdown方法调用后，不会处理新的任务，但是会处理已经进入队列的任务
shutdownNow方法调用后，不会处理新的任务，而且不会处理已经进入队列的任务，而且会停止所有的工作线程
