背景
關於 Java 的線程池我想大家肯定不會陌生，在工作中或者自己平時的學習中多多少少都會用到，那你真的有了解過底層的實現原理嗎？還是說只停留在用的階段呢？而且關於 Java 線程池也是在面試中的一個高頻的面試題，就像 HashMap 的實現原理一樣，基本上面試必問，估計都已經被問爛大街了。
題外話：HashMap 的實現原理真的已經被問爛了，在我自身的多次面試中都不知道被問了幾遍了，有的時候想想很奇怪，為什么這個被問的爛大街的問題還是會一直被問呢？但是從面試官的角度來想一下，如果一個被問的都爛大街的問題你都不好好准備對待，那怎么能好好的對待工作呢（個人愚見）。
常用的幾種線程池
我們先來看下常用的幾種線程池的創建方式，以及底層采用的實現原理
單個線程: Executors.newSingleThreadExecutor();
public static ExecutorService newSingleThreadExecutor() {return new FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>()));}
緩存線程: Executors.newCachedThreadPool();
public static ExecutorService newCachedThreadPool() {return new ThreadPoolExecutor(0, Integer.MAX_VALUE,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());}
固定線程Executors.newFixedThreadPool(2);
public static ExecutorService newFixedThreadPool(int nThreads) {return new ThreadPoolExecutor(nThreads, nThreads,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>());}
定時線程: Executors.newScheduledThreadPool(3);（父類中）
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue) {this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,Executors.defaultThreadFactory(), defaultHandler);}
核心 ThreadPoolExecutor
通過上面的幾個線程池的底層實現，我們可以發現底層都是通過 ThreadPoolExecutor 類來實現的，只是參數不一樣，那我們就很有必要來看一下ThreadPoolExecutor 這個類了
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;}
通過 JDK 的源碼我們可以看到 ThreadPoolExecutor 在 Java 的 concurrent 包下面，並且有四個構造方法，下面依次介紹下各個參數的含義：
•corePoolSize: 核心線程數的大小•maximumPoolSize: 線程池中允許的最大線程數•keepAliveTime: 空閑線程允許的最大的存活時間•unit: 存活時間的單位•workQueue: 阻塞任務隊列•threadFactory: 線程工廠用來創建線程•handler: 拒絕策略，針對當隊列滿了時新來任務的處理方式
通過上面參數的分析，我們可以知道，單個線程的線程池就是線程池中只有一個線程負責任務，所以 corePoolSize 和 maximumPoolSize 的數值都是為 1；當這個線程出現任何異常后，線程池會自動創建一個線程，始終保持線程池中有且只有一個存活的線程。而且其他線程池也只是參數的設置不一樣而已。 我們還需要知道幾個常見的線程池類和接口的關系，以及一些方法，如下圖
ThreadPoolExecutor繼承AbstractExecutorService；AbstractExecutorService實現ExecutorService，ExecutorService繼承Executor
源碼分析
根據源碼可以發現整個線程池大致分為 3 個部分，1. 是創建 worker 線程，2. 添加任務到 workQueue; 3.worker 線程執行具體任務
創建 worker 線程，現在我們來看下核心的 execute(Runnable command) 方法，如果工作線程小於指定的核心線程數時會嘗試去創建新的線程，
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);}
再看下addWorker(Runnable firstTask, boolean core) 方法
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;if (compareAndIncrementWorkerCount(c))break retry;c = ctl.get(); // Re-read ctlif (runStateOf(c) != rs)continue retry;}}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 startablethrow 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;}
添加任務到 workQueue，這個阻塞隊列內部的方法
public boolean offer(E e) {if (e == null) throw new NullPointerException();final AtomicInteger count = this.count;if (count.get() == capacity)return false;int c = -1;Node<E> node = new Node<E>(e);final ReentrantLock putLock = this.putLock;putLock.lock();try {if (count.get() < capacity) {enqueue(node);c = count.getAndIncrement();if (c + 1 < capacity)notFull.signal();}} finally {putLock.unlock();}if (c == 0)signalNotEmpty();return c >= 0;}
worker 線程執行具體任務，阻塞或者超時去獲取隊列中的任務，進行執行
final void runWorker(Worker w) {Thread wt = Thread.currentThread();Runnable task = w.firstTask;w.firstTask = null;w.unlock(); // allow interruptsboolean 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++;w.unlock();}}completedAbruptly = false;} finally {processWorkerExit(w, completedAbruptly);}}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?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;}}}
在剛剛創建線程池的時候，內部線程的數量是 0，當首個任務進行添加的時候，會根據參數的配置進行線程的創建，並隨着任務數的增加，會逐漸創建新的線程直到不能創建新的線程為止。不能創建新的線程后，會將來的任務存放到阻塞隊列中，讓空閑的線程去處理。當沒有空閑線程並且隊列滿了時候就會采用拒絕策略去丟棄或者其他策略來處理。 拒絕策略主要有四種，不同的拒絕策略有不同的使用場景，需要根據情況決定使用。
•CallerRunsPolicy : 調用線程處理任務•AbortPolicy : 拋出異常•DiscardPolicy : 直接丟棄•DiscardOldestPolicy : 丟棄隊列中最老的任務，執行新任務
小結
線程池在工作中的使用必不可少，如何優雅的使用線程池能很大程度的提升性能和效率。根據實際的應用場景，配置合適的線程池參數可以很大的提升項目的性能，也可以充分利用服務器的性能。
Java 極客技術公眾號，是由一群熱愛 Java 開發的技術人組建成立，專注分享原創、高質量的 Java 文章。如果您覺得我們的文章還不錯，請幫忙贊賞、在看、轉發支持，鼓勵我們分享出更好的文章。
關注公眾號，大家可以在公眾號后台回復“博客園”，免費獲得作者 Java 知識體系/面試必看資料。
