前言

我曾經(jīng)對(duì)自己的小弟說(shuō)，如果你實(shí)在搞不清楚什么時(shí)候用HashMap，什么時(shí)候用ConcurrentHashMap，那么就用后者，你的代碼bug會(huì)很少。

他問(wèn)我：ConcurrentHashMap是什么？ -.-

編程不是炫技。大多數(shù)情況下，怎么把代碼寫(xiě)簡(jiǎn)單，才是能力。

多線(xiàn)程生來(lái)就是復(fù)雜的，也是容易出錯(cuò)的。一些難以理解的概念，要規(guī)避。本文不講基礎(chǔ)知識(shí)，因?yàn)槟闶掷锞陀衘dk的源碼。

線(xiàn)程

Thread

第一類(lèi)就是Thread類(lèi)。大家都知道有兩種實(shí)現(xiàn)方式。第一可以繼承Thread覆蓋它的run方法；第二種是實(shí)現(xiàn)Runnable接口，實(shí)現(xiàn)它的run方法；而第三種創(chuàng)建線(xiàn)程的方法，就是通過(guò)線(xiàn)程池。

我們的具體代碼實(shí)現(xiàn)，就放在run方法中。

我們關(guān)注兩種情況。一個(gè)是線(xiàn)程退出條件，一個(gè)是異常處理情況。

線(xiàn)程退出

有的run方法執(zhí)行完成后，線(xiàn)程就會(huì)退出。但有的run方法是永遠(yuǎn)不會(huì)結(jié)束的。結(jié)束一個(gè)線(xiàn)程肯定不是通過(guò)Thread.stop()方法，這個(gè)方法已經(jīng)在java1.2版本就廢棄了。所以我們大體有兩種方式控制線(xiàn)程。

定義退出標(biāo)志放在while中

代碼一般長(zhǎng)這樣。

private volatile boolean flag= true;
public void run() {
 while (flag) {
 }
}

標(biāo)志一般使用volatile進(jìn)行修飾，使其讀可見(jiàn)，然后通過(guò)設(shè)置這個(gè)值來(lái)控制線(xiàn)程的運(yùn)行，這已經(jīng)成了約定俗成的套路。

使用interrupt方法終止線(xiàn)程

類(lèi)似這種。

while（！isInterrupted（））{……}

對(duì)于InterruptedException，比如Thread.sleep所拋出的，我們一般是補(bǔ)獲它，然后靜悄悄的忽略。中斷允許一個(gè)可取消任務(wù)來(lái)清理正在進(jìn)行的工作，然后通知其他任務(wù)它要被取消，最后才終止，在這種情況下，此類(lèi)異常需要被仔細(xì)處理。

interrupt方法不一定會(huì)真正”中斷”線(xiàn)程，它只是一種協(xié)作機(jī)制。interrupt方法通常不能中斷一些處于阻塞狀態(tài)的I/O操作。比如寫(xiě)文件，或者socket傳輸?shù)?。這種情況，需要同時(shí)調(diào)用正在阻塞操作的close方法，才能夠正常退出。

interrupt系列使用時(shí)候一定要注意，會(huì)引入bug，甚至死鎖。

異常處理

java中會(huì)拋出兩種異常。一種是必須要捕獲的，比如InterruptedException，否則無(wú)法通過(guò)編譯；另外一種是可以處理也可以不處理的，比如NullPointerException等。

在我們的任務(wù)運(yùn)行中，很有可能拋出這兩種異常。對(duì)于第一種異常，是必須放在try,catch中的。但第二種異常如果不去處理的話(huà)，會(huì)影響任務(wù)的正常運(yùn)行。

有很多同學(xué)在處理循環(huán)的任務(wù)時(shí)，沒(méi)有捕獲一些隱式的異常，造成任務(wù)在遇到異常的情況下，并不能繼續(xù)執(zhí)行下去。如果不能確定異常的種類(lèi)，可以直接捕獲Exception或者更通用的Throwable。

while（！isInterrupted（））{
 try{
 ……
 }catch(Exception ex){
 ……
 }
}

同步方式

java中實(shí)現(xiàn)同步的方式有很多，大體分為以下幾種。

• synchronized 關(guān)鍵字
• wait、notify等
• Concurrent包中的ReentrantLock
• volatile關(guān)鍵字
• ThreadLocal局部變量

生產(chǎn)者、消費(fèi)者是wait、notify最典型的應(yīng)用場(chǎng)景，這些函數(shù)的調(diào)用，是必須要放在synchronized代碼塊里才能夠正常運(yùn)行的。它們同信號(hào)量一樣，大多數(shù)情況下屬于炫技，對(duì)代碼的可讀性影響較大，不推薦。關(guān)于ObjectMonitor相關(guān)的幾個(gè)函數(shù)，只要搞懂下面的圖，就基本ok了。

使用ReentrantLock最容易發(fā)生錯(cuò)誤的就是忘記在finally代碼塊里關(guān)閉鎖。大多數(shù)同步場(chǎng)景下，使用Lock就足夠了，而且它還有讀寫(xiě)鎖的概念進(jìn)行粒度上的控制。我們一般都使用非公平鎖，讓任務(wù)自由競(jìng)爭(zhēng)。非公平鎖性能高于公平鎖性能，非公平鎖能更充分的利用cpu的時(shí)間片，盡量的減少cpu空閑的狀態(tài)時(shí)間。非公平鎖還會(huì)造成餓死現(xiàn)象：有些任務(wù)一直獲取不到鎖。

synchronized通過(guò)鎖升級(jí)機(jī)制，速度不見(jiàn)得就比lock慢。而且，通過(guò)jstack，能夠方便的看到其堆棧，使用還是比較廣泛。

volatile總是能保證變量的讀可見(jiàn)，但它的目標(biāo)是基本類(lèi)型和它鎖的基本對(duì)象。假如是它修飾的是集合類(lèi)，比如Map，那么它保證的讀可見(jiàn)是map的引用，而不是map對(duì)象，這點(diǎn)一定要注意。

synchronized和volatile都體現(xiàn)在字節(jié)碼上（monitorenter、monitorexit），主要是加入了內(nèi)存屏障。而Lock，是純粹的java api。

ThreadLocal很方便，每個(gè)線(xiàn)程一份數(shù)據(jù)，也很安全，但要注意內(nèi)存泄露。假如線(xiàn)程存活時(shí)間長(zhǎng)，我們要保證每次使用完ThreadLocal，都調(diào)用它的remove()方法（具體來(lái)說(shuō)是expungeStaleEntry），來(lái)清除數(shù)據(jù)。

關(guān)于Concurrent包

concurrent包是在AQS的基礎(chǔ)上搭建起來(lái)的，AQS提供了一種實(shí)現(xiàn)阻塞鎖和一系列依賴(lài)FIFO等待隊(duì)列的同步器的框架。

線(xiàn)程池

最全的線(xiàn)程池大概有7個(gè)參數(shù)，想要合理使用線(xiàn)程池，肯定不會(huì)不會(huì)放過(guò)這些參數(shù)的優(yōu)化。

線(xiàn)程池參數(shù)

concurrent包最常用的就是線(xiàn)程池，平常工作建議直接使用線(xiàn)程池，Thread類(lèi)就可以降低優(yōu)先級(jí)了。我們常用的主要有newSingleThreadExecutor、newFixedThreadPool、newCachedThreadPool、調(diào)度等，使用Executors工廠(chǎng)類(lèi)創(chuàng)建。

newSingleThreadExecutor可以用于快速創(chuàng)建一個(gè)異步線(xiàn)程，非常方便。而newCachedThreadPool永遠(yuǎn)不要用在高并發(fā)的線(xiàn)上環(huán)境，它用的是無(wú)界隊(duì)列對(duì)任務(wù)進(jìn)行緩沖，可能會(huì)擠爆你的內(nèi)存。

我習(xí)慣性自定義ThreadPoolExecutor，也就是參數(shù)最全的那個(gè)。

public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) 

假如我的任務(wù)可以預(yù)估，corePoolSize，maximumPoolSize一般都設(shè)成一樣大的，然后存活時(shí)間設(shè)的特別的長(zhǎng)。可以避免線(xiàn)程頻繁創(chuàng)建、關(guān)閉的開(kāi)銷(xiāo)。I/O密集型和CPU密集型的應(yīng)用線(xiàn)程開(kāi)的大小是不一樣的，一般I/O密集型的應(yīng)用線(xiàn)程就可以開(kāi)的多一些。

threadFactory我一般也會(huì)定義一個(gè)，主要是給線(xiàn)程們起一個(gè)名字。這樣，在使用jstack等一些工具的時(shí)候，能夠直觀的看到我所創(chuàng)建的線(xiàn)程。

監(jiān)控

高并發(fā)下的線(xiàn)程池，最好能夠監(jiān)控起來(lái)?？梢允褂萌罩?、存儲(chǔ)等方式保存下來(lái)，對(duì)后續(xù)的問(wèn)題排查幫助很大。

通常，可以通過(guò)繼承ThreadPoolExecutor，覆蓋beforeExecute、afterExecute、terminated方法，達(dá)到對(duì)線(xiàn)程行為的控制和監(jiān)控。

線(xiàn)程池飽和策略

最容易被遺忘的可能就是線(xiàn)程的飽和策略了。也就是線(xiàn)程和緩沖隊(duì)列的空間全部用完了，新加入的任務(wù)將如何處置。jdk默認(rèn)實(shí)現(xiàn)了4種策略，默認(rèn)實(shí)現(xiàn)的是AbortPolicy，也就是直接拋出異常。下面介紹其他幾種。

DiscardPolicy 比abort更加激進(jìn)，直接丟掉任務(wù)，連異常信息都沒(méi)有。

CallerRunsPolicy 由調(diào)用的線(xiàn)程來(lái)處理這個(gè)任務(wù)。比如一個(gè)web應(yīng)用中，線(xiàn)程池資源占滿(mǎn)后，新進(jìn)的任務(wù)將會(huì)在tomcat線(xiàn)程中運(yùn)行。這種方式能夠延緩部分任務(wù)的執(zhí)行壓力，但在更多情況下，會(huì)直接阻塞主線(xiàn)程的運(yùn)行。

DiscardOldestPolicy 丟棄隊(duì)列最前面的任務(wù)，然后重新嘗試執(zhí)行任務(wù)（重復(fù)此過(guò)程）。

很多情況下，這些飽和策略可能并不能滿(mǎn)足你的需求，你可以自定義自己的策略，比如將任務(wù)持久化到一些存儲(chǔ)中。

阻塞隊(duì)列

阻塞隊(duì)列會(huì)對(duì)當(dāng)前的線(xiàn)程進(jìn)行阻塞。當(dāng)隊(duì)列中有元素后，被阻塞的線(xiàn)程會(huì)自動(dòng)被喚醒，這極大的提高的編碼的靈活性，非常方便。在并發(fā)編程中，一般推薦使用阻塞隊(duì)列，這樣實(shí)現(xiàn)可以盡量地避免程序出現(xiàn)意外的錯(cuò)誤。阻塞隊(duì)列使用最經(jīng)典的場(chǎng)景就是socket數(shù)據(jù)的讀取、解析，讀數(shù)據(jù)的線(xiàn)程不斷將數(shù)據(jù)放入隊(duì)列，解析線(xiàn)程不斷從隊(duì)列取數(shù)據(jù)進(jìn)行處理。

ArrayBlockingQueue對(duì)訪(fǎng)問(wèn)者的調(diào)用默認(rèn)是不公平的，我們可以通過(guò)設(shè)置構(gòu)造方法參數(shù)將其改成公平阻塞隊(duì)列。

LinkedBlockingQueue隊(duì)列的默認(rèn)最大長(zhǎng)度為Integer.MAX_VALUE，這在用做線(xiàn)程池隊(duì)列的時(shí)候，會(huì)比較危險(xiǎn)。

SynchronousQueue是一個(gè)不存儲(chǔ)元素的阻塞隊(duì)列。每一個(gè)put操作必須等待一個(gè)take操作，否則不能繼續(xù)添加元素。隊(duì)列本身不存儲(chǔ)任何元素，吞吐量非常高。對(duì)于提交的任務(wù)，如果有空閑線(xiàn)程，則使用空閑線(xiàn)程來(lái)處理；否則新建一個(gè)線(xiàn)程來(lái)處理任務(wù)”。它更像是一個(gè)管道，在一些通訊框架中（比如rpc），通常用來(lái)快速處理某個(gè)請(qǐng)求，應(yīng)用較為廣泛。

DelayQueue是一個(gè)支持延時(shí)獲取元素的無(wú)界阻塞隊(duì)列。放入DelayQueue的對(duì)象需要實(shí)現(xiàn)Delayed接口，主要是提供一個(gè)延遲的時(shí)間，以及用于延遲隊(duì)列內(nèi)部比較排序。這種方式通常能夠比大多數(shù)非阻塞的while循環(huán)更加節(jié)省cpu資源。

另外還有PriorityBlockingQueue和LinkedTransferQueue等，根據(jù)字面意思就能猜測(cè)它的用途。在線(xiàn)程池的構(gòu)造參數(shù)中，我們使用的隊(duì)列，一定要注意其特性和邊界。比如，即使是最簡(jiǎn)單的newFixedThreadPool，在某些場(chǎng)景下，也是不安全的，因?yàn)樗褂昧藷o(wú)界隊(duì)列。

CountDownLatch

假如有一堆接口A-Y，每個(gè)接口的耗時(shí)最大是200ms，最小是100ms。

我的一個(gè)服務(wù)，需要提供一個(gè)接口Z，調(diào)用A-Y接口對(duì)結(jié)果進(jìn)行聚合。接口的調(diào)用沒(méi)有順序需求，接口Z如何在300ms內(nèi)返回這些數(shù)據(jù)？

此類(lèi)問(wèn)題典型的還有賽馬問(wèn)題，只有通過(guò)并行計(jì)算才能完成問(wèn)題。歸結(jié)起來(lái)可以分為兩類(lèi)：

• 實(shí)現(xiàn)任務(wù)的并行性
• 開(kāi)始執(zhí)行前等待n個(gè)線(xiàn)程完成任務(wù)

在concurrent包出現(xiàn)之前，需要手工的編寫(xiě)這些同步過(guò)程，非常復(fù)雜?，F(xiàn)在就可以使用CountDownLatch和CyclicBarrier進(jìn)行便捷的編碼。

CountDownLatch是通過(guò)一個(gè)計(jì)數(shù)器來(lái)實(shí)現(xiàn)的，計(jì)數(shù)器的初始值為線(xiàn)程的數(shù)量。每當(dāng)一個(gè)線(xiàn)程完成了自己的任務(wù)后，計(jì)數(shù)器的值就會(huì)減1。當(dāng)計(jì)數(shù)器值到達(dá)0時(shí)，它表示所有的線(xiàn)程已經(jīng)完成了任務(wù)，然后在閉鎖上等待的線(xiàn)程就可以恢復(fù)執(zhí)行任務(wù)。

CyclicBarrier與其類(lèi)似，可以實(shí)現(xiàn)同樣的功能。不過(guò)在日常的工作中，使用CountDownLatch會(huì)更頻繁一些。

信號(hào)量

Semaphore雖然有一些應(yīng)用場(chǎng)景，但大部分屬于炫技，在編碼中應(yīng)該盡量少用。

信號(hào)量可以實(shí)現(xiàn)限流的功能，但它只是常用限流方式的一種。其他兩種是漏桶算法、令牌桶算法。

hystrix的熔斷功能，也有使用信號(hào)量進(jìn)行資源的控制。

Lock && Condition

在Java中，對(duì)于Lock和Condition可以理解為對(duì)傳統(tǒng)的synchronized和wait/notify機(jī)制的替代。concurrent包中的許多阻塞隊(duì)列，就是使用Condition實(shí)現(xiàn)的。

但這些類(lèi)和函數(shù)對(duì)于初中級(jí)碼農(nóng)來(lái)說(shuō)，難以理解，容易產(chǎn)生bug，應(yīng)該在業(yè)務(wù)代碼中嚴(yán)格禁止。但在網(wǎng)絡(luò)編程、或者一些框架類(lèi)工程中，這些功能是必須的，萬(wàn)不可將這部分的工作隨便分配給某個(gè)小弟。

End

不管是wait、notify，還是同步關(guān)鍵字或者鎖，能不用就不用，因?yàn)樗鼈儠?huì)引發(fā)程序的復(fù)雜性。最好的方式，是直接使用concurrent包所提供的機(jī)制，來(lái)規(guī)避一些編碼方面的問(wèn)題。

concurrent包中的CAS概念，在一定程度上算是無(wú)鎖的一種實(shí)現(xiàn)。更專(zhuān)業(yè)的有類(lèi)似disruptor的無(wú)鎖隊(duì)列框架，但它依然是建立在CAS的編程模型上的。近些年，類(lèi)似AKKA這樣的事件驅(qū)動(dòng)模型正在走紅，但編程模型簡(jiǎn)單，不代表實(shí)現(xiàn)簡(jiǎn)單，背后的工作依然需要多線(xiàn)程去協(xié)調(diào)。

golang引入?yún)f(xié)程(coroutine)概念以后，對(duì)多線(xiàn)程加入了更加輕量級(jí)的補(bǔ)充。java中可以通過(guò)javaagent技術(shù)加載quasar補(bǔ)充一些功能，但我覺(jué)得你不會(huì)為了這丁點(diǎn)效率去犧牲編碼的可讀性。

總結(jié)

以上就是這篇文章的全部?jī)?nèi)容了，希望本文的內(nèi)容對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，謝謝大家對(duì)腳本之家的支持。

最新評(píng)論