聊聊一線互聯網公司中併發編程的那些事（內含源碼及面試題）

時間 2019-11-20

原文原文鏈接

導讀：以前寫了一系列關於併發編程的文章，也對今年的一些大型互聯網公司的併發編程面試題作了一個大體的彙總，今天，就來寫一下關於Java併發編程的總結與思考。java

編寫優質的併發代碼是一件難度極高的事情。Java語言從初版本開始內置了對多線程的支持，這一點在當年是很是了不得的，可是當咱們對併發編程有了更深入的認識和更多的實踐後，實現併發編程就有了更多的方案和更好的選擇。本文是對併發編程的一點總結和思考，同時也分享了Java 5之後的版本中如何編寫併發代碼的一點點經驗。程序員

爲何須要併發

併發實際上是一種解耦合的策略，它幫助咱們把作什麼（目標）和何時作（時機）分開。這樣作能夠明顯改進應用程序的吞吐量（得到更多的CPU調度時間）和結構（程序有多個部分在協同工做）。作過Java Web開發的人都知道，Java Web中的Servlet程序在Servlet容器的支持下采用單實例多線程的工做模式，Servlet容器幫助你處理了併發請求的問題。面試

誤解和正解

最多見的對併發編程的誤解有如下這些：算法

A. 併發總能改進性能。（真相：併發在CPU有不少空閒時間時能明顯改進程序的性能，但當線程數量較多的時候，線程間頻繁的調度切換反而會讓系統的性能降低）數據庫

B. 編寫併發程序無需修改原有的設計。（真相：目的與時機的解耦每每會對系統結構產生巨大的影響）編程

C. 在使用Web或EJB容器時不用關注併發問題。數組

（真相：只有瞭解了容器在作什麼，才能更好的使用容器）緩存

下面的這些說法纔是對併發編程比較客觀的認識：安全

A. 編寫併發程序會在代碼上增長額外的開銷。性能優化

B. 正確的併發是很是複雜的，即便對於很簡單的問題。

C. 併發中的缺陷由於不易重現也不容易被發現。

D. 併發每每須要對設計策略從根本上進行修改。

併發編程的原則和技巧

1. 單一職責原則：分離併發相關代碼和其餘代碼（併發相關代碼有本身的開發、修改和調優生命週期）。

2. 限制數據做用域：兩個線程修改共享對象的同一字段時可能會相互干擾，致使不可預期的行爲，解決方案之一是構造臨界區，可是必須限制臨界區的數量。

3. 使用數據副本：數據副本是避免共享數據的好方法，複製出來的對象只是以只讀的方式對待。Java 5的java.util.concurrent包中增長一個名爲CopyOnWriteArrayList的類，它是List接口的子類型，因此你能夠認爲它是ArrayList的線程安全的版本，它使用了寫時複製的方式建立數據副本進行操做來避免對共享數據併發訪問而引起的問題。

4. 線程應儘量獨立：讓線程存在於本身的世界中，不與其餘線程共享數據。有過Java Web開發經驗的人都知道，Servlet就是以單實例多線程的方式工做，和每一個請求相關的數據都是經過Servlet子類的service方法（或者是doGet或doPost方法）的參數傳入的。只要Servlet中的代碼只使用局部變量，Servlet就不會致使同步問題。Spring MVC的控制器也是這麼作的，從請求中得到的對象都是以方法的參數傳入而不是做爲類的成員，很明顯Struts 2的作法就正好相反，所以Struts 2中做爲控制器的Action類都是每一個請求對應一個實例。

Java 5之前的併發編程

Java的線程模型創建在搶佔式線程調度的基礎上，也就是說：

1. 全部線程能夠很容易的共享同一進程中的對象。

2. 可以引用這些對象的任何線程均可以修改這些對象。

3. 爲了保護數據，對象能夠被鎖住。

Java基於線程和鎖的併發過於底層，並且使用鎖不少時候都是很萬惡的，由於它至關於讓全部的併發都變成了排隊等待。

在Java 5之前，能夠用synchronized關鍵字來實現鎖的功能，它能夠用在代碼塊和方法上，表示在執行整個代碼塊或方法以前線程必須取得合適的鎖。對於類的非靜態方法（成員方法）而言，這意味這要取得對象實例的鎖，對於類的靜態方法（類方法）而言，要取得類的Class對象的鎖，對於同步代碼塊，程序員能夠指定要取得的是那個對象的鎖。

無論是同步代碼塊仍是同步方法，每次只有一個線程能夠進入，若是其餘線程試圖進入（無論是同一同步塊仍是不一樣的同步塊），JVM會將它們掛起（放入到等鎖池中）。這種結構在併發理論中稱爲臨界區（critical section）。這裏咱們能夠對Java中用synchronized實現同步和鎖的功能作一個總結：

- 只能鎖定對象，不能鎖定基本數據類型。

- 被鎖定的對象數組中的單個對象不會被鎖定。

- 同步方法能夠視爲包含整個方法的synchronized(this) { ... }代碼塊。

- 靜態同步方法會鎖定它的Class對象。

- 內部類的同步是獨立於外部類的。

- synchronized修飾符並非方法簽名的組成部分，因此不能出如今接口的方法聲明中。

- 非同步的方法不關心鎖的狀態，它們在同步方法運行時仍然能夠得以運行。

- synchronized實現的鎖是可重入的鎖。

在JVM內部，爲了提升效率，同時運行的每一個線程都會有它正在處理的數據的緩存副本，當咱們使用synchronzied進行同步的時候，真正被同步的是在不一樣線程中表示被鎖定對象的內存塊（副本數據會保持和主內存的同步，如今知道爲何要用同步這個詞彙了吧），簡單的說就是在同步塊或同步方法執行完後，對被鎖定的對象作的任何修改要在釋放鎖以前寫回到主內存中；在進入同步塊獲得鎖以後，被鎖定對象的數據是從主內存中讀出來的，持有鎖的線程的數據副本必定和主內存中的數據視圖是同步的。

在Java最初的版本中，就有一個叫volatile的關鍵字，它是一種簡單的同步的處理機制，由於被volatile修飾的變量遵循如下規則：

- 變量的值在使用以前總會從主內存中再讀取出來。

- 對變量值的修改總會在完成以後寫回到主內存中。

使用volatile關鍵字能夠在多線程環境下預防編譯器不正確的優化假設（編譯器可能會將在一個線程中值不會發生改變的變量優化成常量），但只有修改時不依賴當前狀態（讀取時的值）的變量才應該聲明爲volatile變量。

不變模式也是併發編程時能夠考慮的一種設計。讓對象的狀態是不變的，若是但願修改對象的狀態，就會建立對象的副本並將改變寫入副本而不改變原來的對象，這樣就不會出現狀態不一致的狀況，所以不變對象是線程安全的。Java中咱們使用頻率極高的String類就採用了這樣的設計。若是對不變模式不熟悉，能夠閱讀閻宏博士的《Java與模式》一書的第34章。說到這裏你可能也體會到final關鍵字的重要意義了。

Java 5的併發編程

無論從此的Java向着何種方向發展或者滅忙，Java 5絕對是Java發展史中一個極其重要的版本，咱們必需要感謝Doug Lea在Java 5中提供了他里程碑式的傑做java.util.concurrent包，它的出現讓Java的併發編程有了更多的選擇和更好的工做方式。Doug Lea的傑做主要包括如下內容：

- 更好的線程安全的容器

- 線程池和相關的工具類

- 可選的非阻塞解決方案

- 顯示的鎖和信號量機制

原子類

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

/**
ID序列生成器
*/public class IdGenerator { private final AtomicLong sequenceNumber = new AtomicLong(0); public long next() { return sequenceNumber.getAndIncrement();
}
}

顯示鎖

基於synchronized關鍵字的鎖機制有如下問題：

- 鎖只有一種類型，並且對全部同步操做都是同樣的做用

- 鎖只能在代碼塊或方法開始的地方得到，在結束的地方釋放

- 線程要麼獲得鎖，要麼阻塞，沒有其餘的可能性

Java 5對鎖機制進行了重構，提供了顯示的鎖，這樣能夠在如下幾個方面提高鎖機制：

- 能夠添加不一樣類型的鎖，例如讀取鎖和寫入鎖。

- 能夠在一個方法中加鎖，在另外一個方法中解鎖。

- 可使用tryLock方式嘗試得到鎖，若是得不到鎖能夠等待、回退或者乾點別的事情，固然也能夠在超時以後放棄操做。

顯示的鎖都實現了java.util.concurrent.Lock接口，主要有兩個實現類：

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

- ReentrantReadWriteLock - 在讀操做不少寫操做不多時性能更好的一種重入鎖

對於如何使用顯示鎖，能夠參考個人Java面試系列文章《Java面試題集51-70》中第60題的代碼。只有一點須要提醒，解鎖的方法unlock的調用最好可以在finally塊中，由於這裏是釋放外部資源最好的地方，固然也是釋放鎖的最佳位置，由於無論正常異常可能都要釋放掉鎖來給其餘線程以運行的機會。

CountDownLatch

CountDownLatch是一種簡單的同步模式，它讓一個線程能夠等待一個或多個線程完成它們的工做從而避免對臨界資源併發訪問所引起的各類問題。下面借用別人的一段代碼（我對它作了一些重構）來演示CountDownLatch是如何工做的。

import java.util.concurrent.CountDownLatch;/**
* 工人類
* @author 駱昊
*
*/class Worker { private String name; // 名字
private long workDuration; // 工做持續時間
/**
 * 構造器
 */
public Worker(String name, long workDuration) { this.name = name; this.workDuration = workDuration;
} /**
 * 完成工做
 */
public void doWork() {
System.out.println(name + " begins to work..."); try {
Thread.sleep(workDuration); // 用休眠模擬工做執行的時間
} catch(InterruptedException ex) {
ex.printStackTrace();
}
System.out.println(name + " has finished the job...");
}
}/**
* 測試線程
* @author 駱昊
*
*/class WorkerTestThread implements Runnable { private Worker worker; private CountDownLatch cdLatch; public WorkerTestThread(Worker worker, CountDownLatch cdLatch) { this.worker = worker; this.cdLatch = cdLatch;
} @Override
public void run() {
worker.doWork(); // 讓工人開始工做
cdLatch.countDown(); // 工做完成後倒計時次數減1
}
}class CountDownLatchTest { private static final int MAX_WORK_DURATION = 5000; // 最大工做時間
private static final int MIN_WORK_DURATION = 1000; // 最小工做時間
// 產生隨機的工做時間
private static long getRandomWorkDuration(long min, long max) { return (long) (Math.random() * (max - min) + min);
} public static void main(String[] args) {
CountDownLatch latch = new CountDownLatch(2); // 建立倒計時閂並指定倒計時次數爲2
Worker w1 = new Worker("駱昊", getRandomWorkDuration(MIN_WORK_DURATION, MAX_WORK_DURATION));
Worker w2 = new Worker("王大錘", getRandomWorkDuration(MIN_WORK_DURATION, MAX_WORK_DURATION)); new Thread(new WorkerTestThread(w1, latch)).start(); new Thread(new WorkerTestThread(w2, latch)).start(); try {
latch.await(); // 等待倒計時閂減到0
System.out.println("All jobs have been finished!");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

ConcurrentHashMap

ConcurrentHashMap是HashMap在併發環境下的版本，你們可能要問，既然已經能夠經過Collections.synchronizedMap得到線程安全的映射型容器，爲何還須要ConcurrentHashMap呢？由於經過Collections工具類得到的線程安全的HashMap會在讀寫數據時對整個容器對象上鎖，這樣其餘使用該容器的線程不管如何也沒法再得到該對象的鎖，也就意味着要一直等待前一個得到鎖的線程離開同步代碼塊以後纔有機會執行。實際上，HashMap是經過哈希函數來肯定存放鍵值對的桶（桶是爲了解決哈希衝突而引入的），修改HashMap時並不須要將整個容器鎖住，只須要鎖住即將修改的「桶」就能夠了。HashMap的數據結構以下圖所示。

圖1. HashMap的數據結構

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

- putIfAbsent：若是尚未對應的鍵值對映射，就將其添加到HashMap中

- remove：若是鍵存在並且值與當前狀態相等（equals比較結果爲true），則用原子方式移除該鍵值對映射

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

CopyOnWriteArrayList

CopyOnWriteArrayList是ArrayList在併發環境下的替代品。

CopyOnWriteArrayList經過增長寫時複製語義來避免併發訪問引發的問題，也就是說任何修改操做都會在底層建立一個列表的副本，也就意味着以前已有的迭代器不會碰到意料以外的修改。這種方式對於不要嚴格讀寫同步的場景很是有用，由於它提供了更好的性能。記住，要儘可能減小鎖的使用，由於那勢必帶來性能的降低（對數據庫中數據的併發訪問不也是如此嗎？若是能夠的話就應該放棄悲觀鎖而使用樂觀鎖），CopyOnWriteArrayList很明顯也是經過犧牲空間得到了時間（在計算機的世界裏，時間和空間一般是不可調和的矛盾，能夠犧牲空間來提高效率得到時間，固然也能夠經過犧牲時間來減小對空間的使用）。

圖1. CopyOnWriteArrayList原理示意圖

能夠經過下面兩段代碼的運行情況來驗證一下CopyOnWriteArrayList是否是線程安全的容器。

import java.util.ArrayList;import java.util.List;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;class AddThread implements Runnable { private List<Double> list; public AddThread(List<Double> list) { this.list = list;
}
@Override public void run() { for(int i = 0; i < 10000; ++i) { list.add(Math.random());
}
}
}public class Test05 { private static final int THREAD_POOL_SIZE = 2; public static void main(String[] args) {
List<Double> list = new ArrayList<>();
ExecutorService es = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
es.execute(new AddThread(list));
es.execute(new AddThread(list));
es.shutdown();
}
}

上面的代碼會在運行時產生ArrayIndexOutOfBoundsException，試一試將上面代碼25行的ArrayList換成CopyOnWriteArrayList再從新運行。

List<Double> list = new CopyOnWriteArrayList<>();

Queue

隊列是一個無處不在的美妙概念，它提供了一種簡單又可靠的方式將資源分發給處理單元（也能夠說是將工做單元分配給待處理的資源，這取決於你看待問題的方式）。實現中的併發編程模型不少都依賴隊列來實現，由於它能夠在線程之間傳遞工做單元。

Java 5中的BlockingQueue就是一個在併發環境下很是好用的工具，在調用put方法向隊列中插入元素時，若是隊列已滿，它會讓插入元素的線程等待隊列騰出空間；在調用take方法從隊列中取元素時，若是隊列爲空，取出元素的線程就會阻塞。

圖3. BlockingQueue示意圖

能夠用BlockingQueue來實現生產者-消費者併發模型（下一節中有介紹），固然在Java 5之前也能夠經過wait和notify來實現線程調度，比較一下兩種代碼就知道基於已有的併發工具類來重構併發代碼到底好在哪裏了。

基於wait和notify的實現

import java.util.ArrayList;import java.util.List;import java.util.UUID;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;/**
* 公共常量
* @author 駱昊
*
*/class Constants { public static final int MAX_BUFFER_SIZE = 10; public static final int NUM_OF_PRODUCER = 2; public static final int NUM_OF_CONSUMER = 3;
}/**
* 工做任務
* @author 駱昊
*
*/class Task { private String id; // 任務的編號
public Task() {
id = UUID.randomUUID().toString();
}
@Override public String toString() { return "Task[" + id + "]";
}
}/**
* 消費者
* @author 駱昊
*
*/class Consumer implements Runnable { private List<Task> buffer; public Consumer(List<Task> buffer) { this.buffer = buffer;
}
@Override public void run() { while(true) { synchronized(buffer) { while(buffer.isEmpty()) { try {
buffer.wait();
} catch(InterruptedException e) {
e.printStackTrace();
}
} Task task = buffer.remove(0);
buffer.notifyAll();
System.out.println("Consumer[" + Thread.currentThread().getName() + "] got " + task);
}
}
}
}/**
* 生產者
* @author 駱昊
*
*/class Producer implements Runnable { private List<Task> buffer; public Producer(List<Task> buffer) { this.buffer = buffer;
}
@Override public void run() { while(true) { synchronized (buffer) { while(buffer.size() >= Constants.MAX_BUFFER_SIZE) { try {
buffer.wait();
} catch(InterruptedException e) {
e.printStackTrace();
}
} Task task = new Task();
buffer.add(task);
buffer.notifyAll();
System.out.println("Producer[" + Thread.currentThread().getName() + "] put " + task);
}
}
}
}public class Test06 { public static void main(String[] args) {
List<Task> buffer = new ArrayList<>(Constants.MAX_BUFFER_SIZE);
ExecutorService es = Executors.newFixedThreadPool(Constants.NUM_OF_CONSUMER + Constants.NUM_OF_PRODUCER); for(int i = 1; i <= Constants.NUM_OF_PRODUCER; ++i) {
es.execute(new Producer(buffer));
} for(int i = 1; i <= Constants.NUM_OF_CONSUMER; ++i) {
es.execute(new Consumer(buffer));
}
}
}

基於BlockingQueue的實現

import java.util.UUID;import java.util.concurrent.BlockingQueue;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.LinkedBlockingQueue;/**
* 公共常量
* @author 駱昊
*
*/class Constants { public static final int MAX_BUFFER_SIZE = 10; public static final int NUM_OF_PRODUCER = 2; public static final int NUM_OF_CONSUMER = 3;
}/**
* 工做任務
* @author 駱昊
*
*/class Task { private String id; // 任務的編號
public Task() {
id = UUID.randomUUID().toString();
} @Override
public String toString() { return "Task[" + id + "]";
}
}/**
* 消費者
* @author 駱昊
*
*/class Consumer implements Runnable { private BlockingQueue<Task> buffer; public Consumer(BlockingQueue<Task> buffer) { this.buffer = buffer;
} @Override
public void run() { while(true) { try {
Task task = buffer.take();
System.out.println("Consumer[" + Thread.currentThread().getName() + "] got " + task);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}/**
* 生產者
* @author 駱昊
*
*/class Producer implements Runnable { private BlockingQueue<Task> buffer; public Producer(BlockingQueue<Task> buffer) { this.buffer = buffer;
} @Override
public void run() { while(true) { try {
Task task = new Task();
buffer.put(task);
System.out.println("Producer[" + Thread.currentThread().getName() + "] put " + task);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}public class Test07 { public static void main(String[] args) {
BlockingQueue<Task> buffer = new LinkedBlockingQueue<>(Constants.MAX_BUFFER_SIZE);
ExecutorService es = Executors.newFixedThreadPool(Constants.NUM_OF_CONSUMER + Constants.NUM_OF_PRODUCER); for(int i = 1; i <= Constants.NUM_OF_PRODUCER; ++i) {
es.execute(new Producer(buffer));
} for(int i = 1; i <= Constants.NUM_OF_CONSUMER; ++i) {
es.execute(new Consumer(buffer));
}
}
}

使用BlockingQueue後代碼優雅了不少。

併發模型

在繼續下面的探討以前，咱們仍是重溫一下幾個概念：

概念	解釋
臨界資源	併發環境中有着固定數量的資源
互斥	對資源的訪問是排他式的
飢餓	一個或一組線程長時間或永遠沒法取得進展
死鎖	兩個或多個線程相互等待對方結束
活鎖	想要執行的線程老是發現其餘的線程正在執行以致於長時間或永遠沒法執行

重溫了這幾個概念後，咱們能夠探討一下下面的幾種併發模型。

生產者-消費者

一個或多個生產者建立某些工做並將其置於緩衝區或隊列中，一個或多個消費者會從隊列中得到這些工做並完成之。這裏的緩衝區或隊列是臨界資源。當緩衝區或隊列放滿的時候，生產這會被阻塞；而緩衝區或隊列爲空的時候，消費者會被阻塞。生產者和消費者的調度是經過兩者相互交換信號完成的。

讀者-寫者

當存在一個主要爲讀者提供信息的共享資源，它偶爾會被寫者更新，可是須要考慮系統的吞吐量，又要防止飢餓和陳舊資源得不到更新的問題。在這種併發模型中，如何平衡讀者和寫者是最困難的，固然這個問題至今仍是一個被熱議的問題，恐怕必須根據具體的場景來提供合適的解決方案而沒有那種放之四海而皆準的方法（不像我在國內的科研文獻中看到的那樣）。

哲學家進餐

1965年，荷蘭計算機科學家圖靈獎得主Edsger Wybe Dijkstra提出並解決了一個他稱之爲哲學家進餐的同步問題。這個問題能夠簡單地描述以下：五個哲學家圍坐在一張圓桌周圍，每一個哲學家面前都有一盤通心粉。因爲通心粉很滑，因此須要兩把叉子才能夾住。相鄰兩個盤子之間放有一把叉子以下圖所示。哲學家的生活中有兩種交替活動時段：即吃飯和思考。當一個哲學家以爲餓了時，他就試圖分兩次去取其左邊和右邊的叉子，每次拿一把，但不分次序。若是成功地獲得了兩把叉子，就開始吃飯，吃完後放下叉子繼續思考。

把上面問題中的哲學家換成線程，把叉子換成競爭的臨界資源，上面的問題就是線程競爭資源的問題。若是沒有通過精心的設計，系統就會出現死鎖、活鎖、吞吐量降低等問題。

圖4.哲學家進餐模型

下面是用信號量原語來解決哲學家進餐問題的代碼，使用了Java 5併發工具包中的Semaphore類（代碼不夠漂亮可是已經足以說明問題了）。

//import java.util.concurrent.ExecutorService;//import java.util.concurrent.Executors;import java.util.concurrent.Semaphore;/**
* 存放線程共享信號量的上下問
* @author 駱昊
*
*/class AppContext { public static final int NUM_OF_FORKS = 5; // 叉子數量(資源)
public static final int NUM_OF_PHILO = 5; // 哲學家數量(線程)
public static Semaphore[] forks; // 叉子的信號量
public static Semaphore counter; // 哲學家的信號量
static {
forks = new Semaphore[NUM_OF_FORKS]; for (int i = 0, len = forks.length; i < len; ++i) {
forks[i] = new Semaphore(1); // 每一個叉子的信號量爲1
}
counter = new Semaphore(NUM_OF_PHILO - 1); // 若是有N個哲學家，最多隻容許N-1人同時取叉子
} /**
 * 取得叉子
 * @param index 第幾個哲學家
 * @param leftFirst 是否先取得左邊的叉子
 * @throws InterruptedException
 */
public static void putOnFork(int index, boolean leftFirst) 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 第幾個哲學家
 * @param leftFirst 是否先放回左邊的叉子
 * @throws InterruptedException
 */
public static void putDownFork(int index, boolean leftFirst) 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();
}
}
}/**
* 哲學家
* @author 駱昊
*
*/class Philosopher implements Runnable { private int index; // 編號
private String name; // 名字
public Philosopher(int index, String name) { this.index = index; this.name = name;
} @Override
public void run() { while(true) { try {
AppContext.counter.acquire(); boolean leftFirst = index % 2 == 0;
AppContext.putOnFork(index, leftFirst);
System.out.println(name + "正在吃意大利麪（通心粉）..."); // 取到兩個叉子就能夠進食
AppContext.putDownFork(index, leftFirst);
AppContext.counter.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}public class Test04 { public static void main(String[] args) {
String[] names = { "駱昊", "王大錘", "張三丰", "楊過", "李莫愁" }; // 5位哲學家的名字// ExecutorService es = Executors.newFixedThreadPool(AppContext.NUM_OF_PHILO); // 建立固定大小的線程池// for(int i = 0, len = names.length; i < len; ++i) {// es.execute(new Philosopher(i, names[i])); // 啓動線程// }// es.shutdown();
for(int i = 0, len = names.length; i < len; ++i) { new Thread(new Philosopher(i, names[i])).start();
}
}
}

現實中的併發問題基本上都是這三種模型或者是這三種模型的變體。

測試併發代碼

對併發代碼的測試也是很是棘手的事情，棘手到無需說明你們也很清楚的程度，因此這裏咱們只是探討一下如何解決這個棘手的問題。咱們建議你們編寫一些可以發現問題的測試並常常性的在不一樣的配置和不一樣的負載下運行這些測試。不要忽略掉任何一次失敗的測試，線程代碼中的缺陷可能在上萬次測試中僅僅出現一次。具體來講有這麼幾個注意事項：

- 不要將系統的失效歸結於偶發事件，就像拉不出屎的時候不能怪地球沒有引力。

- 先讓非併發代碼工做起來，不要試圖同時找到併發和非併發代碼中的缺陷。

- 編寫能夠在不一樣配置環境下運行的線程代碼。

- 編寫容易調整的線程代碼，這樣能夠調整線程使性能達到最優。

- 讓線程的數量多於CPU或CPU核心的數量，這樣CPU調度切換過程當中潛在的問題纔會暴露出來。

- 讓併發代碼在不一樣的平臺上運行。

- 經過自動化或者硬編碼的方式向併發代碼中加入一些輔助測試的代碼。

知識體系

我向你們推薦一下我認爲比較全面且最系統化的學習體系（分解後的，完整的加羣能夠獲取）

1、源碼分析

2、分佈式架構

3、微服務

4、性能優化

5、Java工程化

以上就是我推薦給你們的最具備系統化的學習體系，若果你想學習以上的知識內容，你能夠加這個羣獲取：交流學習羣：650385180裏面會分享一些資深架構師錄製的視頻錄像：有Spring，MyBatis，Netty源碼分析，高併發、高性能、分佈式、微服務架構的原理，JVM性能優化這些成爲架構師必備的知識體系。還能領取免費的學習資源，目前受益良多，面試題也在羣裏面。

Java 7的併發編程

Java 7中引入了TransferQueue，它比BlockingQueue多了一個叫transfer的方法，若是接收線程處於等待狀態，該操做能夠立刻將任務交給它，不然就會阻塞直至取走該任務的線程出現。能夠用TransferQueue代替BlockingQueue，由於它能夠得到更好的性能。

剛纔忘記了一件事情，Java 5中還引入了Callable接口、Future接口和FutureTask接口，經過他們也能夠構建併發應用程序，代碼以下所示。

import java.util.ArrayList;import java.util.List;import java.util.concurrent.Callable;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.Future;public class Test07 { private static final int POOL_SIZE = 10; static class CalcThread implements Callable<Double> { private List<Double> dataList = new ArrayList<>(); public CalcThread() { for(int i = 0; i < 10000; ++i) {
dataList.add(Math.random());
}
} @Override
public Double call() throws Exception { double total = 0; for(Double d : dataList) {
total += d;
} return total / dataList.size();
}
} public static void main(String[] args) {
List<Future<Double>> fList = new ArrayList<>();
ExecutorService es = Executors.newFixedThreadPool(POOL_SIZE); for(int i = 0; i < POOL_SIZE; ++i) {
fList.add(es.submit(new CalcThread()));
} for(Future<Double> f : fList) { try {
System.out.println(f.get());
} catch (Exception e) {
e.printStackTrace();
}
}
es.shutdown();
}
}

Callable接口也是一個單方法接口，顯然這是一個回調方法，相似於函數式編程中的回調函數，在Java 8 之前，Java中還不能使用Lambda表達式來簡化這種函數式編程。和Runnable接口不一樣的是Callable接口的回調方法call方法會返回一個對象，這個對象能夠用未來時的方式在線程執行結束的時候得到信息。上面代碼中的call方法就是將計算出的10000個0到1之間的隨機小數的平均值返回，咱們經過一個Future接口的對象獲得了這個返回值。目前最新的Java版本中，Callable接口和Runnable接口都被打上了@FunctionalInterface的註解，也就是說它能夠用函數式編程的方式（Lambda表達式）建立接口對象。

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

- get()：獲取結果。若是結果尚未準備好，get方法會阻塞直到取得結果；固然也能夠經過參數設置阻塞超時時間

- cancel()：在運算結束前取消

- isDone()：能夠用來判斷運算是否結束

Java 7中還提供了分支/合併（fork/join）框架，它能夠實現線程池中任務的自動調度，而且這種調度對用戶來講是透明的。爲了達到這種效果，必須按照用戶指定的方式對任務進行分解，而後再將分解出的小型任務的執行結果合併成原來任務的執行結果。這顯然是運用了分治法（divide-and-conquer）的思想。下面的代碼使用了分支/合併框架來計算1到10000的和，固然對於如此簡單的任務根本不須要分支/合併框架，由於分支和合並自己也會帶來必定的開銷，可是這裏咱們只是探索一下在代碼中如何使用分支/合併框架，讓咱們的代碼可以充分利用現代多核CPU的強大運算能力。

import java.util.concurrent.ForkJoinPool;import java.util.concurrent.Future;import java.util.concurrent.RecursiveTask;class Calculator extends RecursiveTask<Integer> { private static final long serialVersionUID = 7333472779649130114L; private static final int THRESHOLD = 10; private int start; private int end; public Calculator(int start, int end) { this.start = start; this.end = end;
} @Override
public Integer compute() { int sum = 0; if ((end - start) < THRESHOLD) { // 當問題分解到可求解程度時直接計算結果
for (int i = start; i <= end; i++) {
sum += i;
}
} else { int middle = (start + end) >>> 1; // 將任務一分爲二
Calculator left = new Calculator(start, middle);
Calculator right = new Calculator(middle + 1, end);
left.fork();
right.fork(); // 注意：因爲此處是遞歸式的任務分解，也就意味着接下來會二分爲四，四分爲八...
sum = left.join() + right.join(); // 合併兩個子任務的結果
} return sum;
}
}public class Test08 { public static void main(String[] args) throws Exception {
ForkJoinPool forkJoinPool = new ForkJoinPool();
Future<Integer> result = forkJoinPool.submit(new Calculator(1, 10000));
System.out.println(result.get());
}
}

在已經到來的Java 7中，Java中默認的數組排序算法已經再也不是經典的快速排序（雙樞軸快速排序）了，新的排序算法叫TimSort，它是歸併排序和插入排序的混合體，TimSort能夠經過分支合併框架充分利用現代處理器的多核特性，從而得到更好的性能（更短的排序時間）。

而伴隨着Java10的到來，排序算法會不會有一些新的變化呢，新的變化又會是什麼呢，讓咱們拭目以待吧。