3年工作經(jīng)驗(yàn)，工作中還不會使用多線程？阿里P6:別慌，我都總結(jié)好了
掌握線程池是后端程序員的基本要求，相信大家求職面試過程中，幾乎都會被問到有關(guān)于線程池的問題。我在網(wǎng)上搜集了幾道經(jīng)典的線程池面試題，并以此為切入點(diǎn)，談?wù)勎覍€程池的理解。如果有哪里理解不正確，非常希望大家指出，接下來大家一起分析學(xué)習(xí)吧。
經(jīng)典面試題
- 面試問題1：Java的線程池說一下，各個(gè)參數(shù)的作用，如何進(jìn)行的?
- 面試問題2：按線程池內(nèi)部機(jī)制，當(dāng)提交新任務(wù)時(shí)，有哪些異常要考慮。
- 面試問題3：線程池都有哪幾種工作隊(duì)列？
- 面試問題4：使用無界隊(duì)列的線程池會導(dǎo)致內(nèi)存飆升嗎？
- 面試問題5：說說幾種常見的線程池及使用場景?
線程池概念
線程池： 簡單理解，它就是一個(gè)管理線程的池子。
- 它幫我們管理線程，避免增加創(chuàng)建線程和銷毀線程的資源損耗。因?yàn)榫€程其實(shí)也是一個(gè)對象，創(chuàng)建一個(gè)對象，需要經(jīng)過類加載過程，銷毀一個(gè)對象，需要走GC垃圾回收流程，都是需要資源開銷的。
- 提高響應(yīng)速度。 如果任務(wù)到達(dá)了，相對于從線程池拿線程，重新去創(chuàng)建一條線程執(zhí)行，速度肯定慢很多。
- 重復(fù)利用。 線程用完，再放回池子，可以達(dá)到重復(fù)利用的效果，節(jié)省資源。
線程池的創(chuàng)建
線程池可以通過ThreadPoolExecutor來創(chuàng)建，我們來看一下它的構(gòu)造函數(shù)：
- public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,long keepAliveTime,TimeUnit unit,
- BlockingQueue<Runnable> workQueue,
- ThreadFactory threadFactory,
- RejectedExecutionHandler handler)
- 復(fù)制代碼
幾個(gè)核心參數(shù)的作用：
- corePoolSize： 線程池核心線程數(shù)最大值
- maximumPoolSize： 線程池最大線程數(shù)大小
- keepAliveTime： 線程池中非核心線程空閑的存活時(shí)間大小
- unit： 線程空閑存活時(shí)間單位
- workQueue： 存放任務(wù)的阻塞隊(duì)列
- threadFactory： 用于設(shè)置創(chuàng)建線程的工廠，可以給創(chuàng)建的線程設(shè)置有意義的名字，可方便排查問題。
- handler： 線城池的飽和策略事件，主要有四種類型。
任務(wù)執(zhí)行
線程池執(zhí)行流程，即對應(yīng)execute()方法：
- 提交一個(gè)任務(wù)，線程池里存活的核心線程數(shù)小于線程數(shù)corePoolSize時(shí)，線程池會創(chuàng)建一個(gè)核心線程去處理提交的任務(wù)。
- 如果線程池核心線程數(shù)已滿，即線程數(shù)已經(jīng)等于corePoolSize，一個(gè)新提交的任務(wù)，會被放進(jìn)任務(wù)隊(duì)列workQueue排隊(duì)等待執(zhí)行。
- 當(dāng)線程池里面存活的線程數(shù)已經(jīng)等于corePoolSize了,并且任務(wù)隊(duì)列workQueue也滿，判斷線程數(shù)是否達(dá)到maximumPoolSize，即最大線程數(shù)是否已滿，如果沒到達(dá)，創(chuàng)建一個(gè)非核心線程執(zhí)行提交的任務(wù)。
- 如果當(dāng)前的線程數(shù)達(dá)到了maximumPoolSize，還有新的任務(wù)過來的話，直接采用拒絕策略處理。
四種拒絕策略
- AbortPolicy(拋出一個(gè)異常，默認(rèn)的)
- DiscardPolicy(直接丟棄任務(wù))
- DiscardOldestPolicy（丟棄隊(duì)列里最老的任務(wù)，將當(dāng)前這個(gè)任務(wù)繼續(xù)提交給線程池）
- CallerRunsPolicy（交給線程池調(diào)用所在的線程進(jìn)行處理)
為了形象描述線程池執(zhí)行，我打個(gè)比喻：
- 核心線程比作公司正式員工
- 非核心線程比作外包員工
- 阻塞隊(duì)列比作需求池
- 提交任務(wù)比作提需求
- 當(dāng)產(chǎn)品提個(gè)需求，正式員工（核心線程）先接需求（執(zhí)行任務(wù)）
- 如果正式員工都有需求在做，即核心線程數(shù)已滿），產(chǎn)品就把需求先放需求池（阻塞隊(duì)列）。
- 如果需求池(阻塞隊(duì)列)也滿了，但是這時(shí)候產(chǎn)品繼續(xù)提需求,怎么辦呢？那就請外包（非核心線程）來做。
- 如果所有員工（最大線程數(shù)也滿了）都有需求在做了，那就執(zhí)行拒絕策略。
- 如果外包員工把需求做完了，它經(jīng)過一段（keepAliveTime）空閑時(shí)間，就離開公司了。
好的，到這里。面試問題1->Java的線程池說一下，各個(gè)參數(shù)的作用，如何進(jìn)行的? 是否已經(jīng)迎刃而解啦， 我覺得這個(gè)問題，回答：線程池構(gòu)造函數(shù)的corePoolSize,maximumPoolSize等參數(shù)，并且能描述清楚線程池的執(zhí)行流程 就差不多啦。
線程池異常處理
在使用線程池處理任務(wù)的時(shí)候，任務(wù)代碼可能拋出RuntimeException，拋出異常后，線程池可能捕獲它，也可能創(chuàng)建一個(gè)新的線程來代替異常的線程，我們可能無法感知任務(wù)出現(xiàn)了異常，因此我們需要考慮線程池異常情況。
當(dāng)提交新任務(wù)時(shí)，異常如何處理?
我們先來看一段代碼：
- ExecutorService threadPool = Executors.newFixedThreadPool(5);
- for (int i = 0; i < 5; i++) {
- threadPool.submit(() -> {
- System.out.println("current thread name" + Thread.currentThread().getName());
- Object object = null;
- System.out.print("result## "+object.toString());
- });
- }
- 復(fù)制代碼
顯然，這段代碼會有異常，我們再來看看執(zhí)行結(jié)果
雖然沒有結(jié)果輸出，但是沒有拋出異常，所以我們無法感知任務(wù)出現(xiàn)了異常，所以需要添加try/catch。 如下圖：
OK，線程的異常處理，我們可以直接try...catch捕獲。線程池exec.submit(runnable)的執(zhí)行流程
通過debug上面有異常的submit方法（建議大家也去debug看一下,圖上的每個(gè)方法內(nèi)部是我打斷點(diǎn)的地方），處理有異常submit方法的主要執(zhí)行流程圖：
- //構(gòu)造feature對象
- /**
- * @throws RejectedExecutionException {@inheritDoc}
- * @throws NullPointerException {@inheritDoc}
- */
- public Future<?> submit(Runnable task) {
- if (task == null) throw new NullPointerException();
- RunnableFuture<Void> ftask = newTaskFor(task, null);
- execute(ftask);
- return ftask;
- }
- protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
- return new FutureTask<T>(runnable, value);
- }
- public FutureTask(Runnable runnable, V result) {
- this.callable = Executors.callable(runnable, result);
- this.state = NEW; // ensure visibility of callable
- }
- public static <T> Callable<T> callable(Runnable task, T result) {
- if (task == null)
- throw new NullPointerException();
- return new RunnableAdapter<T>(task, result);
- }
- //線程池執(zhí)行
- 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);
- }
- //捕獲異常
- public void run() {
- if (state != NEW ||
- !UNSAFE.compareAndSwapObject(this, runnerOffset,
- null, Thread.currentThread()))
- return;
- try {
- Callable<V> c = callable;
- if (c != null && state == NEW) {
- V result;
- boolean ran;
- try {
- result = c.call();
- ran = true;
- } catch (Throwable ex) {
- result = null;
- ran = false;
- setException(ex);
- }
- if (ran)
- set(result);
- }
- } finally {
- // runner must be non-null until state is settled to
- // prevent concurrent calls to run()
- runner = null;
- // state must be re-read after nulling runner to prevent
- // leaked interrupts
- int s = state;
- if (s >= INTERRUPTING)
- handlePossibleCancellationInterrupt(s);
- }
- 制代碼
通過以上分析，submit執(zhí)行的任務(wù)，可以通過Future對象的get方法接收拋出的異常，再進(jìn)行處理。 我們再通過一個(gè)demo，看一下Future對象的get方法處理異常的姿勢，如下圖：
其他兩種處理線程池異常方案
除了以上1.在任務(wù)代碼try/catch捕獲異常，2.通過Future對象的get方法接收拋出的異常，再處理兩種方案外，還有以上兩種方案：
3.為工作者線程設(shè)置UncaughtExceptionHandler，在uncaughtException方法中處理異常
我們直接看這樣實(shí)現(xiàn)的正確姿勢：
- ExecutorService threadPool = Executors.newFixedThreadPool(1, r -> {
- Thread t = new Thread(r);
- t.setUncaughtExceptionHandler(
- (t1, e) -> {
- System.out.println(t1.getName() + "線程拋出的異常"+e);
- });
- return t;
- });
- threadPool.execute(()->{
- Object object = null;
- System.out.print("result## " + object.toString());
- });
- 復(fù)制代碼
運(yùn)行結(jié)果：
4.重寫ThreadPoolExecutor的afterExecute方法，處理傳遞的異常引用
這是jdk文檔的一個(gè)demo：
- class ExtendedExecutor extends ThreadPoolExecutor {
- // 這可是jdk文檔里面給的例子。。
- protected void afterExecute(Runnable r, Throwable t) {
- super.afterExecute(r, t);
- if (t == null && r instanceof Future<?>) {
- try {
- Object result = ((Future<?>) r).get();
- } catch (CancellationException ce) {
- t = ce;
- } catch (ExecutionException ee) {
- t = ee.getCause();
- } catch (InterruptedException ie) {
- Thread.currentThread().interrupt(); // ignore/reset
- }
- }
- if (t != null)
- System.out.println(t);
- }
- }}
- 復(fù)制代碼
因此，被問到線程池異常處理，如何回答？
。線程池的工作隊(duì)列
線程池都有哪幾種工作隊(duì)列？
- ArrayBlockingQueue
- LinkedBlockingQueue
- DelayQueue
- PriorityBlockingQueue
- SynchronousQueue
ArrayBlockingQueue
ArrayBlockingQueue（有界隊(duì)列）是一個(gè)用數(shù)組實(shí)現(xiàn)的有界阻塞隊(duì)列，按FIFO排序量。
LinkedBlockingQueue
LinkedBlockingQueue（可設(shè)置容量隊(duì)列）基于鏈表結(jié)構(gòu)的阻塞隊(duì)列，按FIFO排序任務(wù)，容量可以選擇進(jìn)行設(shè)置，不設(shè)置的話，將是一個(gè)無邊界的阻塞隊(duì)列，最大長度為Integer.MAX_VALUE，吞吐量通常要高于ArrayBlockingQuene；newFixedThreadPool線程池使用了這個(gè)隊(duì)列
DelayQueue
DelayQueue（延遲隊(duì)列）是一個(gè)任務(wù)定時(shí)周期的延遲執(zhí)行的隊(duì)列。根據(jù)指定的執(zhí)行時(shí)間從小到大排序，否則根據(jù)插入到隊(duì)列的先后排序。newScheduledThreadPool線程池使用了這個(gè)隊(duì)列。
PriorityBlockingQueue
PriorityBlockingQueue（優(yōu)先級隊(duì)列）是具有優(yōu)先級的無界阻塞隊(duì)列；
SynchronousQueue
SynchronousQueue（同步隊(duì)列）一個(gè)不存儲元素的阻塞隊(duì)列，每個(gè)插入操作必須等到另一個(gè)線程調(diào)用移除操作，否則插入操作一直處于阻塞狀態(tài)，吞吐量通常要高于LinkedBlockingQuene，newCachedThreadPool線程池使用了這個(gè)隊(duì)列。
針對面試題：線程池都有哪幾種工作隊(duì)列？ 我覺得，回答以上幾種ArrayBlockingQueue，LinkedBlockingQueue，SynchronousQueue等，說出它們的特點(diǎn)，并結(jié)合使用到對應(yīng)隊(duì)列的常用線程池(如newFixedThreadPool線程池使用LinkedBlockingQueue)，進(jìn)行展開闡述， 就可以啦。
幾種常用的線程池
- newFixedThreadPool (固定數(shù)目線程的線程池)
- newCachedThreadPool(可緩存線程的線程池)
- newSingleThreadExecutor(單線程的線程池)
- newScheduledThreadPool(定時(shí)及周期執(zhí)行的線程池)
newFixedThreadPool
- public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
- return new ThreadPoolExecutor(nThreads, nThreads,
- 0L, TimeUnit.MILLISECONDS,
- new LinkedBlockingQueue<Runnable>(),
- threadFactory);
- }
- 復(fù)制代碼
線程池特點(diǎn)：
- 核心線程數(shù)和最大線程數(shù)大小一樣
- 沒有所謂的非空閑時(shí)間，即keepAliveTime為0
- 阻塞隊(duì)列為無界隊(duì)列LinkedBlockingQueue
工作機(jī)制：
- 提交任務(wù)
- 如果線程數(shù)少于核心線程，創(chuàng)建核心線程執(zhí)行任務(wù)
- 如果線程數(shù)等于核心線程，把任務(wù)添加到LinkedBlockingQueue阻塞隊(duì)列
- 如果線程執(zhí)行完任務(wù)，去阻塞隊(duì)列取任務(wù)，繼續(xù)執(zhí)行。
實(shí)例代碼
- ExecutorService executor = Executors.newFixedThreadPool(10);
- for (int i = 0; i < Integer.MAX_VALUE; i++) {
- executor.execute(()->{
- try {
- Thread.sleep(10000);
- } catch (InterruptedException e) {
- //do nothing
- }
- });
- 復(fù)制代碼
IDE指定JVM參數(shù)：-Xmx8m -Xms8m :
run以上代碼，會拋出OOM：
因此，面試題：使用無界隊(duì)列的線程池會導(dǎo)致內(nèi)存飆升嗎？
答案 ：會的，newFixedThreadPool使用了無界的阻塞隊(duì)列LinkedBlockingQueue，如果線程獲取一個(gè)任務(wù)后，任務(wù)的執(zhí)行時(shí)間比較長(比如，上面demo設(shè)置了10秒)，會導(dǎo)致隊(duì)列的任務(wù)越積越多，導(dǎo)致機(jī)器內(nèi)存使用不停飆升， 最終導(dǎo)致OOM。
使用場景
FixedThreadPool 適用于處理CPU密集型的任務(wù)，確保CPU在長期被工作線程使用的情況下，盡可能的少的分配線程，即適用執(zhí)行長期的任務(wù)。
newCachedThreadPool
- public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
- return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
- 60L, TimeUnit.SECONDS,
- new SynchronousQueue<Runnable>(),
- threadFactory);
- }
- 復(fù)制代碼
線程池特點(diǎn)：
- 核心線程數(shù)為0
- 最大線程數(shù)為Integer.MAX_VALUE
- 阻塞隊(duì)列是SynchronousQueue
- 非核心線程空閑存活時(shí)間為60秒
當(dāng)提交任務(wù)的速度大于處理任務(wù)的速度時(shí)，每次提交一個(gè)任務(wù)，就必然會創(chuàng)建一個(gè)線程。極端情況下會創(chuàng)建過多的線程，耗盡 CPU 和內(nèi)存資源。由于空閑 60 秒的線程會被終止，長時(shí)間保持空閑的 CachedThreadPool 不會占用任何資源。
工作機(jī)制
- 提交任務(wù)
- 因?yàn)闆]有核心線程，所以任務(wù)直接加到SynchronousQueue隊(duì)列。
- 判斷是否有空閑線程，如果有，就去取出任務(wù)執(zhí)行。
- 如果沒有空閑線程，就新建一個(gè)線程執(zhí)行。
- 執(zhí)行完任務(wù)的線程，還可以存活60秒，如果在這期間，接到任務(wù)，可以繼續(xù)活下去；否則，被銷毀。
實(shí)例代碼
- ExecutorService executor = Executors.newCachedThreadPool();
- for (int i = 0; i < 5; i++) {
- executor.execute(() -> {
- System.out.println(Thread.currentThread().getName()+"正在執(zhí)行");
- });
- }
- 復(fù)制代碼
運(yùn)行結(jié)果：
使用場景
用于并發(fā)執(zhí)行大量短期的小任務(wù)。
newSingleThreadExecutor
- public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
- return new FinalizableDelegatedExecutorService
- (new ThreadPoolExecutor(1, 1,
- 0L, TimeUnit.MILLISECONDS,
- new LinkedBlockingQueue<Runnable>(),
- threadFactory));
- }
- 制代碼
線程池特點(diǎn)
- 核心線程數(shù)為1
- 最大線程數(shù)也為1
- 阻塞隊(duì)列是LinkedBlockingQueue
- keepAliveTime為0
工作機(jī)制
- 提交任務(wù)
- 線程池是否有一條線程在，如果沒有，新建線程執(zhí)行任務(wù)
- 如果有，講任務(wù)加到阻塞隊(duì)列
- 當(dāng)前的唯一線程，從隊(duì)列取任務(wù)，執(zhí)行完一個(gè)，再繼續(xù)取，一個(gè)人（一條線程）夜以繼日地干活。
實(shí)例代碼
- ExecutorService executor = Executors.newSingleThreadExecutor();
- for (int i = 0; i < 5; i++) {
- executor.execute(() -> {
- System.out.println(Thread.currentThread().getName()+"正在執(zhí)行");
- });
- }
- 復(fù)制代碼
運(yùn)行結(jié)果：
使用場景
適用于串行執(zhí)行任務(wù)的場景，一個(gè)任務(wù)一個(gè)任務(wù)地執(zhí)行。
newScheduledThreadPool
- public ScheduledThreadPoolExecutor(int corePoolSize) {
- super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
- new DelayedWorkQueue());
- }
- 復(fù)制代碼
線程池特點(diǎn)
- 最大線程數(shù)為Integer.MAX_VALUE
- 阻塞隊(duì)列是DelayedWorkQueue
- keepAliveTime為0
- scheduleAtFixedRate() ：按某種速率周期執(zhí)行
- scheduleWithFixedDelay()：在某個(gè)延遲后執(zhí)行
工作機(jī)制
- 添加一個(gè)任務(wù)
- 線程池中的線程從 DelayQueue 中取任務(wù)
- 線程從 DelayQueue 中獲取 time 大于等于當(dāng)前時(shí)間的task
- 執(zhí)行完后修改這個(gè) task 的 time 為下次被執(zhí)行的時(shí)間
- 這個(gè) task 放回DelayQueue隊(duì)列中
實(shí)例代碼
- /**
- 創(chuàng)建一個(gè)給定初始延遲的間隔性的任務(wù)，之后的下次執(zhí)行時(shí)間是上一次任務(wù)從執(zhí)行到結(jié)束所需要的時(shí)間+* 給定的間隔時(shí)間
- */
- ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
- scheduledExecutorService.scheduleWithFixedDelay(()->{
- System.out.println("current Time" + System.currentTimeMillis());
- System.out.println(Thread.currentThread().getName()+"正在執(zhí)行");
- }, 1, 3, TimeUnit.SECONDS);
- 復(fù)制代碼
運(yùn)行結(jié)果：
- /**
- 創(chuàng)建一個(gè)給定初始延遲的間隔性的任務(wù)，之后的每次任務(wù)執(zhí)行時(shí)間為 初始延遲 + N * delay(間隔)
- */
- ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
- scheduledExecutorService.scheduleAtFixedRate(()->{
- System.out.println("current Time" + System.currentTimeMillis());
- System.out.println(Thread.currentThread().getName()+"正在執(zhí)行");
- }, 1, 3, TimeUnit.SECONDS);;
- 復(fù)制代碼
使用場景
周期性執(zhí)行任務(wù)的場景，需要限制線程數(shù)量的場景
回到面試題：說說幾種常見的線程池及使用場景？
回答這四種經(jīng)典線程池 ：newFixedThreadPool，newSingleThreadExecutor，newCachedThreadPool，newScheduledThreadPool，分線程池特點(diǎn)，工作機(jī)制，使用場景分開描述，再分析可能存在的問題，比如newFixedThreadPool內(nèi)存飆升問題 即可
線程池狀態(tài)
線程池有這幾個(gè)狀態(tài)：RUNNING,SHUTDOWN,STOP,TIDYING,TERMINATED。
- //線程池狀態(tài)
- 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;
- 復(fù)制代碼
線程池各個(gè)狀態(tài)切換圖：
RUNNING
- 該狀態(tài)的線程池會接收新任務(wù)，并處理阻塞隊(duì)列中的任務(wù);
- 調(diào)用線程池的shutdown()方法，可以切換到SHUTDOWN狀態(tài);
- 調(diào)用線程池的shutdownNow()方法，可以切換到STOP狀態(tài);
SHUTDOWN
- 該狀態(tài)的線程池不會接收新任務(wù)，但會處理阻塞隊(duì)列中的任務(wù)；
- 隊(duì)列為空，并且線程池中執(zhí)行的任務(wù)也為空,進(jìn)入TIDYING狀態(tài);
STOP
- 該狀態(tài)的線程不會接收新任務(wù)，也不會處理阻塞隊(duì)列中的任務(wù)，而且會中斷正在運(yùn)行的任務(wù)；
- 線程池中執(zhí)行的任務(wù)為空,進(jìn)入TIDYING狀態(tài);
TIDYING
- 該狀態(tài)表明所有的任務(wù)已經(jīng)運(yùn)行終止，記錄的任務(wù)數(shù)量為0。
- terminated()執(zhí)行完畢，進(jìn)入TERMINATED狀態(tài)
TERMINATED
- 該狀態(tài)表示線程池徹底終止
