10问10答:你真的了解线程池吗?
簡介:?《Java開發手冊》中強調,線程資源必須通過線程池提供,而創建線程池必須使用ThreadPoolExecutor。手冊主要強調利用線程池避免兩個問題,一是線程過渡切換,二是避免請求過多時造成OOM。但是如果參數配置錯誤,還是會引發上面的兩個問題。所以本節我們主要是討論ThreadPoolExecutor的一些技術細節,并且給出幾個常用的最佳實踐建議。
作者 | 風樓
來源 | 阿里技術公眾號
《Java開發手冊》中強調,線程資源必須通過線程池提供,而創建線程池必須使用ThreadPoolExecutor。手冊主要強調利用線程池避免兩個問題,一是線程過渡切換,二是避免請求過多時造成OOM。但是如果參數配置錯誤,還是會引發上面的兩個問題。所以本節我們主要是討論ThreadPoolExecutor的一些技術細節,并且給出幾個常用的最佳實踐建議。
我在查找資料的過程中,發現有些問題存在爭議。后面發現,一部分原因是因為不同JDK版本的現實是有差異的。因此,下面的分析是基于當下最常用的版本JDK1.8,并且對于存在爭議的問題,我們分析源碼,源碼才是最準確的。
1 corePoolSize=0會怎么樣
這是一個爭議點。我發現大部分博文,不論是國內的還是國外的,都是這樣回答這個問題的:
- 提交任務后,先判斷當前池中線程數是否小于corePoolSize,如果小于,則創建新線程執行這個任務。
- 否者,判斷等待隊列是否已滿,如果沒有滿,則添加到等待隊列。
- 否者,判斷當前池中線程數是否大于maximumPoolSize,如果大于則拒絕。
- 否者,創建一個新的線程執行這個任務。
按照上面的描述,如果corePoolSize=0,則會判斷等待隊列的容量,如果還有容量,則排隊,并且不會創建新的線程。
—— 但其實,這是老版本的實現方式,從1.6之后,實現方式就變了。我們直接看execute的源碼(submit也依賴它),我備注出了關鍵一行:
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);// 注意這一行代碼,添加到等待隊列成功后,判斷當前池內線程數是否為0,如果是則創建一個firstTask為null的worker,這個worker會從等待隊列中獲取任務并執行。else if (workerCountOf(recheck) == 0)addWorker(null, false);}else if (!addWorker(command, false))reject(command);- 線程池提交任務后,首先判斷當前池中線程數是否小于corePoolSize。
- 如果小于則嘗試創建新的線程執行該任務;否則嘗試添加到等待隊列。
- 如果添加隊列成功,判斷當前池內線程數是否為0,如果是則創建一個firstTask為null的worker,這個worker會從等待隊列中獲取任務并執行。
- 如果添加到等待隊列失敗,一般是隊列已滿,才會再嘗試創建新的線程。
- 但在創建之前需要與maximumPoolSize比較,如果小于則創建成功。
- 否則執行拒絕策略。
答
上述問題需區分JDK版本。在1.6版本之后,如果corePoolSize=0,提交任務時如果線程池為空,則會立即創建一個線程來執行任務(先排隊再獲取);如果提交任務的時候,線程池不為空,則先在等待隊列中排隊,只有隊列滿了才會創建新線程。
所以,優化在于,在隊列沒有滿的這段時間內,會有一個線程在消費提交的任務;1.6之前的實現是,必須等隊列滿了之后,才開始消費。
2 線程池創建之后,會立即創建核心線程么
之前有人問過我這個問題,因為他發現應用中有些Bean創建了線程池,但是這個Bean一般情況下用不到,所以咨詢我是否需要把這個線程池注釋掉,以減少應用運行時的線程數(該應用運行時線程過多。)
答
不會。從上面的源碼可以看出,在剛剛創建ThreadPoolExecutor的時候,線程并不會立即啟動,而是要等到有任務提交時才會啟動,除非調用了prestartCoreThread/prestartAllCoreThreads事先啟動核心線程。
- prestartCoreThread:Starts a core thread, causing it to idly wait for work. This overrides the default policy of starting core threads only when new tasks are executed.
- prestartAllCoreThreads:Starts all core threads.
3 核心線程永遠不會銷毀么
這個問題有點tricky。首先我們要明確一下概念,雖然在JavaDoc中也使用了“core/non-core threads”這樣的描述,但其實這是一個動態的概念,JDK并沒有給一部分線程打上“core”的標記,做什么特殊化的處理。這個問題我認為想要探討的是閑置線程終結策略的問題。
在JDK1.6之前,線程池會盡量保持corePoolSize個核心線程,即使這些線程閑置了很長時間。這一點曾被開發者詬病,所以從JDK1.6開始,提供了方法allowsCoreThreadTimeOut,如果傳參為true,則允許閑置的核心線程被終止。
請注意這種策略和corePoolSize=0的區別。我總結的區別是:
- corePoolSize=0:在一般情況下只使用一個線程消費任務,只有當并發請求特別多、等待隊列都滿了之后,才開始用多線程。
- allowsCoreThreadTimeOut=true && corePoolSize>1:在一般情況下就開始使用多線程(corePoolSize個),當并發請求特別多,等待隊列都滿了之后,繼續加大線程數。但是當請求沒有的時候,允許核心線程也終止。
所以corePoolSize=0的效果,基本等同于allowsCoreThreadTimeOut=true && corePoolSize=1,但實現細節其實不同。
答
在JDK1.6之后,如果allowsCoreThreadTimeOut=true,核心線程也可以被終止。
4 如何保證線程不被銷毀
首先我們要明確一下線程池模型。線程池有個內部類Worker,它實現了Runnable接口,首先,它自己要run起來。然后它會在合適的時候獲取我們提交的Runnable任務,然后調用任務的run()接口。一個Worker不終止的話可以不斷執行任務。
我們前面說的“線程池中的線程”,其實就是Worker;等待隊列中的元素,是我們提交的Runnable任務。
每一個Worker在創建出來的時候,會調用它本身的run()方法,實現是runWorker(this),這個實現的核心是一個while循環,這個循環不結束,Worker線程就不會終止,就是這個基本邏輯。
- 在這個while條件中,有個getTask()方法是核心中的核心,它所做的事情就是從等待隊列中取出任務來執行:
- 如果沒有達到corePoolSize,則創建的Worker在執行完它承接的任務后,會用workQueue.take()取任務、注意,這個接口是阻塞接口,如果取不到任務,Worker線程一直阻塞。
- 如果超過了corePoolSize,或者allowCoreThreadTimeOut,一個Worker在空閑了之后,會用workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)取任務。注意,這個接口只阻塞等待keepAliveTime時間,超過這個時間返回null,則Worker的while循環執行結束,則被終止了。
答
實現方式非常巧妙,核心線程(Worker)即使一直空閑也不終止,是通過workQueue.take()實現的,它會一直阻塞到從等待隊列中取到新的任務。非核心線程空閑指定時間后終止是通過workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS)實現的,一個空閑的Worker只等待keepAliveTime,如果還沒有取到任務則循環終止,線程也就運行結束了。
引申思考
Worker本身就是個線程,它再調用我們傳入的Runnable.run(),會啟動一個子線程么?如果你還沒有答案,再回想一下Runnable和Thread的關系。
5 空閑線程過多會有什么問題
籠統地回答是會占用內存,我們分析一下占用了哪些內存。首先,比較普通的一部分,一個線程的內存模型:
- 虛擬機棧
- 本地方法棧
- 程序計數器
我想額外強調是下面這幾個內存占用,需要小心:
- ThreadLocal:業務代碼是否使用了ThreadLocal?就算沒有,Spring框架中也大量使用了ThreadLocal,你所在公司的框架可能也是一樣。
- 局部變量:線程處于阻塞狀態,肯定還有棧幀沒有出棧,棧幀中有局部變量表,凡是被局部變量表引用的內存都不能回收。所以如果這個線程創建了比較大的局部變量,那么這一部分內存無法GC。
- TLAB機制:如果你的應用線程數處于高位,那么新的線程初始化可能因為Eden沒有足夠的空間分配TLAB而觸發YoungGC。
答
線程池保持空閑的核心線程是它的默認配置,一般來講是沒有問題的,因為它占用的內存一般不大。怕的就是業務代碼中使用ThreadLocal緩存的數據過大又不清理。
如果你的應用線程數處于高位,那么需要觀察一下YoungGC的情況,估算一下Eden大小是否足夠。如果不夠的話,可能要謹慎地創建新線程,并且讓空閑的線程終止;必要的時候,可能需要對JVM進行調參。
6 keepAliveTime=0會怎么樣
這也是個爭議點。有的博文說等于0表示空閑線程永遠不會終止,有的說表示執行完立刻終止。還有的說等于-1表示空閑線程永遠不會終止。其實稍微看一下源碼知道了,這里我直接拋出答案。
答
在JDK1.8中,keepAliveTime=0表示非核心線程執行完立刻終止。
默認情況下,keepAliveTime小于0,初始化的時候才會報錯;但如果allowsCoreThreadTimeOut,keepAliveTime必須大于0,不然初始化報錯。
7 怎么進行異常處理
很多代碼的寫法,我們都習慣按照常見范式去編寫,而沒有去思考為什么。比如:
- 如果我們使用execute()提交任務,我們一般要在Runable任務的代碼加上try-catch進行異常處理。
- 如果我們使用submit()提交任務,我們一般要在主線程中,對Future.get()進行try-catch進行異常處理。
—— 但是在上面,我提到過,submit()底層實現依賴execute(),兩者應該統一呀,為什么有差異呢?下面再扒一扒submit()的源碼,它的實現蠻有意思。
首先,ThreadPoolExecutor中沒有submit的代碼,而是在它的父類AbstractExecutorService中,有三個submit的重載方法,代碼非常簡單,關鍵代碼就兩行:
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,所以我才說submit底層依賴execute。通過查看這里execute的實現,我們不難發現,它就是ThreadPoolExecutor中的實現,所以,造成submit和execute的差異化的代碼,不在這。那么造成差異的一定在newTaskFor方法中。這個方法也就new了一個FutureTask而已,FutureTask實現RunnableFuture接口,RunnableFuture接口繼承Runnable接口和Future接口。而Callable只是FutureTask的一個成員變量。
所以講到這里,就有另一個Java基礎知識點:Callable和Future的關系。我們一般用Callable編寫任務代碼,Future是異步返回對象,通過它的get方法,阻塞式地獲取結果。FutureTask的核心代碼就是實現了Future接口,也就是get方法的實現:
public V get() throws InterruptedException, ExecutionException {int s = state;if (s <= COMPLETING)// 核心代碼s = awaitDone(false, 0L);return report(s);}private int awaitDone(boolean timed, long nanos)throws InterruptedException {final long deadline = timed ? System.nanoTime() + nanos : 0L;WaitNode q = null;boolean queued = false;// 死循環for (;;) {if (Thread.interrupted()) {removeWaiter(q);throw new InterruptedException();}int s = state;// 只有任務的狀態是’已完成‘,才會跳出死循環if (s > COMPLETING) {if (q != null)q.thread = null;return s;}else if (s == COMPLETING) // cannot time out yetThread.yield();else if (q == null)q = new WaitNode();else if (!queued)queued = UNSAFE.compareAndSwapObject(this, waitersOffset,q.next = waiters, q);else if (timed) {nanos = deadline - System.nanoTime();if (nanos <= 0L) {removeWaiter(q);return state;}LockSupport.parkNanos(this, nanos);}elseLockSupport.park(this);}}get的核心實現是有個awaitDone方法,這是一個死循環,只有任務的狀態是“已完成”,才會跳出死循環;否則會依賴UNSAFE包下的LockSupport.park原語進行阻塞,等待LockSupport.unpark信號量。而這個信號量只有當運行結束獲得結果、或者出現異常的情況下,才會發出來。分別對應方法set和setException。這就是異步執行、阻塞獲取的原理,扯得有點遠了。
回到最初我們的疑問,為什么submit之后,通過get方法可以獲取到異常?原因是FutureTask有一個Object類型的outcome成員變量,用來記錄執行結果。這個結果可以是傳入的泛型,也可以是Throwable異常:
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 interruptsint s = state;if (s >= INTERRUPTING)handlePossibleCancellationInterrupt(s);}}// get方法中依賴的,報告執行結果private V report(int s) throws ExecutionException {Object x = outcome;if (s == NORMAL)return (V)x;if (s >= CANCELLED)throw new CancellationException();throw new ExecutionException((Throwable)x);}FutureTask的另一個巧妙的地方就是借用RunnableAdapter內部類,將submit的Runnable封裝成Callable。所以就算你submit的是Runnable,一樣可以用get獲取到異常。
答
- 不論是用execute還是submit,都可以自己在業務代碼上加try-catch進行異常處理。我一般喜歡使用這種方式,因為我喜歡對不同業務場景的異常進行差異化處理,至少打不一樣的日志吧。
- 如果是execute,還可以自定義線程池,繼承ThreadPoolExecutor并復寫其afterExecute(Runnable r, Throwable t)方法。
- 或者實現Thread.UncaughtExceptionHandler接口,實現void uncaughtException(Thread t, Throwable e);方法,并將該handler傳遞給線程池的ThreadFactory。
- 但是注意,afterExecute和UncaughtExceptionHandler都不適用submit。因為通過上面的FutureTask.run()不難發現,它自己對Throwable進行了try-catch,封裝到了outcome屬性,所以底層方法execute的Worker是拿不到異常信息的。
8 線程池需不需要關閉
答
一般來講,線程池的生命周期跟隨服務的生命周期。如果一個服務(Service)停止服務了,那么需要調用shutdown方法進行關閉。所以ExecutorService.shutdown在Java以及一些中間件的源碼中,是封裝在Service的shutdown方法內的。
如果是Server端不重啟就不停止提供服務,我認為是不需要特殊處理的。
9 shutdown和shutdownNow的區別
答
- shutdown => 平緩關閉,等待所有已添加到線程池中的任務執行完再關閉。
- shutdownNow => 立刻關閉,停止正在執行的任務,并返回隊列中未執行的任務。
本來想分析一下兩者的源碼的,但是發現本文的篇幅已經過長了,源碼也貼了不少。感興趣的朋友自己看一下即可。
10 Spring中有哪些和ThreadPoolExecutor類似的工具
答
這里我想著重強調的就是SimpleAsyncTaskExecutor,Spring中使用的@Async注解,底層就是基于SimpleAsyncTaskExecutor去執行任務,只不過它不是線程池,而是每次都新開一個線程。
另外想要強調的是Executor接口。Java初學者容易想當然的以為Executor結尾的類就是一個線程池,而上面的都是反例。我們可以在JDK的execute方法上看到這個注釋:
/** * Executes the given command at some time in the future. The command * may execute in a new thread, in a pooled thread, or in the calling * thread, at the discretion of the {@code Executor} implementation. */所以,它的職責并不是提供一個線程池的接口,而是提供一個“將來執行命令”的接口。真正能代表線程池意義的,是ThreadPoolExecutor類,而不是Executor接口。
最佳實踐總結
- 【強制】使用ThreadPoolExecutor的構造函數聲明線程池,避免使用Executors類的 newFixedThreadPool和newCachedThreadPool。
- 【強制】 創建線程或線程池時請指定有意義的線程名稱,方便出錯時回溯。即threadFactory參數要構造好。
- 【建議】建議不同類別的業務用不同的線程池。
- 【建議】CPU密集型任務(N+1):這種任務消耗的主要是CPU資源,可以將線程數設置為N(CPU核心數)+1,比CPU核心數多出來的一個線程是為了防止線程偶發的缺頁中斷,或者其它原因導致的任務暫停而帶來的影響。一旦任務暫停,CPU就會處于空閑狀態,而在這種情況下多出來的一個線程就可以充分利用CPU的空閑時間。
- 【建議】I/O密集型任務(2N):這種任務應用起來,系統會用大部分的時間來處理I/O交互,而線程在處理I/O的時間段內不會占用CPU來處理,這時就可以將CPU交出給其它線程使用。因此在I/O密集型任務的應用中,我們可以多配置一些線程,具體的計算方法是2N。
- 【建議】workQueue不要使用無界隊列,盡量使用有界隊列。避免大量任務等待,造成OOM。
- 【建議】如果是資源緊張的應用,使用allowsCoreThreadTimeOut可以提高資源利用率。
- 【建議】雖然使用線程池有多種異常處理的方式,但在任務代碼中,使用try-catch最通用,也能給不同任務的異常處理做精細化。
- 【建議】對于資源緊張的應用,如果擔心線程池資源使用不當,可以利用ThreadPoolExecutor的API實現簡單的監控,然后進行分析和優化。
線程池初始化示例:
private static final ThreadPoolExecutor pool;static {ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("po-detail-pool-%d").build();pool = new ThreadPoolExecutor(4, 8, 60L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(512),threadFactory, new ThreadPoolExecutor.AbortPolicy());pool.allowCoreThreadTimeOut(true);}- threadFactory:給出帶業務語義的線程命名。
- corePoolSize:快速啟動4個線程處理該業務,是足夠的。
- maximumPoolSize:IO密集型業務,我的服務器是4C8G的,所以4*2=8。
- keepAliveTime:服務器資源緊張,讓空閑的線程快速釋放。
- pool.allowCoreThreadTimeOut(true):也是為了在可以的時候,讓線程釋放,釋放資源。
- workQueue:一個任務的執行時長在100~300ms,業務高峰期8個線程,按照10s超時(已經很高了)。10s鐘,8個線程,可以處理10?1000ms / 200ms?8 = 400個任務左右,往上再取一點,512已經很多了。
- handler:極端情況下,一些任務只能丟棄,保護服務端。
原文鏈接
本文為阿里云原創內容,未經允許不得轉載。
總結
以上是生活随笔為你收集整理的10问10答:你真的了解线程池吗?的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: 如何将实时计算 Flink 与自身环境打
- 下一篇: 多中心容灾实践:如何实现真正的异地多活?