導(dǎo)讀：之前寫了一系列關(guān)于并發(fā)編程的文章，也對(duì)今年的一些大型互聯(lián)網(wǎng)公司的并發(fā)編程面試題做了一個(gè)大致的匯總，今天，就來寫一下關(guān)于Java并發(fā)編程的總結(jié)與思考。

編寫優(yōu)質(zhì)的并發(fā)代碼是一件難度極高的事情。Java語言從第一版本開始內(nèi)置了對(duì)多線程的支持，這一點(diǎn)在當(dāng)年是非常了不起的，但是當(dāng)我們對(duì)并發(fā)編程有了更深刻的認(rèn)識(shí)和更多的實(shí)踐后，實(shí)現(xiàn)并發(fā)編程就有了更多的方案和更好的選擇。本文是對(duì)并發(fā)編程的一點(diǎn)總結(jié)和思考，同時(shí)也分享了Java 5以后的版本中如何編寫并發(fā)代碼的一點(diǎn)點(diǎn)經(jīng)驗(yàn)。

為什么需要并發(fā)

并發(fā)其實(shí)是一種解耦合的策略，它幫助我們把做什么（目標(biāo)）和什么時(shí)候做（時(shí)機(jī)）分開。這樣做可以明顯改進(jìn)應(yīng)用程序的吞吐量（獲得更多的CPU調(diào)度時(shí)間）和結(jié)構(gòu)（程序有多個(gè)部分在協(xié)同工作）。做過Java Web開發(fā)的人都知道，Java Web中的Servlet程序在Servlet容器的支持下采用單實(shí)例多線程的工作模式，Servlet容器幫助你處理了并發(fā)請(qǐng)求的問題。

誤解和正解

最常見的對(duì)并發(fā)編程的誤解有以下這些：

A. 并發(fā)總能改進(jìn)性能。 （真相：并發(fā)在CPU有很多空閑時(shí)間時(shí)能明顯改進(jìn)程序的性能，但當(dāng)線程數(shù)量較多的時(shí)候，線程間頻繁的調(diào)度切換反而會(huì)讓系統(tǒng)的性能下降）

B. 編寫并發(fā)程序無需修改原有的設(shè)計(jì)。 （真相：目的與時(shí)機(jī)的解耦往往會(huì)對(duì)系統(tǒng)結(jié)構(gòu)產(chǎn)生巨大的影響）

C. 在使用Web或EJB容器時(shí)不用關(guān)注并發(fā)問題。

（真相：只有了解了容器在做什么，才能更好的使用容器）

下面的這些說法才是對(duì)并發(fā)編程比較客觀的認(rèn)識(shí)：

A. 編寫并發(fā)程序會(huì)在代碼上增加額外的開銷。

B. 正確的并發(fā)是非常復(fù)雜的，即使對(duì)于很簡單的問題。

C. 并發(fā)中的缺陷因?yàn)椴灰字噩F(xiàn)也不容易被發(fā)現(xiàn)。

D. 并發(fā)往往需要對(duì)設(shè)計(jì)策略從根本上進(jìn)行修改。

并發(fā)編程的原則和技巧

1. 單一職責(zé)原則：分離并發(fā)相關(guān)代碼和其他代碼（并發(fā)相關(guān)代碼有自己的開發(fā)、修改和調(diào)優(yōu)生命周期）。

2. 限制數(shù)據(jù)作用域：兩個(gè)線程修改共享對(duì)象的同一字段時(shí)可能會(huì)相互干擾，導(dǎo)致不可預(yù)期的行為，解決方案之一是構(gòu)造臨界區(qū)，但是必須限制臨界區(qū)的數(shù)量。

3. 使用數(shù)據(jù)副本：數(shù)據(jù)副本是避免共享數(shù)據(jù)的好方法，復(fù)制出來的對(duì)象只是以只讀的方式對(duì)待。Java 5的java.util.concurrent包中增加一個(gè)名為CopyOnWriteArrayList的類，它是List接口的子類型，所以你可以認(rèn)為它是ArrayList的線程安全的版本，它使用了寫時(shí)復(fù)制的方式創(chuàng)建數(shù)據(jù)副本進(jìn)行操作來避免對(duì)共享數(shù)據(jù)并發(fā)訪問而引發(fā)的問題。

4. 線程應(yīng)盡可能獨(dú)立：讓線程存在于自己的世界中，不與其他線程共享數(shù)據(jù)。有過Java Web開發(fā)經(jīng)驗(yàn)的人都知道，Servlet就是以單實(shí)例多線程的方式工作，和每個(gè)請(qǐng)求相關(guān)的數(shù)據(jù)都是通過Servlet子類的service方法（或者是doGet或doPost方法）的參數(shù)傳入的。只要Servlet中的代碼只使用局部變量，Servlet就不會(huì)導(dǎo)致同步問題。Spring MVC的控制器也是這么做的，從請(qǐng)求中獲得的對(duì)象都是以方法的參數(shù)傳入而不是作為類的成員，很明顯Struts 2的做法就正好相反，因此Struts 2中作為控制器的Action類都是每個(gè)請(qǐng)求對(duì)應(yīng)一個(gè)實(shí)例。

Java 5以前的并發(fā)編程

Java的線程模型建立在搶占式線程調(diào)度的基礎(chǔ)上，也就是說：

1. 所有線程可以很容易的共享同一進(jìn)程中的對(duì)象。

2. 能夠引用這些對(duì)象的任何線程都可以修改這些對(duì)象。

3. 為了保護(hù)數(shù)據(jù)，對(duì)象可以被鎖住。

Java基于線程和鎖的并發(fā)過于底層，而且使用鎖很多時(shí)候都是很萬惡的，因?yàn)樗喈?dāng)于讓所有的并發(fā)都變成了排隊(duì)等待。

在Java 5以前，可以用synchronized關(guān)鍵字來實(shí)現(xiàn)鎖的功能，它可以用在代碼塊和方法上，表示在執(zhí)行整個(gè)代碼塊或方法之前線程必須取得合適的鎖。對(duì)于類的非靜態(tài)方法（成員方法）而言，這意味這要取得對(duì)象實(shí)例的鎖，對(duì)于類的靜態(tài)方法（類方法）而言，要取得類的Class對(duì)象的鎖，對(duì)于同步代碼塊，程序員可以指定要取得的是那個(gè)對(duì)象的鎖。

不管是同步代碼塊還是同步方法，每次只有一個(gè)線程可以進(jìn)入，如果其他線程試圖進(jìn)入（不管是同一同步塊還是不同的同步塊），JVM會(huì)將它們掛起（放入到等鎖池中）。這種結(jié)構(gòu)在并發(fā)理論中稱為臨界區(qū)（critical section）。這里我們可以對(duì)Java中用synchronized實(shí)現(xiàn)同步和鎖的功能做一個(gè)總結(jié)：

- 只能鎖定對(duì)象，不能鎖定基本數(shù)據(jù)類型。

- 被鎖定的對(duì)象數(shù)組中的單個(gè)對(duì)象不會(huì)被鎖定。

- 同步方法可以視為包含整個(gè)方法的synchronized(this) { ... }代碼塊。

- 靜態(tài)同步方法會(huì)鎖定它的Class對(duì)象。

- 內(nèi)部類的同步是獨(dú)立于外部類的。

- synchronized修飾符并不是方法簽名的組成部分，所以不能出現(xiàn)在接口的方法聲明中。

- 非同步的方法不關(guān)心鎖的狀態(tài)，它們?cè)谕椒椒ㄟ\(yùn)行時(shí)仍然可以得以運(yùn)行。

- synchronized實(shí)現(xiàn)的鎖是可重入的鎖。

在JVM內(nèi)部，為了提高效率，同時(shí)運(yùn)行的每個(gè)線程都會(huì)有它正在處理的數(shù)據(jù)的緩存副本，當(dāng)我們使用synchronzied進(jìn)行同步的時(shí)候，真正被同步的是在不同線程中表示被鎖定對(duì)象的內(nèi)存塊（副本數(shù)據(jù)會(huì)保持和主內(nèi)存的同步，現(xiàn)在知道為什么要用同步這個(gè)詞匯了吧），簡單的說就是在同步塊或同步方法執(zhí)行完后，對(duì)被鎖定的對(duì)象做的任何修改要在釋放鎖之前寫回到主內(nèi)存中；在進(jìn)入同步塊得到鎖之后，被鎖定對(duì)象的數(shù)據(jù)是從主內(nèi)存中讀出來的，持有鎖的線程的數(shù)據(jù)副本一定和主內(nèi)存中的數(shù)據(jù)視圖是同步的 。

在Java最初的版本中，就有一個(gè)叫volatile的關(guān)鍵字，它是一種簡單的同步的處理機(jī)制，因?yàn)楸籿olatile修飾的變量遵循以下規(guī)則：

- 變量的值在使用之前總會(huì)從主內(nèi)存中再讀取出來。

- 對(duì)變量值的修改總會(huì)在完成之后寫回到主內(nèi)存中。

使用volatile關(guān)鍵字可以在多線程環(huán)境下預(yù)防編譯器不正確的優(yōu)化假設(shè)（編譯器可能會(huì)將在一個(gè)線程中值不會(huì)發(fā)生改變的變量優(yōu)化成常量），但只有修改時(shí)不依賴當(dāng)前狀態(tài)（讀取時(shí)的值）的變量才應(yīng)該聲明為volatile變量。

不變模式也是并發(fā)編程時(shí)可以考慮的一種設(shè)計(jì)。讓對(duì)象的狀態(tài)是不變的，如果希望修改對(duì)象的狀態(tài)，就會(huì)創(chuàng)建對(duì)象的副本并將改變寫入副本而不改變?cè)瓉淼膶?duì)象，這樣就不會(huì)出現(xiàn)狀態(tài)不一致的情況，因此不變對(duì)象是線程安全的。Java中我們使用頻率極高的String類就采用了這樣的設(shè)計(jì)。如果對(duì)不變模式不熟悉，可以閱讀閻宏博士的《Java與模式》一書的第34章。說到這里你可能也體會(huì)到final關(guān)鍵字的重要意義了。

Java 5的并發(fā)編程

不管今后的Java向著何種方向發(fā)展或者滅忙，Java 5絕對(duì)是Java發(fā)展史中一個(gè)極其重要的版本，我們必須要感謝Doug Lea在Java 5中提供了他里程碑式的杰作java.util.concurrent包，它的出現(xiàn)讓Java的并發(fā)編程有了更多的選擇和更好的工作方式。Doug Lea的杰作主要包括以下內(nèi)容：

- 更好的線程安全的容器

- 線程池和相關(guān)的工具類

- 可選的非阻塞解決方案

- 顯示的鎖和信號(hào)量機(jī)制

原子類

Java 5中的java.util.concurrent包下面有一個(gè)atomic子包，其中有幾個(gè)以Atomic打頭的類，例如AtomicInteger和AtomicLong。它們利用了現(xiàn)代處理器的特性，可以用非阻塞的方式完成原子操作，代碼如下所示：

/**

ID序列生成器

*/publicclassIdGenerator{privatefinal AtomicLong sequenceNumber =newAtomicLong(0);publiclongnext(){returnsequenceNumber.getAndIncrement();}}

顯示鎖

基于synchronized關(guān)鍵字的鎖機(jī)制有以下問題：

- 鎖只有一種類型，而且對(duì)所有同步操作都是一樣的作用

- 鎖只能在代碼塊或方法開始的地方獲得，在結(jié)束的地方釋放

- 線程要么得到鎖，要么阻塞，沒有其他的可能性

Java 5對(duì)鎖機(jī)制進(jìn)行了重構(gòu)，提供了顯示的鎖，這樣可以在以下幾個(gè)方面提升鎖機(jī)制：

- 可以添加不同類型的鎖，例如讀取鎖和寫入鎖。

- 可以在一個(gè)方法中加鎖，在另一個(gè)方法中解鎖。

- 可以使用tryLock方式嘗試獲得鎖，如果得不到鎖可以等待、回退或者干點(diǎn)別的事情，當(dāng)然也可以在超時(shí)之后放棄操作。

顯示的鎖都實(shí)現(xiàn)了java.util.concurrent.Lock接口，主要有兩個(gè)實(shí)現(xiàn)類：

- ReentrantLock - 比synchronized稍微靈活一些的重入鎖

- ReentrantReadWriteLock - 在讀操作很多寫操作很少時(shí)性能更好的一種重入鎖

對(duì)于如何使用顯示鎖，可以參考我的Java面試系列文章《Java面試題集51-70》中第60題的代碼。只有一點(diǎn)需要提醒，解鎖的方法unlock的調(diào)用最好能夠在finally塊中，因?yàn)檫@里是釋放外部資源最好的地方，當(dāng)然也是釋放鎖的最佳位置，因?yàn)椴还苷．惓？赡芏家尫诺翩i來給其他線程以運(yùn)行的機(jī)會(huì)。

CountDownLatch

CountDownLatch是一種簡單的同步模式，它讓一個(gè)線程可以等待一個(gè)或多個(gè)線程完成它們的工作從而避免對(duì)臨界資源并發(fā)訪問所引發(fā)的各種問題。下面借用別人的一段代碼（我對(duì)它做了一些重構(gòu)）來演示CountDownLatch是如何工作的。

import java.util.concurrent.CountDownLatch;/**

* 工人類

* @author 駱昊

*

*/classWorker{privateString name;// 名字privatelongworkDuration;// 工作持續(xù)時(shí)間/**

* 構(gòu)造器

*/publicWorker(String name,longworkDuration){this.name = name;this.workDuration = workDuration;}/**

* 完成工作

*/publicvoiddoWork(){System.out.println(name +" begins to work...");try{Thread.sleep(workDuration);// 用休眠模擬工作執(zhí)行的時(shí)間}catch(InterruptedException ex) {ex.printStackTrace();}System.out.println(name +" has finished the job...");}}/**

* 測試線程

* @author 駱昊

*

*/classWorkerTestThreadimplementsRunnable{privateWorker worker;privateCountDownLatch cdLatch;publicWorkerTestThread(Worker worker, CountDownLatch cdLatch){this.worker = worker;this.cdLatch = cdLatch;} @Overridepublicvoidrun(){worker.doWork();// 讓工人開始工作cdLatch.countDown();// 工作完成后倒計(jì)時(shí)次數(shù)減1}}classCountDownLatchTest{privatestaticfinalintMAX_WORK_DURATION =5000;// 最大工作時(shí)間privatestaticfinalintMIN_WORK_DURATION =1000;// 最小工作時(shí)間// 產(chǎn)生隨機(jī)的工作時(shí)間privatestaticlonggetRandomWorkDuration(longmin,longmax){return(long) (Math.random() * (max - min) + min);}publicstaticvoidmain(String[] args){CountDownLatch latch =newCountDownLatch(2);// 創(chuàng)建倒計(jì)時(shí)閂并指定倒計(jì)時(shí)次數(shù)為2Worker w1 =newWorker("駱昊", getRandomWorkDuration(MIN_WORK_DURATION, MAX_WORK_DURATION));Worker w2 =newWorker("王大錘", getRandomWorkDuration(MIN_WORK_DURATION, MAX_WORK_DURATION));newThread(newWorkerTestThread(w1, latch)).start();newThread(newWorkerTestThread(w2, latch)).start();try{latch.await();// 等待倒計(jì)時(shí)閂減到0System.out.println("All jobs have been finished!");}catch(InterruptedException e) {e.printStackTrace();}}}

ConcurrentHashMap

ConcurrentHashMap是HashMap在并發(fā)環(huán)境下的版本，大家可能要問，既然已經(jīng)可以通過Collections.synchronizedMap獲得線程安全的映射型容器，為什么還需要ConcurrentHashMap呢？因?yàn)橥ㄟ^Collections工具類獲得的線程安全的HashMap會(huì)在讀寫數(shù)據(jù)時(shí)對(duì)整個(gè)容器對(duì)象上鎖，這樣其他使用該容器的線程無論如何也無法再獲得該對(duì)象的鎖，也就意味著要一直等待前一個(gè)獲得鎖的線程離開同步代碼塊之后才有機(jī)會(huì)執(zhí)行。實(shí)際上，HashMap是通過哈希函數(shù)來確定存放鍵值對(duì)的桶（桶是為了解決哈希沖突而引入的），修改HashMap時(shí)并不需要將整個(gè)容器鎖住，只需要鎖住即將修改的“桶”就可以了。HashMap的數(shù)據(jù)結(jié)構(gòu)如下圖所示。

圖1. HashMap的數(shù)據(jù)結(jié)構(gòu)

此外，ConcurrentHashMap還提供了原子操作的方法，如下所示：

- putIfAbsent：如果還沒有對(duì)應(yīng)的鍵值對(duì)映射，就將其添加到HashMap中

- remove：如果鍵存在而且值與當(dāng)前狀態(tài)相等（equals比較結(jié)果為true），則用原子方式移除該鍵值對(duì)映射

- replace：替換掉映射中元素的原子操作

CopyOnWriteArrayList

CopyOnWriteArrayList是ArrayList在并發(fā)環(huán)境下的替代品。

CopyOnWriteArrayList通過增加寫時(shí)復(fù)制語義來避免并發(fā)訪問引起的問題，也就是說任何修改操作都會(huì)在底層創(chuàng)建一個(gè)列表的副本，也就意味著之前已有的迭代器不會(huì)碰到意料之外的修改。這種方式對(duì)于不要嚴(yán)格讀寫同步的場景非常有用，因?yàn)樗峁┝烁玫男阅?。記住，要盡量減少鎖的使用，因?yàn)槟莿?shì)必帶來性能的下降（對(duì)數(shù)據(jù)庫中數(shù)據(jù)的并發(fā)訪問不也是如此嗎？如果可以的話就應(yīng)該放棄悲觀鎖而使用樂觀鎖），CopyOnWriteArrayList很明顯也是通過犧牲空間獲得了時(shí)間（在計(jì)算機(jī)的世界里，時(shí)間和空間通常是不可調(diào)和的矛盾，可以犧牲空間來提升效率獲得時(shí)間，當(dāng)然也可以通過犧牲時(shí)間來減少對(duì)空間的使用）。

圖1. CopyOnWriteArrayList原理示意圖

可以通過下面兩段代碼的運(yùn)行狀況來驗(yàn)證一下CopyOnWriteArrayList是不是線程安全的容器。

importjava.util.ArrayList;importjava.util.List;importjava.util.concurrent.ExecutorService;importjava.util.concurrent.Executors;classAddThreadimplementsRunnable{privateListlist;publicAddThread(Listlist){this.list=list;}@Overridepublicvoidrun(){for(inti =0; i <10000; ++i) {list.add(Math.random());}}}publicclassTest05{privatestaticfinalintTHREAD_POOL_SIZE =2;publicstaticvoidmain(String[] args){Listlist=newArrayList<>();ExecutorService es = Executors.newFixedThreadPool(THREAD_POOL_SIZE);es.execute(newAddThread(list));es.execute(newAddThread(list));es.shutdown();}}

上面的代碼會(huì)在運(yùn)行時(shí)產(chǎn)生ArrayIndexOutOfBoundsException，試一試將上面代碼25行的ArrayList換成CopyOnWriteArrayList再重新運(yùn)行。

Listlist=newCopyOnWriteArrayList<>();

Queue

隊(duì)列是一個(gè)無處不在的美妙概念，它提供了一種簡單又可靠的方式將資源分發(fā)給處理單元（也可以說是將工作單元分配給待處理的資源，這取決于你看待問題的方式）。實(shí)現(xiàn)中的并發(fā)編程模型很多都依賴隊(duì)列來實(shí)現(xiàn)，因?yàn)樗梢栽诰€程之間傳遞工作單元。

Java 5中的BlockingQueue就是一個(gè)在并發(fā)環(huán)境下非常好用的工具，在調(diào)用put方法向隊(duì)列中插入元素時(shí)，如果隊(duì)列已滿，它會(huì)讓插入元素的線程等待隊(duì)列騰出空間；在調(diào)用take方法從隊(duì)列中取元素時(shí)，如果隊(duì)列為空，取出元素的線程就會(huì)阻塞。

圖3. BlockingQueue示意圖

可以用BlockingQueue來實(shí)現(xiàn)生產(chǎn)者-消費(fèi)者并發(fā)模型（下一節(jié)中有介紹），當(dāng)然在Java 5以前也可以通過wait和notify來實(shí)現(xiàn)線程調(diào)度，比較一下兩種代碼就知道基于已有的并發(fā)工具類來重構(gòu)并發(fā)代碼到底好在哪里了。

基于wait和notify的實(shí)現(xiàn)

importjava.util.ArrayList;importjava.util.List;importjava.util.UUID;importjava.util.concurrent.ExecutorService;importjava.util.concurrent.Executors;/**

* 公共常量

* @author 駱昊

*

*/class Constants {publicstaticfinalintMAX_BUFFER_SIZE =10;publicstaticfinalintNUM_OF_PRODUCER =2;publicstaticfinalintNUM_OF_CONSUMER =3;}/**

* 工作任務(wù)

* @author 駱昊

*

*/classTask{privateStringid;// 任務(wù)的編號(hào)publicTask() {id = UUID.randomUUID().toString();}@OverridepublicStringtoString() {return"Task["+ id +"]";}}/**

* 消費(fèi)者

* @author 駱昊

*

*/class Consumer implements Runnable {privateListbuffer;publicConsumer(Listbuffer) {this.buffer=buffer;}@Overridepublicvoidrun() {while(true) { synchronized(buffer) {while(buffer.isEmpty()) {try{buffer.wait();}catch(InterruptedException e) {e.printStackTrace();}}Tasktask =buffer.remove(0);buffer.notifyAll();System.out.println("Consumer["+ Thread.currentThread().getName() +"] got "+ task);}}}}/**

* 生產(chǎn)者

* @author 駱昊

*

*/class Producer implements Runnable {privateListbuffer;publicProducer(Listbuffer) {this.buffer=buffer;}@Overridepublicvoidrun() {while(true) { synchronized (buffer) {while(buffer.size() >= Constants.MAX_BUFFER_SIZE) {try{buffer.wait();}catch(InterruptedException e) {e.printStackTrace();}}Tasktask =newTask();buffer.add(task);buffer.notifyAll();System.out.println("Producer["+ Thread.currentThread().getName() +"] put "+ task);}}}}publicclass Test06 {publicstaticvoidmain(String[] args) {Listbuffer=newArrayList<>(Constants.MAX_BUFFER_SIZE);ExecutorService es = Executors.newFixedThreadPool(Constants.NUM_OF_CONSUMER + Constants.NUM_OF_PRODUCER);for(inti =1; i <= Constants.NUM_OF_PRODUCER; ++i) {es.execute(newProducer(buffer));}for(inti =1; i <= Constants.NUM_OF_CONSUMER; ++i) {es.execute(newConsumer(buffer));}}}

基于BlockingQueue的實(shí)現(xiàn)

importjava.util.UUID;importjava.util.concurrent.BlockingQueue;importjava.util.concurrent.ExecutorService;importjava.util.concurrent.Executors;importjava.util.concurrent.LinkedBlockingQueue;/**

* 公共常量

* @author 駱昊

*

*/class Constants {publicstaticfinalintMAX_BUFFER_SIZE =10;publicstaticfinalintNUM_OF_PRODUCER =2;publicstaticfinalintNUM_OF_CONSUMER =3;}/**

* 工作任務(wù)

* @author 駱昊

*

*/classTask{privateStringid;// 任務(wù)的編號(hào)publicTask() {id = UUID.randomUUID().toString();} @OverridepublicStringtoString() {return"Task["+ id +"]";}}/**

* 消費(fèi)者

* @author 駱昊

*

*/class Consumer implements Runnable {privateBlockingQueuebuffer;publicConsumer(BlockingQueuebuffer) {this.buffer=buffer;} @Overridepublicvoidrun() {while(true) {try{Tasktask =buffer.take();System.out.println("Consumer["+ Thread.currentThread().getName() +"] got "+ task);}catch(InterruptedException e) {e.printStackTrace();}}}}/**

* 生產(chǎn)者

* @author 駱昊

*

*/class Producer implements Runnable {privateBlockingQueuebuffer;publicProducer(BlockingQueuebuffer) {this.buffer=buffer;} @Overridepublicvoidrun() {while(true) {try{Tasktask =newTask();buffer.put(task);System.out.println("Producer["+ Thread.currentThread().getName() +"] put "+ task);}catch(InterruptedException e) {e.printStackTrace();}}}}publicclass Test07 {publicstaticvoidmain(String[] args) {BlockingQueuebuffer=newLinkedBlockingQueue<>(Constants.MAX_BUFFER_SIZE);ExecutorService es = Executors.newFixedThreadPool(Constants.NUM_OF_CONSUMER + Constants.NUM_OF_PRODUCER);for(inti =1; i <= Constants.NUM_OF_PRODUCER; ++i) {es.execute(newProducer(buffer));}for(inti =1; i <= Constants.NUM_OF_CONSUMER; ++i) {es.execute(newConsumer(buffer));}}}

使用BlockingQueue后代碼優(yōu)雅了很多。

并發(fā)模型

在繼續(xù)下面的探討之前，我們還是重溫一下幾個(gè)概念：

概念解釋

臨界資源并發(fā)環(huán)境中有著固定數(shù)量的資源

互斥對(duì)資源的訪問是排他式的

饑餓一個(gè)或一組線程長時(shí)間或永遠(yuǎn)無法取得進(jìn)展

死鎖兩個(gè)或多個(gè)線程相互等待對(duì)方結(jié)束

活鎖想要執(zhí)行的線程總是發(fā)現(xiàn)其他的線程正在執(zhí)行以至于長時(shí)間或永遠(yuǎn)無法執(zhí)行

重溫了這幾個(gè)概念后，我們可以探討一下下面的幾種并發(fā)模型。

生產(chǎn)者-消費(fèi)者

一個(gè)或多個(gè)生產(chǎn)者創(chuàng)建某些工作并將其置于緩沖區(qū)或隊(duì)列中，一個(gè)或多個(gè)消費(fèi)者會(huì)從隊(duì)列中獲得這些工作并完成之。這里的緩沖區(qū)或隊(duì)列是臨界資源。當(dāng)緩沖區(qū)或隊(duì)列放滿的時(shí)候，生產(chǎn)這會(huì)被阻塞；而緩沖區(qū)或隊(duì)列為空的時(shí)候，消費(fèi)者會(huì)被阻塞。生產(chǎn)者和消費(fèi)者的調(diào)度是通過二者相互交換信號(hào)完成的。

讀者-寫者

當(dāng)存在一個(gè)主要為讀者提供信息的共享資源，它偶爾會(huì)被寫者更新，但是需要考慮系統(tǒng)的吞吐量，又要防止饑餓和陳舊資源得不到更新的問題。在這種并發(fā)模型中，如何平衡讀者和寫者是最困難的，當(dāng)然這個(gè)問題至今還是一個(gè)被熱議的問題，恐怕必須根據(jù)具體的場景來提供合適的解決方案而沒有那種放之四海而皆準(zhǔn)的方法（不像我在國內(nèi)的科研文獻(xiàn)中看到的那樣）。

哲學(xué)家進(jìn)餐

1965年，荷蘭計(jì)算機(jī)科學(xué)家圖靈獎(jiǎng)得主Edsger Wybe Dijkstra提出并解決了一個(gè)他稱之為哲學(xué)家進(jìn)餐的同步問題。這個(gè)問題可以簡單地描述如下：五個(gè)哲學(xué)家圍坐在一張圓桌周圍，每個(gè)哲學(xué)家面前都有一盤通心粉。由于通心粉很滑，所以需要兩把叉子才能夾住。相鄰兩個(gè)盤子之間放有一把叉子如下圖所示。哲學(xué)家的生活中有兩種交替活動(dòng)時(shí)段：即吃飯和思考。當(dāng)一個(gè)哲學(xué)家覺得餓了時(shí)，他就試圖分兩次去取其左邊和右邊的叉子，每次拿一把，但不分次序。如果成功地得到了兩把叉子，就開始吃飯，吃完后放下叉子繼續(xù)思考。

把上面問題中的哲學(xué)家換成線程，把叉子換成競爭的臨界資源，上面的問題就是線程競爭資源的問題。如果沒有經(jīng)過精心的設(shè)計(jì)，系統(tǒng)就會(huì)出現(xiàn)死鎖、活鎖、吞吐量下降等問題。

圖4.哲學(xué)家進(jìn)餐模型

下面是用信號(hào)量原語來解決哲學(xué)家進(jìn)餐問題的代碼，使用了Java 5并發(fā)工具包中的Semaphore類（代碼不夠漂亮但是已經(jīng)足以說明問題了）。

//import java.util.concurrent.ExecutorService;//import java.util.concurrent.Executors;import java.util.concurrent.Semaphore;/*** 存放線程共享信號(hào)量的上下問* @author 駱昊**/classAppContext{publicstaticfinalintNUM_OF_FORKS =5;// 叉子數(shù)量(資源)publicstaticfinalintNUM_OF_PHILO =5;// 哲學(xué)家數(shù)量(線程)publicstaticSemaphore[] forks;// 叉子的信號(hào)量publicstaticSemaphore counter;// 哲學(xué)家的信號(hào)量static{forks =newSemaphore[NUM_OF_FORKS];for(inti =0, len = forks.length; i < len; ++i) {forks[i] =newSemaphore(1);// 每個(gè)叉子的信號(hào)量為1}counter =newSemaphore(NUM_OF_PHILO -1);// 如果有N個(gè)哲學(xué)家，最多只允許N-1人同時(shí)取叉子}/**

* 取得叉子

* @param index 第幾個(gè)哲學(xué)家

* @param leftFirst 是否先取得左邊的叉子

* @throws InterruptedException

*/publicstaticvoidputOnFork(intindex,booleanleftFirst) throws InterruptedException {if(leftFirst) {forks[index].acquire();forks[(index+1) % NUM_OF_PHILO].acquire();}else{forks[(index+1) % NUM_OF_PHILO].acquire();forks[index].acquire();}}/**

* 放回叉子

* @param index 第幾個(gè)哲學(xué)家

* @param leftFirst 是否先放回左邊的叉子

* @throws InterruptedException

*/publicstaticvoidputDownFork(intindex,booleanleftFirst) throws InterruptedException {if(leftFirst) {forks[index].release();forks[(index+1) % NUM_OF_PHILO].release();}else{forks[(index+1) % NUM_OF_PHILO].release();forks[index].release();}}}/**

* 哲學(xué)家

* @author 駱昊

*

*/classPhilosopherimplementsRunnable{privateintindex;// 編號(hào)privateString name;// 名字publicPhilosopher(intindex, String name) {this.index=index;this.name = name;} @Overridepublicvoidrun() {while(true) {try{AppContext.counter.acquire();booleanleftFirst =index%2==0;AppContext.putOnFork(index, leftFirst);System.out.println(name +"正在吃意大利面（通心粉）...");// 取到兩個(gè)叉子就可以進(jìn)食AppContext.putDownFork(index, leftFirst);AppContext.counter.release();}catch(InterruptedException e) {e.printStackTrace();}}}}publicclassTest04{publicstaticvoidmain(String[] args) {String[] names = {"駱昊","王大錘","張三豐","楊過","李莫愁"};// 5位哲學(xué)家的名字// ExecutorService es = Executors.newFixedThreadPool(AppContext.NUM_OF_PHILO); // 創(chuàng)建固定大小的線程池// for(int i = 0, len = names.length; i < len; ++i) {// es.execute(new Philosopher(i, names[i])); // 啟動(dòng)線程// }// es.shutdown();for(inti =0, len = names.length; i < len; ++i) {newThread(newPhilosopher(i, names[i])).start();}}}

現(xiàn)實(shí)中的并發(fā)問題基本上都是這三種模型或者是這三種模型的變體。

測試并發(fā)代碼

對(duì)并發(fā)代碼的測試也是非常棘手的事情，棘手到無需說明大家也很清楚的程度，所以這里我們只是探討一下如何解決這個(gè)棘手的問題。我們建議大家編寫一些能夠發(fā)現(xiàn)問題的測試并經(jīng)常性的在不同的配置和不同的負(fù)載下運(yùn)行這些測試。不要忽略掉任何一次失敗的測試，線程代碼中的缺陷可能在上萬次測試中僅僅出現(xiàn)一次。具體來說有這么幾個(gè)注意事項(xiàng)：

- 不要將系統(tǒng)的失效歸結(jié)于偶發(fā)事件，就像拉不出屎的時(shí)候不能怪地球沒有引力。

- 先讓非并發(fā)代碼工作起來，不要試圖同時(shí)找到并發(fā)和非并發(fā)代碼中的缺陷。

- 編寫可以在不同配置環(huán)境下運(yùn)行的線程代碼。

- 編寫容易調(diào)整的線程代碼，這樣可以調(diào)整線程使性能達(dá)到最優(yōu)。

- 讓線程的數(shù)量多于CPU或CPU核心的數(shù)量，這樣CPU調(diào)度切換過程中潛在的問題才會(huì)暴露出來。

- 讓并發(fā)代碼在不同的平臺(tái)上運(yùn)行。

- 通過自動(dòng)化或者硬編碼的方式向并發(fā)代碼中加入一些輔助測試的代碼。

推薦體系

我向大家推薦一下我認(rèn)為比較全面且最系統(tǒng)化的學(xué)習(xí)體系（分解后的，完整的加群可以獲?。?/p>

一、源碼分析

二、分布式架構(gòu)

三、微服務(wù)

四、性能優(yōu)化

五、Java工程化

以上就是我推薦給大家的最具有系統(tǒng)化的學(xué)習(xí)體系，若果你想學(xué)習(xí)以上的知識(shí)內(nèi)容，你可以加這個(gè)群獲?。航涣鲗W(xué)習(xí)群：650385180里面會(huì)分享一些資深架構(gòu)師錄制的視頻錄像：有Spring，MyBatis，Netty源碼分析，高并發(fā)、高性能、分布式、微服務(wù)架構(gòu)的原理，JVM性能優(yōu)化這些成為架構(gòu)師必備的知識(shí)體系。還能領(lǐng)取免費(fèi)的學(xué)習(xí)資源，目前受益良多，面試題也在群里面。

Java 7的并發(fā)編程

Java 7中引入了TransferQueue，它比BlockingQueue多了一個(gè)叫transfer的方法，如果接收線程處于等待狀態(tài)，該操作可以馬上將任務(wù)交給它，否則就會(huì)阻塞直至取走該任務(wù)的線程出現(xiàn)。可以用TransferQueue代替BlockingQueue，因?yàn)樗梢垣@得更好的性能。

剛才忘記了一件事情，Java 5中還引入了Callable接口、Future接口和FutureTask接口，通過他們也可以構(gòu)建并發(fā)應(yīng)用程序，代碼如下所示。

importjava.util.ArrayList;importjava.util.List;importjava.util.concurrent.Callable;importjava.util.concurrent.ExecutorService;importjava.util.concurrent.Executors;importjava.util.concurrent.Future;publicclassTest07{privatestaticfinalintPOOL_SIZE =10;staticclassCalcThreadimplementsCallable{privateList dataList =newArrayList<>();publicCalcThread(){for(inti =0; i <10000; ++i) {dataList.add(Math.random());}}@OverridepublicDoublecall()throwsException{doubletotal =0;for(Double d : dataList) {total += d;}returntotal / dataList.size();}}publicstaticvoidmain(String[] args){List> fList =newArrayList<>();ExecutorService es = Executors.newFixedThreadPool(POOL_SIZE);for(inti =0; i < POOL_SIZE; ++i) {fList.add(es.submit(newCalcThread()));}for(Future f : fList) {try{System.out.println(f.get());}catch(Exception e) {e.printStackTrace();}}es.shutdown();}}

Callable接口也是一個(gè)單方法接口，顯然這是一個(gè)回調(diào)方法，類似于函數(shù)式編程中的回調(diào)函數(shù)，在Java 8 以前，Java中還不能使用Lambda表達(dá)式來簡化這種函數(shù)式編程。和Runnable接口不同的是Callable接口的回調(diào)方法call方法會(huì)返回一個(gè)對(duì)象，這個(gè)對(duì)象可以用將來時(shí)的方式在線程執(zhí)行結(jié)束的時(shí)候獲得信息。上面代碼中的call方法就是將計(jì)算出的10000個(gè)0到1之間的隨機(jī)小數(shù)的平均值返回，我們通過一個(gè)Future接口的對(duì)象得到了這個(gè)返回值。目前最新的Java版本中，Callable接口和Runnable接口都被打上了@FunctionalInterface的注解，也就是說它可以用函數(shù)式編程的方式（Lambda表達(dá)式）創(chuàng)建接口對(duì)象。

下面是Future接口的主要方法：

- get()：獲取結(jié)果。如果結(jié)果還沒有準(zhǔn)備好，get方法會(huì)阻塞直到取得結(jié)果；當(dāng)然也可以通過參數(shù)設(shè)置阻塞超時(shí)時(shí)間

- cancel()：在運(yùn)算結(jié)束前取消

- isDone()：可以用來判斷運(yùn)算是否結(jié)束

Java 7中還提供了分支/合并（fork/join）框架，它可以實(shí)現(xiàn)線程池中任務(wù)的自動(dòng)調(diào)度，并且這種調(diào)度對(duì)用戶來說是透明的。為了達(dá)到這種效果，必須按照用戶指定的方式對(duì)任務(wù)進(jìn)行分解，然后再將分解出的小型任務(wù)的執(zhí)行結(jié)果合并成原來任務(wù)的執(zhí)行結(jié)果。這顯然是運(yùn)用了分治法（divide-and-conquer）的思想。下面的代碼使用了分支/合并框架來計(jì)算1到10000的和，當(dāng)然對(duì)于如此簡單的任務(wù)根本不需要分支/合并框架，因?yàn)榉种Ш秃喜⒈旧硪矔?huì)帶來一定的開銷，但是這里我們只是探索一下在代碼中如何使用分支/合并框架，讓我們的代碼能夠充分利用現(xiàn)代多核CPU的強(qiáng)大運(yùn)算能力。

importjava.util.concurrent.ForkJoinPool;importjava.util.concurrent.Future;importjava.util.concurrent.RecursiveTask;classCalculatorextendsRecursiveTask{privatestaticfinallongserialVersionUID =7333472779649130114L;privatestaticfinalintTHRESHOLD =10;privateintstart;privateintend;publicCalculator(intstart,intend){this.start = start;this.end = end;}@OverridepublicIntegercompute(){intsum =0;if((end - start) < THRESHOLD) {// 當(dāng)問題分解到可求解程度時(shí)直接計(jì)算結(jié)果for(inti = start; i <= end; i++) {sum += i;}}else{intmiddle = (start + end) >>>1;// 將任務(wù)一分為二Calculator left =newCalculator(start, middle);Calculator right =newCalculator(middle +1, end);left.fork();right.fork();// 注意：由于此處是遞歸式的任務(wù)分解，也就意味著接下來會(huì)二分為四，四分為八...sum = left.join() + right.join();// 合并兩個(gè)子任務(wù)的結(jié)果}returnsum;}}publicclassTest08{publicstaticvoidmain(String[] args)throwsException{ForkJoinPool forkJoinPool =newForkJoinPool();Future result = forkJoinPool.submit(newCalculator(1,10000));System.out.println(result.get());}}

在已經(jīng)到來的Java 7中，Java中默認(rèn)的數(shù)組排序算法已經(jīng)不再是經(jīng)典的快速排序（雙樞軸快速排序）了，新的排序算法叫TimSort，它是歸并排序和插入排序的混合體，TimSort可以通過分支合并框架充分利用現(xiàn)代處理器的多核特性，從而獲得更好的性能（更短的排序時(shí)間）。

而伴隨著Java10的到來，排序算法會(huì)不會(huì)有一些新的變化呢，新的變化又會(huì)是什么呢，讓我們拭目以待吧。

總結(jié)：

為了適應(yīng)社會(huì)新的需要，開發(fā)人員需能夠自我激勵(lì)，主動(dòng)學(xué)習(xí)新技術(shù)，并在職業(yè)生涯中給自己扣上很多帽子。 繼而不斷挑戰(zhàn)自我，然后更好地解決問題，這就是編程的本質(zhì)。 知識(shí)很重要，在某些復(fù)雜問題的情況下更是如此。在變化如此之快的IT技術(shù)領(lǐng)域中，知識(shí)的獲取在任何時(shí)候比我們已會(huì)的技能更為重要。

關(guān)于Java并發(fā)編程的一些總結(jié)與思考已經(jīng)寫完了，受限于我的視野，所以可能寫的不是很全面，大家要是有不同意見的，可以分享出來，一起交流，要是想深入了解并發(fā)編程的，也可以加上面的群，希望可以幫助在這個(gè)行業(yè)發(fā)展的朋友和童鞋們，在論壇博客等地方少花些時(shí)間找資料，把有限的時(shí)間，真正花在學(xué)習(xí)上。