本文的組織形式如下，主要會介紹到同步容器類，作業系統的併發工具，Java 開發工具包（只是簡單介紹一下，後面會有原始碼分析）。同步工具類有哪些。 下面我們就來介紹一下 Java 併發中都涉及哪些模組，這些併發模組都是 Java 併發類庫所提供的。 ## 同步容器類 同步容器主要包括兩類，一種是本來就是執行緒安全實現的容器，這類容器有 **Vector、Hashtable、Stack**，這類容器的方法上都加了 `synchronized` 鎖，是執行緒安全的實現。 > Vector、Hashtable、Stack 這些容器我們現在幾乎都不在使用，因為這些容器在多執行緒環境下的效率不高。 還有一類是由 `Collections.synchronizedxxx` 實現的非執行緒安全的容器，使用 Collections.synchronized 會把它們封裝起來程式設計執行緒安全的容器，舉出兩個例子 * Collections.synchronizedList * Collections.synchronizedMap 我們可以通過 Collections 原始碼可以看出這些執行緒安全的實現 要不為啥要稱 Collections 為集合工具類呢？Collections 會把這些容器類的狀態封裝起來，並對每個同步方法進行同步，使得每次只有一個執行緒能夠訪問容器的狀態。 其中每個 `synchronized xxx `都是相當於建立了一個靜態內部類。  雖然同步容器類都是執行緒安全的，但是在某些情況下需要額外的客戶端加鎖來保證一些複合操作的安全性，複合操作就是有兩個及以上的方法組成的操作，比如最典型的就是 `若沒有則新增`，用偽程式碼表示則是 ```java if(a == null){ a = get(); } ``` 比如可以用來判斷 Map 中是否有某個 key，如果沒有則新增進 Map 中。這些複合操作在沒有客戶端加鎖的情況下是執行緒安全的，但是當多個執行緒併發修改容器時，可能會表現出意料之外的行為。例如下面這段程式碼 ```java public class TestVector implements Runnable{ static Vector vector = new Vector(); static void addVector(){ for(int i = 0;i < 10000;i++){ vector.add(i); } } static Object getVector(){ int index = vector.size() - 1; return vector.get(index); } static void removeVector(){ int index = vector.size() - 1; vector.remove(index); } @Override public void run() { getVector(); } public static void main(String[] args) { TestVector testVector = new TestVector(); testVector.addVector(); Thread t1 = new Thread(() -> { for(int i = 0;i < vector.size();i++){ getVector(); } }); Thread t2 = new Thread(() -> { for(int i = 0;i < vector.size();i++){ removeVector(); } }); t1.start(); t2.start(); } } ``` 這些方法看似沒有問題，因為 Vector 能夠保證執行緒安全性，無論多少個執行緒訪問 Vector 也不會造成 Vector 的內部產生破壞，但是從整個系統來說，是存線上程安全性的，事實上你執行一下，也會發現報錯。 會出現 如果執行緒 A 在包含這麼多元素的基礎上呼叫 `getVector` 方法，會得到一個數值，getVector 只是取得該元素，而並不是從 vector 中移除，`removeVector` 方法是得到一個元素進行移除，這段程式碼的不安全因素就是，因為執行緒的時間片是亂序的，而且 getVector 和 removeVector 並不會保證互斥，所以在 removeVector 方法把某個值比如 6666 移除後，vector 中就不存在這個 6666 的元素，此時 getVector 方法取得 6666 ，就會丟擲陣列越界異常。為什麼是陣列越界異常呢？可以看一下 vector 的原始碼 如果用圖表示的話，則會是下面這樣。 所以，從系統的層面來看，上面這段程式碼也要保證執行緒安全性才可以，也就是在`客戶端加鎖` 實現，只要我們讓複合操作使用一把鎖，那麼這些操作就和其他單獨的操作一樣都是原子性的。如下面例子所示 ```java static Object getVector(){ synchronized (vector){ int index = vector.size() - 1; return vector.get(index); } } static void removeVector(){ synchronized (vector) { int index = vector.size() - 1; vector.remove(index); } } ``` 也可以通過鎖住 `.class` 來保證原子性操作，也能達到同樣的效果。 ```java static Object getVector(){ synchronized (TestVector.class){ int index = vector.size() - 1; return vector.get(index); } } static void removeVector(){ synchronized (TestVector.class) { int index = vector.size() - 1; vector.remove(index); } } ``` 在呼叫 size 和 get 之間，Vector 的長度可能會發生變化，這種變化在對 Vector 進行排序時出現，如下所示 ```java for(int i = 0;i< vector.size();i++){ doSomething(vector.get(i)); } ``` 這種迭代的操作正確性取決於運氣，即在呼叫 size 和 get 之間會修改 Vector，在單執行緒環境中，這種假設完全成立，但是再有其他執行緒併發修改 Vector 時，則可能會導致麻煩。 我們仍舊可以通過客戶端加鎖的方式來避免這種情況 ```java synchronized(vector){ for(int i = 0;i< vector.size();i++){ doSomething(vector.get(i)); } } ``` 這種方式為客戶端的可靠性提供了保證，但是犧牲了伸縮性，而且這種在遍歷過程中進行加鎖，也不是我們所希望看到的。 ### fail-fast 針對上面這種情況，很多集合類都提供了一種 `fail-fast` 機制，因為大部分集合內部都是使用 Iterator 進行遍歷，在迴圈中使用同步鎖的開銷會很大，而 Iterator 的建立是輕量級的，所以在集合內部如果有併發修改的操作，集合會進行`快速失敗`，也就是 `fail-fast`。當他們發現容器在迭代過程中被修改時，會丟擲 `ConcurrentModificationException` 異常，這種快速失敗不是一種完備的處理機制，而只是 `善意`的捕獲併發錯誤。 如果檢視過 ConcurrentModificationException 的註解，你會發現，ConcurrentModificationException 丟擲的原則由兩種，如下 造成這種異常的原因是由於多個執行緒在遍歷集合的同時對集合類內部進行了修改，這也就是 fail-fast 機制。 該註解還聲明瞭另外一種方式 這個問題也是很經典的一個問題，我們使用 ArrayList 來舉例子。如下程式碼所示 ```java public static void main(String[] args) { List list = new ArrayList<>(); for (int i = 0 ; i < 10 ; i++ ) { list.add(i + ""); } Iterator iterator = list.iterator(); int i = 0 ; while(iterator.hasNext()) { if (i == 3) { list.remove(3); } System.out.println(iterator.next()); i ++; } } ``` 該段程式碼會發生異常，因為在 ArrayList 內部，有兩個屬性，一個是 `modCount` ，一個是 `expectedModCount` ，ArrayList 在 remove 等對集合結構的元素造成數量上的操作會有 `checkForComodification` 的判斷，如下所示，這也是這段程式碼的錯誤原因。 ###fail-safe `fail-safe` 是 Java 中的一種 `安全失敗` 機制，它表示的是在遍歷時不是直接在原集合上進行訪問，而是先複製原有集合內容，在拷貝的集合上進行遍歷。 由於迭代時是對原集合的拷貝進行遍歷，所以在遍歷過程中對原集合所作的修改並不能被迭代器檢測到，所以不會觸發 ConcurrentModificationException。`java.util.concurrent` 包下的容器都是安全失敗的，可以在多執行緒條件下使用，併發修改。 比如 `CopyOnWriteArrayList`， 它就是一種 fail-safe 機制的集合，它就不會出現異常，例如如下操作 ```java List integers = new CopyOnWriteArrayList<>(); integers.add(1); integers.add(2); integers.add(3); Iterator itr = integers.iterator(); while (itr.hasNext()) { Integer a = itr.next(); integers.remove(a); } ``` CopyOnWriteArrayList 就是 ArrayList 的一種執行緒安全的變體，CopyOnWriteArrayList 中的所有可變操作比如 add 和 set 等等都是通過對陣列進行全新複製來實現的。 ## 作業系統中的併發工具 講到併發容器，就不得不提作業系統級別實現了哪些程序/執行緒間的併發容器，說白了其實就是資料結構的設計。下面我們就來一起看一下作業系統級別的併發工具 ### 訊號量 訊號量是 E.W.Dijkstra 在 1965 年提出的一種方法，它使用一個整形變數來累計喚醒次數，以供之後使用。在他的觀點中，有一個新的變數型別稱作 `訊號量(semaphore)`。一個訊號量的取值可以是 0 ，或任意正數。0 表示的是不需要任何喚醒，任意的正數表示的就是喚醒次數。 Dijkstra 提出了訊號量有兩個操作，現在通常使用 `down` 和 `up`（分別可以用 sleep 和 wakeup 來表示）。down 這個指令的操作會檢查值是否大於 0 。如果大於 0 ，則將其值減 1 ；若該值為 0 ，則程序將睡眠，而且此時 down 操作將會繼續執行。檢查數值、修改變數值以及可能發生的睡眠操作均為一個單一的、不可分割的 `原子操作(atomic action)` 完成。 ### 互斥量 如果不需要訊號量的計數能力時，可以使用訊號量的一個簡單版本，稱為 `mutex(互斥量)`。互斥量的優勢就在於在一些共享資源和一段程式碼中保持互斥。由於互斥的實現既簡單又有效，這使得互斥量在實現使用者空間執行緒包時非常有用。 互斥量是一個處於兩種狀態之一的共享變數：`解鎖(unlocked)` 和 `加鎖(locked)`。這樣，只需要一個二進位制位來表示它，不過一般情況下，通常會用一個 `整型(integer)` 來表示。0 表示解鎖，其他所有的值表示加鎖，比 1 大的值表示加鎖的次數。 mutex 使用兩個過程，當一個執行緒（或者程序）需要訪問關鍵區域時，會呼叫 `mutex_lock` 進行加鎖。如果互斥鎖當前處於解鎖狀態（表示關鍵區域可用），則呼叫成功，並且呼叫執行緒可以自由進入關鍵區域。 另一方面，如果 mutex 互斥量已經鎖定的話，呼叫執行緒會阻塞直到關鍵區域內的執行緒執行完畢並且呼叫了 `mutex_unlock` 。如果多個執行緒在 mutex 互斥量上阻塞，將隨機選擇一個執行緒並允許它獲得鎖。 #### Futexes 隨著並行的增加，有效的`同步(synchronization)`和`鎖定(locking)` 對於效能來說是非常重要的。如果程序等待時間很短，那麼`自旋鎖(Spin lock)` 是非常有效；但是如果等待時間比較長，那麼這會浪費 CPU 週期。如果程序很多，那麼阻塞此程序，並僅當鎖被釋放的時候讓核心解除阻塞是更有效的方式。不幸的是，這種方式也會導致另外的問題：它可以在程序競爭頻繁的時候執行良好，但是在競爭不是很激烈的情況下核心切換的消耗會非常大，而且更困難的是，預測鎖的競爭數量更不容易。 有一種有趣的解決方案是把兩者的優點結合起來，提出一種新的思想，稱為 `futex`，或者是 `快速使用者空間互斥(fast user space mutex)`，是不是聽起來很有意思？ futex 是 `Linux` 中的特性實現了基本的鎖定（很像是互斥鎖）而且避免了陷入核心中，因為核心的切換的開銷非常大，這樣做可以大大提高效能。futex 由兩部分組成：**核心服務和使用者庫**。核心服務提供了了一個 `等待佇列(wait queue)` 允許多個程序在鎖上排隊等待。除非核心明確的對他們解除阻塞，否則它們不會執行。 #### Pthreads 中的互斥量 Pthreads 提供了一些功能用來同步執行緒。最基本的機制是使用互斥量變數，可以鎖定和解鎖，用來保護每個關鍵區域。希望進入關鍵區域的執行緒首先要嘗試獲取 mutex。如果 mutex 沒有加鎖，執行緒能夠馬上進入並且互斥量能夠自動鎖定，從而阻止其他執行緒進入。如果 mutex 已經加鎖，呼叫執行緒會阻塞，直到 mutex 解鎖。如果多個執行緒在相同的互斥量上等待，當互斥量解鎖時，只有一個執行緒能夠進入並且重新加鎖。這些鎖並不是必須的，程式設計師需要正確使用它們。 下面是與互斥量有關的函式呼叫 和我們想象中的一樣，mutex 能夠被建立和銷燬，扮演這兩個角色的分別是 `Phread_mutex_init` 和 `Pthread_mutex_destroy`。mutex 也可以通過 `Pthread_mutex_lock` 來進行加鎖，如果互斥量已經加鎖，則會阻塞呼叫者。還有一個呼叫`Pthread_mutex_trylock` 用來嘗試對執行緒加鎖，當 mutex 已經被加鎖時，會返回一個錯誤程式碼而不是阻塞呼叫者。這個呼叫允許執行緒有效的進行忙等。最後，`Pthread_mutex_unlock` 會對 mutex 解鎖並且釋放一個正在等待的執行緒。 除了互斥量以外，`Pthreads` 還提供了第二種同步機制： `條件變數(condition variables)` 。mutex 可以很好的允許或阻止對關鍵區域的訪問。條件變數允許執行緒由於未滿足某些條件而阻塞。絕大多數情況下這兩種方法是一起使用的。下面我們進一步來研究執行緒、互斥量、條件變數之間的關聯。 下面再來重新認識一下生產者和消費者問題：一個執行緒將東西放在一個緩衝區內，由另一個執行緒將它們取出。如果生產者發現緩衝區沒有空槽可以使用了，生產者執行緒會阻塞起來直到有一個執行緒可以使用。生產者使用 mutex 來進行原子性檢查從而不受其他執行緒干擾。但是當發現緩衝區已經滿了以後，生產者需要一種方法來阻塞自己並在以後被喚醒。這便是條件變數做的工作。 下面是一些與條件變數有關的最重要的 pthread 呼叫 上表中給出了一些呼叫用來建立和銷燬條件變數。條件變數上的主要屬性是 `Pthread_cond_wait` 和 `Pthread_cond_signal`。前者阻塞呼叫執行緒，直到其他執行緒發出訊號為止（使用後者呼叫）。阻塞的執行緒通常需要等待喚醒的訊號以此來釋放資源或者執行某些其他活動。只有這樣阻塞的執行緒才能繼續工作。條件變數允許等待與阻塞原子性的程序。`Pthread_cond_broadcast` 用來喚醒多個阻塞的、需要等待訊號喚醒的執行緒。 >需要注意的是，條件變數（不像是訊號量）不會存在於記憶體中。如果將一個訊號量傳遞給一個沒有執行緒等待的條件變數，那麼這個訊號就會丟失，這個需要注意 ### 管程 為了能夠編寫更加準確無誤的程式，Brinch Hansen 和 Hoare 提出了一個更高階的同步原語叫做 `管程(monitor)`。管程有一個很重要的特性，即在任何時候管程中只能有一個活躍的程序，這一特性使管程能夠很方便的實現互斥操作。管程是程式語言的特性，所以編譯器知道它們的特殊性，因此可以採用與其他過程呼叫不同的方法來處理對管程的呼叫。通常情況下，當程序呼叫管程中的程式時，該程式的前幾條指令會檢查管程中是否有其他活躍的程序。如果有的話，呼叫程序將被掛起，直到另一個程序離開管程才將其喚醒。如果沒有活躍程序在使用管程，那麼該呼叫程序才可以進入。 進入管程中的互斥由編譯器負責，但是一種通用做法是使用 `互斥量(mutex)` 和 `二進位制訊號量(binary semaphore)`。由於編譯器而不是程式設計師在操作，因此出錯的機率會大大降低。在任何時候，編寫管程的程式設計師都無需關心編譯器是如何處理的。他只需要知道將所有的臨界區轉換成為管程過程即可。絕不會有兩個程序同時執行臨界區中的程式碼。 即使管程提供了一種簡單的方式來實現互斥，但在我們看來，這還不夠。因為我們還需要一種在程序無法執行被阻塞。在生產者-消費者問題中，很容易將針對緩衝區滿和緩衝區空的測試放在管程程式中，但是生產者在發現緩衝區滿的時候該如何阻塞呢？ 解決的辦法是引入`條件變數(condition variables)` 以及相關的兩個操作 `wait` 和 `signal`。當一個管程程式發現它不能執行時（例如，生產者發現緩衝區已滿），它會在某個條件變數（如 full）上執行 `wait` 操作。這個操作造成呼叫程序阻塞，並且還將另一個以前等在管程之外的程序調入管程。在前面的 pthread 中我們已經探討過條件變數的實現細節了。另一個程序，比如消費者可以通過執行 `signal` 來喚醒阻塞的呼叫程序。 **通過臨界區自動的互斥，管程比訊號量更容易保證並行程式設計的正確性**。但是管程也有缺點，我們前面說到過管程是一個程式語言的概念，編譯器必須要識別管程並用某種方式對其互斥作出保證。**C、Pascal 以及大多數其他程式語言都沒有管程**，所以不能依靠編譯器來遵守互斥規則。 與管程和訊號量有關的另一個問題是，這些機制都是設計用來解決訪問共享記憶體的一個或多個 CPU 上的互斥問題的。通過將訊號量放在共享記憶體中並用 `TSL` 或 `XCHG` 指令來保護它們，可以避免競爭。但是如果是在分散式系統中，可能同時具有多個 CPU 的情況，並且每個 CPU 都有自己的私有記憶體呢，它們通過網路相連，那麼這些原語將會失效。因為訊號量太低階了，而管程在少數幾種程式語言之外無法使用，所以還需要其他方法。 ### 訊息傳遞 上面提到的其他方法就是 `訊息傳遞(messaage passing)`。這種程序間通訊的方法使用兩個原語 `send` 和 `receive` ，它們像訊號量而不像管程，是系統呼叫而不是語言級別。示例如下 ```c send(destination, &message); receive(source, &message); ``` send 方法用於向一個給定的目標傳送一條訊息，receive 從一個給定的源接收一條訊息。如果沒有訊息，接受者可能被阻塞，直到接收一條訊息或者帶著錯誤碼返回。 訊息傳遞系統現在面臨著許多訊號量和管程所未涉及的問題和設計難點，尤其對那些在網路中不同機器上的通訊狀況。例如，訊息有可能被網路丟失。為了防止訊息丟失，傳送方和接收方可以達成一致：一旦接受到訊息後，接收方馬上回送一條特殊的 `確認(acknowledgement)` 訊息。如果傳送方在一段時間間隔內未收到確認，則重發訊息。 現在考慮訊息本身被正確接收，而返回給傳送著的確認訊息丟失的情況。傳送者將重發訊息，這樣接受者將收到兩次相同的訊息。 對於接收者來說，如何區分新的訊息和一條重發的老訊息是非常重要的。通常採用在每條原始訊息中嵌入一個連續的序號來解決此問題。如果接受者收到一條訊息，它具有與前面某一條訊息一樣的序號，就知道這條訊息是重複的，可以忽略。 訊息系統還必須處理如何命名程序的問題，以便在傳送或接收呼叫中清晰的指明程序。`身份驗證(authentication)` 也是一個問題，比如客戶端怎麼知道它是在與一個真正的檔案伺服器通訊，從傳送方到接收方的資訊有可能被中間人所篡改。 ### 屏障 最後一個同步機制是準備用於程序組而不是程序間的生產者-消費者情況的。在某些應用中劃分了若干階段，並且規定，除非所有的程序都就緒準備著手下一個階段，否則任何程序都不能進入下一個階段，可以通過在每個階段的結尾安裝一個 `屏障(barrier)` 來實現這種行為。當一個程序到達屏障時，它會被屏障所攔截，直到所有的屏障都到達為止。屏障可用於一組程序同步，如下圖所示 在上圖中我們可以看到，有四個程序接近屏障，這意味著每個程序都在進行運算，但是還沒有到達每個階段的結尾。過了一段時間後，A、B、D 三個程序都到達了屏障，各自的程序被掛起，但此時還不能進入下一個階段呢，因為程序 B 還沒有執行完畢。結果，當最後一個 C 到達屏障後，這個程序組才能夠進入下一個階段。 ### 避免鎖：讀-複製-更新 最快的鎖是根本沒有鎖。問題在於沒有鎖的情況下，我們是否允許對共享資料結構的併發讀寫進行訪問。答案當然是不可以。假設程序 A 正在對一個數字陣列進行排序，而程序 B 正在計算其平均值，而此時你進行 A 的移動，會導致 B 會多次讀到重複值，而某些值根本沒有遇到過。 然而，在某些情況下，我們可以允許寫操作來更新資料結構，即便還有其他的程序正在使用。竅門在於確保每個讀操作要麼讀取舊的版本，要麼讀取新的版本，例如下面的樹 上面的樹中，讀操作從根部到葉子遍歷整個樹。加入一個新節點 X 後，為了實現這一操作，我們要讓這個節點在樹中可見之前使它"恰好正確"：我們對節點 X 中的所有值進行初始化，包括它的子節點指標。然後通過原子寫操作，使 X 稱為 A 的子節點。所有的讀操作都不會讀到前後不一致的版本 在上面的圖中，我們接著移除 B 和 D。首先，將 A 的左子節點指標指向 C 。所有原本在 A 中的讀操作將會後續讀到節點 C ，而永遠不會讀到 B 和 D。也就是說，它們將只會讀取到新版資料。同樣，所有當前在 B 和 D 中的讀操作將繼續按照原始的資料結構指標並且讀取舊版資料。所有操作均能正確執行，我們不需要鎖住任何東西。而不需要鎖住資料就能夠移除 B 和 D 的主要原因就是 `讀-複製-更新(Ready-Copy-Update,RCU)`，將更新過程中的移除和再分配過程分離開。 ## Java 併發工具包 JDK 1.5 提供了許多種併發容器來改進同步容器的效能，同步容器將所有對容器狀態的訪問都序列化，以實現他們之間的執行緒安全性。這種方法的代價是嚴重降低了併發效能，當多個執行緒爭搶容器鎖的同時，嚴重降低吞吐量。 下面我們就來一起認識一下 Java 中都用了哪些併發工具 ### Java 併發工具綜述 在 Java 5.0 中新增加了 `ConcurrentHashMap` 用來替代基於雜湊的 Map 容器；新增加了 `CopyOnWriteArrayList` 和 `CopyOnWriteArraySet` 來分別替代 ArrayList 和 Set 介面實現類；還新增加了兩種容器型別，分別是 `Queue` 和 `BlockingQueue`， Queue 是佇列的意思，它有一些實現分別是傳統的先進先出佇列 `ConcurrentLinkedQueue`以及併發優先順序佇列 `PriorityQueue`。Queue 是一個先入先出的佇列，它的操作不會阻塞，如果佇列為空那麼獲取元素的操作會返回空值。PriorityQueue 擴充套件了 Queue，增加了可阻塞的插入和獲取等操作。如果佇列為空，那麼獲取元素的操作將一直阻塞，直到佇列中出現一個可用的元素為止。如果佇列已滿，那麼插入操作則一直阻塞，直到佇列中有可用的空間為止。 Java 6.0 還引入了 `ConcurrentSkipListMap` 和 `ConcurrentSkipListSet` 分別作為同步的 SortedMap 和 SortedSet 的併發替代品。下面我們就展開探討了，設計不到底層原始碼，因為本篇文章主要目的就是為了描述一下有哪些東西以及用了哪些東西。 ### ConcurrentHashMap 我們先來看一下 ConcurrentHashMap 在併發集合中的位置 可以看到，ConcurrentHashMap 繼承了 `AbstractMap` 介面並實現了 ConcurrentMap 和 Serializable 介面，AbstractMap 和 ConcurrentMap 都是 Map 的實現類，只不過 AbstractMap 是抽象實現。 ConcurrentHashMap 和 Hashtable 的構造非常相似，只不過 Hashtable 容器在激烈競爭的場景中會表現出效率低下的現象，這是因為所有訪問 Hashtable 的執行緒都想獲取同一把鎖，如果容器裡面有多把鎖，並且每一把鎖都只用來鎖定一段資料，那麼當多個執行緒訪問不同的資料段時，就不存在競爭關係。這就是 ConcurreentHashMap 採用的 `分段鎖` 實現。在這種鎖實現中，任意數量的讀取執行緒可以併發的訪問 Map，執行讀取操作的執行緒和執行寫入的執行緒可以併發的訪問 Map，並且在讀取的同時也可以併發修改 Map。 ConcurrentHashMap 分段鎖實現帶來的結果是，在併發環境下可以實現更高的吞吐量，在單執行緒環境下只損失非常小的效能。 你知道 HashMap 是具有 fail-fast 機制的，也就是說它是一種強一致性的集合，在資料不一致的情況下會丟擲 `ConcurrentModificationException` 異常，而 ConcurrentHashMap 是一種 `弱一致性` 的集合，在併發修改其內部結構時，它不會丟擲 ConcurrentModificationException 異常，弱一致性能夠容忍併發修改。 在 HashMap 中，我們一般使用的 size、empty、containsKey 等方法都是標準方法，其返回的結果是一定的，包含就是包含，不包含就是不包含，可以作為判斷條件；而 ConcurrentHashMap 中的這些方法只是參考方法，它不是一個 `精確值`，像是 size、empty 這些方法在併發場景下用處很小，因為他們的返回值總是在不斷變化，所以這些操作的需求就被弱化了。 在 ConcurrentHashMap 中沒有實現對 Map 加鎖從而實現獨佔訪問。線上程安全的 Map 實現 `Hashtable` 和 `Collections.synchronizedMap` 中都實現了獨佔訪問，因此只能單個執行緒修改 Map 。ConcurrentHashMap 與這些 Map 容器相比，具有更多的優勢和更少的劣勢，只有當需要獨佔訪問的需求時才會使用 Hashtable 或者是 Collections.synchronizedMap ，否則其他併發場景下，應該使用 ConcurrentHashMap。 ### ConcurrentMap ConcurrentMap 是一個介面，它繼承了 Map 介面並提供了 Map 介面中四個新的方法，這四個方法都是 `原子性` 方法，進一步擴充套件了 Map 的功能。 ```java public interface ConcurrentMap extends Map { // 僅當 key 沒有相應的對映值時才插入 V putIfAbsent(K key, V value); // 僅當 key 被對映到 value 時才移除 boolean remove(Object key, Object value); // 僅當 key 被對映到 value 時才移除 V replace(K key, V value); // 僅當 key 被對映到 oldValue 時才替換為 newValue boolean replace(K key, V oldValue, V newValue); } ``` #### ConcurrentNavigableMap `java.util.concurrent.ConcurrentNavigableMap` 類是 `java.util.NavigableMap` 的子類，它支援併發訪問，並且允許其檢視的併發訪問。 什麼是檢視呢？檢視就是集合中的一段資料序列，ConcurrentNavigableMap 中支援使用 `headMap`、`subMap`、`tailMap` 返回的檢視。與其重新解釋一下 NavigableMap 中找到的所有方法，不如看一下 ConcurrentNavigableMap 中新增的方法 * headMap 方法：headMap 方法返回一個嚴格小於給定鍵的檢視 * tailMap 方法：tailMap 方法返回包含大於或等於給定鍵的檢視。 * subMap 方法：subMap 方法返回給定兩個引數的檢視 ConcurrentNavigableMap 介面包含一些可能有用的其他方法 * descendingKeySet() * descendingMap() * navigableKeySet() 更多關於方法的描述這裡就不再贅述了，讀者朋友們可自行查閱 javadoc #### ConcurrentSkipListMap `ConcurrentSkipListMap` 是執行緒安全的有序的雜湊表，適用於高併發的場景。 ConcurrentSkipListMap 的底層資料結構是基於`跳錶`實現的。ConcurrentSkipListMap 可以提供 Comparable 內部排序或者是 Comparator 外部排序，具體取決於使用哪個建構函式。 #### ConcurrentSkipListSet `ConcurrentSkipListSet` 是執行緒安全的有序的集合，適用於高併發的場景。ConcurrentSkipListSet 底層是通過 ConcurrentNavigableMap 來實現的，它是一個有序的執行緒安全的集合。 ConcurrentSkipListSet有序的，基於元素的自然排序或者通過比較器確定的順序； ConcurrentSkipListSet是執行緒安全的； ### CopyOnWriteArrayList CopyOnWriteArrayList 是 ArrayList 的變體，在 CopyOnWriteArrayList 中，所有可變操作比如 add、set 其實都是重新建立了一個副本，通過對陣列進行復制而實現的。 CopyOnWriteArrayList 其內部有一個指向陣列的引用，陣列是不會被修改的，每次併發修改 CopyOnWriteArrayList 都相當於重新建立副本，CopyOnWriteArrayList 是一種 `fail-safe` 機制的，它不會丟擲 ConcurrentModificationException 異常，並且返回元素與迭代器建立時的元素相同。 每次併發寫操作都會建立新的副本，這個過程存在一定的開銷，所以，一般在規模很大時，讀操作要遠遠多於寫操作時，為了保證執行緒安全性，會使用 CopyOnWriteArrayList。 類似的，CopyOnWriteArraySet 的作用也相當於替代了 Set 介面。 ### BlockingQueue BlockingQueue 譯為 `阻塞佇列`，它是 JDK 1.5 新增的新的工具類，它繼承於 Queue `佇列`，並擴充套件了 Queue 的功能。 BlockingQueue 在檢索元素時會等待佇列變成非空，並在儲存元素時會等待佇列變為可用。BlockingQueue 的方法有四種實現形式，以不同的方式來處理。 * 第一種是`丟擲異常` * `特殊值`：第二種是根據情況會返回 null 或者 false * `阻塞`：第三種是無限期的阻塞當前執行緒直到操作變為可用後 * `超時`：第四種是給定一個最大的超時時間，超過後才會放棄 BlockingQueue 不允許新增 null 元素，在其實現類的方法 **add、put 或者 offer** 後時新增 null 會丟擲空指標異常。BlockingQueue 會有容量限制。在任意時間內，它都會有一個 remainCapacity，超過該值之前，任意 put 元素都會阻塞。 BlockingQueue 一般用於實現`生產者 - 消費者` 佇列，如下圖所示 BlockingQueue 有多種實現，下面我們一起來認識一下這些容器。 其中 `LinkedBlockingQueue` 和 `ArrayBlockingQueue` 是 FIFO 先入先出佇列，二者分別和 `LinkedList`和 `ArrayList` 對應，比同步 List 具有更好的併發效能。 `PriorityBlockingQueue` 是一個優先順序排序的阻塞佇列，如果你希望按照某種順序而不是 FIFO 處理元素時這個佇列將非常有用。正如其他有序的容器一樣，PriorityBlockingQueue 既可以按照自然順序來比較元素，也可以使用 `Comparator` 比較器進行外部元素比較。`SynchronousQueue` 它維護的是一組執行緒而不是一組佇列，實際上它不是一個佇列，它的每個 insert 操作必須等待其他相關執行緒的 remove 方法後才能執行，反之亦然。 #### LinkedBlockingQueue `LinkedBlockingQueue` 是一種 BlockingQueue 的實現。 它是一種基於連結串列的構造、先入先出的有界阻塞佇列。佇列的 `head` 也就是頭元素是在佇列中等待時間最長的元素；佇列的 `tail`也就是隊尾元素是佇列中等待時間最短的元素。新的元素會被插入到隊尾中，檢索操作將獲取佇列中的頭部元素。連結串列佇列通常比基於陣列的佇列具有更高的吞吐量，但是在大多數併發應用程式中，可預測的效能較差。 #### ArrayBlockingQueue `ArrayBlockingQueue` 是一個用陣列實現的有界佇列，此佇列順序按照`先入先出`的原則對元素進行排序。 預設情況下不保證執行緒公平的訪問佇列，所謂公平訪問佇列指的是阻塞的執行緒，可以按照阻塞的先後順序訪問，即先阻塞執行緒先訪問佇列。非公平性是對先等待的執行緒是非公平的。有可能先阻塞的執行緒最後才訪問佇列。 #### PriorityBlockingQueue `PriorityBlockingQueue` 是一個支援優先順序的阻塞佇列，預設情況下的元素採取自然順序生序或者降序，也可以自己定義 Comparator 進行外部排序。但需要注意的是不能保證同優先順序元素的順序。 #### DelayQueue `DelayQueue` 是一個支援延時獲取元素的無阻塞佇列，其中的元素只能在延遲到期後才能使用，DelayQueue 中的佇列頭是延遲最長時間的元素，如果沒有延遲，則沒有 head 頭元素，poll 方法會返回 null。判斷的依據就是 `getDelay(TimeUnit.NANOSECONDS)` 方法返回一個值小於或者等於 0 就會發生過期。 #### TransferQueue TransferQueue 繼承於 BlockingQueue，它是一個介面，一個 BlockingQueue 是一個生產者可能等待消費者接受元素，TransferQueue 則更進一步，生產者會一直阻塞直到所新增到佇列的元素被某一個消費者所消費，新新增的transfer 方法用來實現這種約束。 TransferQueue 有下面這些方法：兩個 `tryTransfer` 方法，一個是非阻塞的，另一個是帶有 timeout 引數設定超時時間的。還有兩個輔助方法 `hasWaitingConsumer` 和 `getWaitingConcusmerCount`。 #### LinkedTransferQueue 一個無界的基於連結串列的 TransferQueue。這個佇列對任何給定的生產者進行 FIFO 排序，head 是佇列中存在時間最長的元素。tail 是佇列中存在時間最短的元素。 ### BlockingDeque 與 BlockingQueue 相對的還有 BlockingDeque 和 Deque，它們是 JDK1.6 被提出的，分別對 Queue 和 BlockingQueue 做了擴充套件。 `Deque` 是一個雙端佇列，分別實現了在佇列頭和佇列尾的插入。Deque 的實現有 `ArrayDeque`、`ConcurrentLinkedDeque`，BlockingDeque 的實現有 `LinkedBlockingDeque` 。 **阻塞模式一般用於生產者 - 消費者佇列，而雙端佇列適用於工作密取**。在工作密取的設計中，每個消費者都有各自的雙端佇列，如果一個消費者完成了自己雙端佇列的任務，就會去其他雙端佇列的末尾進行消費。密取方式要比傳統的生產者 - 消費者佇列具有更高的可伸縮性，這是因為每個工作密取的工作者都有自己的雙端佇列，不存在競爭的情況。 #### ArrayDeque ArrayDeque 是 Deque 的可動態調整大小的陣列實現，其內部沒有容量限制，他們會根據需要進行增長。ArrayDeque 不是執行緒安全的，如果沒有外部加鎖的情況下，不支援多執行緒訪問。ArrayDeque 禁止空元素，這個類作為棧使用時要比 Stack 快，作為 queue 使用時要比 LinkedList 快。 除了 remove、removeFirstOccurrence、removeLastOccurrence、contains、interator.remove 外，大部分的 ArrayDeque 都以恆定的開銷執行。 >注意：ArrayDeque 是 fail-fast 的，如果建立了迭代器之後，卻使用了迭代器外部的 remove 等修改方法，那麼這個類將會丟擲 ConcurrentModificationException 異常。 #### ConcurrentLinkedDeque `ConcurrentLinkedDeque` 是 JDK1.7 引入的雙向連結串列的無界併發佇列。它與 ConcurrentLinkedQueue 的區別是 ConcurrentLinkedDeque 同時支援 FIFO 和 FILO 兩種操作方式，即可以從佇列的頭和尾同時操作(插入/刪除)。ConcurrentLinkedDeque 也支援 `happen-before` 原則。ConcurrentLinkedDeque 不允許空元素。 #### LinkedBlockingDeque `LinkedBlockingDeque` 是一個由連結串列結構組成的雙向阻塞佇列，即可以從佇列的兩端插入和移除元素。雙向佇列因為多了一個操作佇列的入口，在多執行緒同時入隊時，也就減少了一半的競爭。LinkedBlockingDeque 把初始容量和建構函式繫結，這樣能夠有效過度拓展。初始容量如果沒有指定，就取的是 `Integer.MAX_VALUE`，這也是 LinkedBlockingDeque 的預設建構函式。 ### 同步工具類 同步工具類可以是任何一個物件，只要它根據自身狀態來協調執行緒的控制流。阻塞佇列可以作為同步控制類，其他型別的同步工具類還包括 `訊號量(Semaphore)`、`柵欄(Barrier)` 和 `閉鎖(Latch)`。下面我們就來一起認識一下這些工具類 ### Semaphore Semaphore 翻譯過來就是 `訊號量`，訊號量是什麼？它其實就是一種訊號，在作業系統中，也有訊號量的這個概念，在程序間通訊的時候，我們就會談到訊號量進行通訊。還有在 Linux 作業系統採取中斷時，也會向程序發出中斷訊號，根據程序的種類和訊號的型別判斷是否應該結束程序。 在 Java 中，`Semaphore（訊號量）`是用來控制同時訪問特定資源的執行緒數量，它通過協調各個執行緒，以保證合理的使用公共資源。 Semaphore 管理著一組`許可(permit)`，許可的初始數量由建構函式來指定。在獲取某個資源之前，應該先從訊號量獲取`許可(permit)`，以確保資源是否可用。當執行緒完成對資源的操作後，會把它放在池中並向訊號量返回一個許可，從而允許其他執行緒訪問資源，這叫做釋放許可。如果沒有許可的話，那麼 `acquire` 將會阻塞直到有許可（中斷或者操作超時）為止。`release` 方法將返回一個許可訊號量。 Semaphore 可以用來實現流量控制，例如常用的資料庫連線池，執行緒請求資源時，如果資料庫連線池為空則阻塞執行緒，直接返回失敗，如果連線池不為空時解除阻塞。 ### CountDownLatch `閉鎖(Latch)` 是一種同步工具類，它可以延遲執行緒的進度以直到其到達終止狀態。閉鎖的作用相當於是一扇門，在閉鎖達到結束狀態前，門是一直關著的，沒有任何執行緒能夠通過。當閉鎖到達結束狀態後，這扇門會開啟並且允許任何執行緒通過，然後就一直保持開啟狀態。 `CountDownLatch` 就是閉鎖的一種實現。它可以使一個或者多個執行緒等待一組事件的發生。閉鎖有一個計數器，閉鎖需要對計數器進行初始化，表示需要等待的次數，閉鎖在呼叫 `await` 處進行等待，其他執行緒在呼叫 countDown 把閉鎖 count 次數進行遞減，直到遞減為 0 ，喚醒 await。如下程式碼所示 ```java public class TCountDownLatch { public static void main(String[] args) { CountDownLatch latch = new CountDownLatch(5); Increment increment = new Increment(latch); Decrement decrement = new Decrement(latch); new Thread(increment).start(); new Thread(decrement).start(); try { Thread.sleep(6000); } catch (InterruptedException e) { e.printStackTrace(); } } } class Decrement implements Runnable { CountDownLatch countDownLatch; public Decrement(CountDownLatch countDownLatch){ this.countDownLatch = countDownLatch; } @Override public void run() { try { for(long i = countDownLatch.getCount();i > 0;i--){ Thread.sleep(1000); System.out.println("countdown"); this.countDownLatch.countDown(); } } catch (InterruptedException e) { e.printStackTrace(); } } } class Increment implements Runnable { CountDownLatch countDownLatch; public Increment(CountDownLatch countDownLatch){ this.countDownLatch = countDownLatch; } @Override public void run() { try { System.out.println("await"); countDownLatch.await(); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("Waiter Released"); } } ``` ### Future 我們常見的建立多執行緒的方式有兩種，一種是繼承 Thread 類，一種是實現 Runnable 介面。這兩種方式都沒有返回值。相對的，建立多執行緒還有其他三種方式，那就是使用 `Callable`介面、 `Future` 介面和 `FutureTask` 類。Callable 我們之前聊過，這裡就不再描述了，我們主要來描述一下 Future 和 FutureTask 介面。 Future 就是對具體的 Runnable 或者 Callable 任務的執行結果進行一系列的操作，必要時可通過 `get` 方法獲取執行結果，這個方法會阻塞直到執行結束。Future 中的主要方法有 ```java public interface Future { boolean cancel(boolean mayInterruptIfRunning); boolean isCancelled(); boolean isDone(); V get() throws InterruptedException, ExecutionException; V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException; } ``` * `cancel(boolean mayInterruptIfRunning)` : 嘗試取消任務的執行。如果任務已經完成、已經被取消或者由於某些原因而無法取消，那麼這個嘗試會失敗。如果取消成功，或者在呼叫 cancel 時此任務尚未開始，那麼此任務永遠不會執行。如果任務已經開始，那麼 mayInterruptIfRunning 引數會確定是否中斷執行任務以便於嘗試停止該任務。這個方法返回後，會對 `isDone` 的後續呼叫也返回 true，如果 cancel 返回 true，那麼後續的呼叫 `isCancelled ` 也會返回 true。 * `boolean isCancelled()`：如果此任務在正常完成之前被取消，則返回 true。 * `boolean isDone()`：如果任務完成，返回 true。 * `V get() throws InterruptedException, ExecutionException`：等待必要的計算完成，然後檢索其結果 * `V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException` ： 必要時最多等待給定時間以完成計算，然後檢索其結果。 因為Future只是一個介面，所以是無法直接用來建立物件使用的，因此就有了下面的FutureTask。 ### FutureTask FutureTask 實現了 `RunnableFuture` 介面，RunnableFuture 介面是什麼呢？ RunnableFuture 介面又繼承了 `Runnable` 介面和 `Future` 介面。納尼？在 Java 中不是隻允許單繼承麼，是的，單繼承更多的是說的類與類之間的繼承關係，子類繼承父類，擴充套件父類的介面，這個過程是單向的，就是為了解決多繼承引起的過渡引用問題。而介面之間的繼承是介面的擴充套件，在 Java 程式設計思想中也印證了這一點  對 RunnableFuture 介面的解釋是：成功執行的 run 方法會使 Future 介面的完成並允許訪問其結果。所以它既可以作為 Runnable 被執行緒執行，又可以作為 Future 得到 Callable 的返回值。 FutureTask 也可以用作閉鎖，它可以處於以下三種狀態 * 等待執行 * 正在執行 * 執行完成 FutureTask 在 `Executor` 框架中表示非同步任務，此外還可以表示一些時間較長的計算，這些計算可以在使用計算結果之前啟動。 FutureTask 具體的原始碼我後面會單獨出文章進行描述。 ### Barrier 我們上面聊到了通過閉鎖來啟動一組相關的操作，使用閉鎖來等待一組事件的執行。閉鎖是一種一次性物件，一旦進入終止狀態後，就不能被 `重置`。 `Barrier` 的特點和閉鎖也很類似，它也是阻塞一組執行緒直到某個事件發生。柵欄與閉鎖的區別在於，所有執行緒必須同時到達柵欄的位置，才能繼續執行，就像我們上面作業系統給出的這幅圖一樣。 ABCD 四條執行緒，必須同時到達 Barrier，然後 `手牽手`一起走過幸福的殿堂。 當執行緒到達 Barrier 的位置時會呼叫 `await` 方法，這個方法會阻塞直到所有執行緒都到達 Barrier 的位置，如果所有執行緒都到達 Barrier 的位置，那麼 Barrier 將會開啟使所有執行緒都被釋放，而 Barrier 將被重置以等待下次使用。如果呼叫 await 方法導致超時，或者 await 阻塞的執行緒被中斷，那麼 Barrier 就被認為被打破，所有阻塞的 await 都會丟擲 `BrokenBarrierException` 。如果成功通過柵欄後，await 方法返回一個唯一索引號，可以利用這些索引號選舉一個新的 leader，來處理一下其他工作。 ```java public class TCyclicBarrier { public static void main(String[] args) { Runnable runnable = () -> System.out.println("Barrier 1 開始..."); Runnable runnable2 = () -> System.out.println("Barrier 2 開始..."); CyclicBarrier barrier1 = new CyclicBarrier(2,runnable); CyclicBarrier barrier2 = new CyclicBarrier(2,runnable2); CyclicBarrierRunnable b1 = new CyclicBarrierRunnable(barrier1,barrier2); CyclicBarrierRunnable b2 = new CyclicBarrierRunnable(barrier1,barrier2); new Thread(b1).start(); new Thread(b2).start(); } } class CyclicBarrierRunnable implements Runnable { CyclicBarrier barrier1; CyclicBarrier barrier2; public CyclicBarrierRunnable(CyclicBarrier barrier1,CyclicBarrier barrier2){ this.barrier1 = barrier1; this.barrier2 = barrier2; } @Override public void run() { try { Thread.sleep(1000); System.out.println(Thread.currentThread().getName() + "等待 barrier1" ); barrier1.await(); Thread.sleep(1000); System.out.println(Thread.currentThread().getName() + "等待 barrier2" ); barrier2.await(); } catch (InterruptedException | BrokenBarrierException e) { e.printStackTrace(); } System.out.println(Thread.currentThread().getName() + " 做完了!"); } } ``` ### Exchanger 與 Barrier 相關聯的還有一個工具類就是 `Exchanger`， Exchanger 是一個用於執行緒間協作的工具類。Exchanger用於進行執行緒間的資料交換。 它提供一個同步點，在這個同步點兩個執行緒可以交換彼此的資料。這兩個執行緒通過exchange 方法交換資料， 如果第一個執行緒先執行 exchange方法，它會一直等待第二個執行緒也執行 exchange，當兩個執行緒都到達同步點時，這兩個執行緒就可以交換資料，將本執行緒生產出來的資料傳遞給對方。因此使用Exchanger 的重點是成對的執行緒使用 exchange() 方法，當有一對執行緒達到了同步點，就會進行交換資料。因此該工具類的執行緒物件是成對的。 下面通過一段例子程式碼來講解一下 ```java public class TExchanger { public static void main(String[] args) { Exchanger exchanger = new Exchanger(); ExchangerRunnable exchangerRunnable = new ExchangerRunnable(exchanger,"A"); ExchangerRunnable exchangerRunnable2 = new ExchangerRunnable(exchanger,"B"); new Thread(exchangerRunnable).start(); new Thread(exchangerRunnable2).start(); } } class ExchangerRunnable implements Runnable { Exchanger exchanger; Object object; public ExchangerRunnable(Exchanger exchanger,Object object){ this.exchanger = exchanger; this.object = object; } @Override public void run() { Object previous = object; try { object = this.exchanger.exchange(object); System.out.println( Thread.currentThread().getName() + "改變前是" + previous + "改變後是" + object); } catch (InterruptedException e) { e.printStackTrace(); } } } ``` ## 總結 本篇文章我們從同步容器類入手，主要講了 `fail-fast` 和 `fail-safe` 機制，這兩個機制在併發程式設計中非常重要。然後我們從作業系統的角度，聊了聊作業系統層面實現安全性的幾種方式，然後從作業系統 -> 併發我們聊了聊 Java 中的併發工具包有哪些，以及構建併發的幾種工具類。 你好，我是 cxuan，我自己手寫了四本 PDF，分別是 Java基礎總結、HTTP 核心總結、計算機基礎知識，作業系統核心總結，我已經整理成為 PDF，可以關注公眾號 Java建設者 回覆 PDF 領取優質資料。 ![](https://img2020.cnblogs.com/blog/1515111/202008/1515111-20200831071710331-16195436