Java并發(fā)容器那么多,應(yīng)該怎么選?

“?寫給自己看，說給別人聽。你好，這是think123的第86篇原創(chuàng)文章”

我們先來看看有哪些并發(fā)容器

這么多容器，我們該怎么選? 雖然不能全要，但是我們可以都了解一下，然后挑選適合自己的。

并發(fā)下的Map

我們都知道不能再并發(fā)場景下使用HashMap，因為在JDK7之前，在并發(fā)場景下使用 HashMap 會出現(xiàn)死循環(huán)，從而導(dǎo)致 CPU 使用率居高不下，而擴(kuò)容是導(dǎo)致死循環(huán)的主要原因。

雖然 Java 在 JDK8 中修復(fù)了 HashMap 擴(kuò)容導(dǎo)致的死循環(huán)問題，但在高并發(fā)場景下，依然會有數(shù)據(jù)丟失以及不準(zhǔn)確的情況出現(xiàn)。

HashMap擴(kuò)容出現(xiàn)死循環(huán)的分析可以查看我之前的文章

為了保證線程安全，安全的Map對象 Hashtable、ConcurrentHashMap 以及ConcurrentSkipListMap 三個容器

Hashtable 和 ConcurrentHashMap 都是基于 HashMap 實現(xiàn)的，對小數(shù)據(jù)量的存取比較有優(yōu)勢。

ConcurrentSkipListMap 是基于跳表實現(xiàn)的,其特點是存取平均時間復(fù)雜度是 O(log(n))，適用于大數(shù)據(jù)量存取的場景。

Hashtable VS ConcurrentHashMap

Hashtable在它所有的獲取或者修改數(shù)據(jù)的方法上都添加了Synchronized，因此在高并發(fā)場景下，讀寫操作都會存在大量鎖競爭，給系統(tǒng)帶來性能開銷。

ConcurrentHashMap 在保證線程安全的基礎(chǔ)上兼具了更好的并發(fā)性能。

在 JDK7 中，ConcurrentHashMap 就使用了分段鎖 Segment 減小了鎖粒度，最終優(yōu)化了鎖的并發(fā)操作。?

到了 JDK8，ConcurrentHashMap 做了大量的改動，摒棄了 Segment 的概念。由于Synchronized 鎖在 Java6 之后的性能已經(jīng)得到了很大的提升，所以在 JDK1.8 中，Java重新啟用了 Synchronized 同步鎖，通過 Synchronized 實現(xiàn) HashEntry 作為鎖粒度。

這種改動將數(shù)據(jù)結(jié)構(gòu)變得更加簡單了，操作也更加清晰流暢。

與 JDK7 的 put 方法一樣，JDK8 在添加元素時，在沒有哈希沖突的情況下，會使用CAS 進(jìn)行添加元素操作；如果有沖突，則通過 Synchronized 將鏈表鎖定，再執(zhí)行接下來的操作。

雖然 ConcurrentHashMap 的整體性能要優(yōu)于 Hashtable，但在某些場景中，ConcurrentHashMap 依然不能代替 Hashtable。?

例如，在強一致的場景中ConcurrentHashMap 就不適用，原因是 ConcurrentHashMap 中的 get、size 等方法沒有用到鎖，因此返回的數(shù)據(jù)就不準(zhǔn)確, ConcurrentHashMap 是弱一致性的，因此有可能會導(dǎo)致某次讀無法馬上獲取到寫入的數(shù)據(jù)

ConcurrentHashMap VS ConcurrentSkipListMap

我們都知道 ConcurrentHashMap 數(shù)據(jù)量比較大的時候，鏈表會轉(zhuǎn)換為紅黑樹。紅黑樹在并發(fā)情況下，刪除和插入過程中有個平衡的過程，會牽涉到大量節(jié)點，因此競爭鎖資源的代價相對比較高。

而ConcurrentSkipListMap由于是基于跳表實現(xiàn)的，它需要鎖住的節(jié)點要少一些，在高并發(fā)場景下性能也要好一些。

跳躍表是基于鏈表擴(kuò)展實現(xiàn)的一種特殊鏈表，類似于樹的實現(xiàn)，跳躍表不僅實現(xiàn)了橫向鏈表，還實現(xiàn)了垂直方向的分層索引。?

一個跳躍表由若干層鏈表組成，每一層都實現(xiàn)了一個有序鏈表索引，只有最底層包含了所有數(shù)據(jù)，每一層由下往上依次通過一個指針指向上層相同值的元素，每層數(shù)據(jù)依次減少，等到了最頂層就只會保留部分?jǐn)?shù)據(jù)了。

跳躍表的這種結(jié)構(gòu)，是利用了空間換時間的方法來提高了查詢效率。程序總是從最頂層開始查詢訪問，通過判斷元素值來縮小查詢范圍。

java的實現(xiàn)中，往跳表中插入數(shù)據(jù)的時候,會根據(jù)概率隨機(jī)算出level值，然后重建索引層，這里的代碼還是比較有趣的

//?產(chǎn)生一個隨機(jī)數(shù)
int?rnd?=?ThreadLocalRandom.nextSecondarySeed();

//?0x80000001二進(jìn)制為是?10000000000000000000000000000001?
//?只有隨機(jī)數(shù)是正偶數(shù)才能決定是否要新增層級
if?((rnd?&?0x80000001)?==?0)?{
??int?level?=?1,?max;
??//?從隨機(jī)數(shù)第二位開始,有幾個連續(xù)的1，level值就加幾
??while?(((rnd?>>>=?1)?&?1)?!=?0)
??????++level;

??Index?idx?=?null;
??HeadIndex?h?=?head;
??//?層數(shù)小于等于當(dāng)前跳表的最大層
??if?(level?<=?(max?=?h.level))?{

??????//?從第一層開始建立down的索引鏈表
??????for?(int?i?=?1;?i?<=?level;?++i)
??????????idx?=?new?Index(z,?idx,?null);
??}
??else?{?
????//?超出了現(xiàn)有跳表的層數(shù)，則只加一層，多了沒有意義
????level?=?max?+?1;
????...
??}
??//?省略重建索引層代碼
}

當(dāng)新增一個 key 值為 7 的節(jié)點時，首先新增一個節(jié)點到最底層的鏈表中，根據(jù)概率算出level 值，再根據(jù) level 值新建索引層，最后鏈接索引層的新節(jié)點。

新增節(jié)點和鏈接索引都是基于 CAS 操作實現(xiàn)

當(dāng)刪除某個key 時，首先找到待刪除結(jié)點，將其 value 值通過cas設(shè)置為 null；之后再向待刪除結(jié)點的 next 位置新增一個marker(標(biāo)記)結(jié)點，以便減少并發(fā)沖突。

然后讓待刪結(jié)點的前驅(qū)節(jié)點直接越過本身指向的待刪結(jié)點，直接指向后繼結(jié)點，中間要被刪除的結(jié)點最終將會被JVM 垃圾回收處理掉；最后判斷此次刪除后是否導(dǎo)致某一索引層沒有其它節(jié)點了，如果這一層都沒有節(jié)點了則跳表層數(shù)降級。

經(jīng)過上面的分析，我們可以知道當(dāng)我們不需要知道集合中準(zhǔn)確數(shù)據(jù)的時候使用ConcurrentHashMap，當(dāng)我們需要知道集合中準(zhǔn)確數(shù)據(jù)個數(shù)時，則需要用到HashTable。

如果數(shù)據(jù)量特別大，且存在大量增刪改查操作，則可以考慮使用ConcurrentSkipListMap。

同時需要注意的是這三個線程安全的容器類key,value都不允許為null,而我們常使用的HashMap的key是可以為null,value不能為null的。

并發(fā)下的List

我們都知道Vector也是在它所有暴露出來的public方法中都加上了synchronized 關(guān)鍵字來保證線程安全，所以在讀遠(yuǎn)大于寫的操作場景中，Vector 將會發(fā)生大量鎖競爭， 從而給系統(tǒng)帶來性能開銷。

而CopyOnWriteList實現(xiàn)了讀操作無鎖，寫操作則通過操作底層數(shù)組的新副本來實現(xiàn)，是一種讀寫分離的并發(fā)策略，我們來看下源碼是如何實現(xiàn)的

private?transient?volatile?Object[]?array;

public?E?get(int?index)?{
??return?get(getArray(),?index);
}
private?E?get(Object[]?a,?int?index)?{
????return?(E)?a[index];
}

可以看到讀操作平平無奇，就是從數(shù)組中讀取元素。

再看看寫操作

public?E?set(int?index,?E?element)?{
??final?ReentrantLock?lock?=?this.lock;
??//?獨占鎖加鎖
??lock.lock();
??try?{
????//?獲取存儲數(shù)據(jù)原始數(shù)組
????Object[]?elements?=?getArray();
????E?oldValue?=?get(elements,?index);

????//?設(shè)置新的數(shù)據(jù)
????if?(oldValue?!=?element)?{
??????int?len?=?elements.length;

??????//?將舊數(shù)據(jù)復(fù)制到一個新的數(shù)組
??????Object[]?newElements?=?Arrays.copyOf(elements,?len);

??????//?將數(shù)據(jù)插入到新的數(shù)組中，并將新的數(shù)組設(shè)置為存儲數(shù)據(jù)的數(shù)組
??????newElements[index]?=?element;
??????setArray(newElements);
????}?else?{
??????//?重新設(shè)置數(shù)據(jù)，保證volatile寫語義
??????setArray(elements);
????}
????return?oldValue;
??}?finally?{
????lock.unlock();
??}
}

從源碼中我們知道了 CopyOnWriteList 就是當(dāng)需要寫入數(shù)據(jù)的時候，復(fù)制一個新的數(shù)組，將數(shù)據(jù)寫入到新的數(shù)組中，再將新的數(shù)組賦值給舊數(shù)組。

所以 CopyOnWriteList 更適用于讀遠(yuǎn)大于寫的操作，同時業(yè)務(wù)場景對寫入數(shù)據(jù)的實時獲取并沒有要求,只需要保證最終能獲取到寫入數(shù)組中的數(shù)據(jù)就可以了。

并發(fā)下的Set

CopyOnWriteArraySet是使用CopyOnWriteArrayList實現(xiàn)的，而 ConcurrentSkipListSet 又是使用 ConcurrentSkipListMap實現(xiàn)的，而這兩者區(qū)別可以分別參考它們的實現(xiàn)。這里就不重復(fù)說明了。

并發(fā)下的Queue

Java并發(fā)包中Queue這類并發(fā)容器是最復(fù)雜的，并不是說實現(xiàn)多復(fù)雜，而是它的類比較多，記起來復(fù)雜。不過它仍然可以從兩個維度來分類。

一個維度是阻塞與非阻塞，所謂阻塞指的是當(dāng)隊列已滿時，入隊操作阻塞；當(dāng)隊列已空時，出隊操作阻塞。

另一個維度是單端與雙端，單端指的是只能隊尾入隊，隊首出隊；而雙端指的是隊首隊尾皆可入隊出隊。

Java 并發(fā)包里阻塞隊列都用 Blocking 關(guān)鍵字標(biāo)識，單端隊列使用 Queue 標(biāo)識，雙端隊列使用 Deque 標(biāo)識。

兩個維度可以雙雙組合，將 Queue 細(xì)分為4大類。

單端阻塞隊列

單端阻塞隊列內(nèi)部一般會持有一個隊列，這個隊列可能是數(shù)組(ArrayBlockingQueue)也可以是鏈表(LinkedBlockingQueue)。?

甚至還可以不持有隊列（其實現(xiàn)是 SynchronousQueue），此時生產(chǎn)者線程的入隊操作必須等待消費者線程的出隊操作。

需要注意的是需要調(diào)用put/take方法隊列才會阻塞，add,offer方法是不會阻塞的。

ArrayBlockingQueue 首先使用的是數(shù)組來實現(xiàn)隊列，然后阻塞功能則是借助 ReentrantLock 和 Condition 來實現(xiàn)的。

LinkedBlockingQueue 則是借助的鏈表來實現(xiàn)隊列, 阻塞功能也是借助 ReentrantLock 和 Condition 來實現(xiàn)的,和 ArrayBlockingQueue 還有一個不同之處在于 LinkedBlockingQueue 中 put 和 take 分別使用了不同的 ReentrantLock, 這樣做的原因是 put的時候會將元素添加到隊列尾，take的時候則是從隊列頭移除。

SynchronousQueue 也作為阻塞隊列在線程池中出現(xiàn)過，它用于在兩個線程之間直接移交元素，一個線程生產(chǎn)了數(shù)據(jù)，另一個線程才能消費數(shù)據(jù)。

雖然它內(nèi)部沒有緩沖，但是如果同一個模式(生產(chǎn)或者消費)的線程過多，它們都會被存儲起來，然后被阻塞，所以它更適合生產(chǎn)和消費速率相當(dāng)?shù)那闆r下使用。

PriorityBlockingQueue 支持按照優(yōu)先級出隊,其底層實現(xiàn)就是 PriorityQueue + ReentrantLock ,關(guān)于PriorityQueue可以參考我的這篇 優(yōu)先級隊列源碼解析

DelayQueue 是一個支持延時獲取元素的阻塞隊列，其使用 ?PriorityQueue(存儲元素) + ReentrantLock 實現(xiàn)，同時元素必須實現(xiàn) Delayed 接口。在創(chuàng)建元素時可以指定多久才可以從隊列中獲取當(dāng)前元素，只有在延遲期滿時才能從隊列中提取元素。我們可以使用它實現(xiàn)定時調(diào)度和緩存系統(tǒng)。

LinkedTransferQueue 融合 LinkedBlockingQueue 和 SynchronousQueue 的功能，性能比 LinkedBlockingQueue 更好,它底層數(shù)據(jù)結(jié)構(gòu)采用的是雙隊列(dual queue)。

消費者線程取元素時，如果隊列為空，那就生成一個節(jié)點(節(jié)點元素為null)入隊，然后消費者線程park住，后面生產(chǎn)者線程入隊時發(fā)現(xiàn)有一個元素為null的節(jié)點，生產(chǎn)者線程就不入隊了，直接就將元素填充到該節(jié)點，喚醒該節(jié)點上park住線程，被喚醒的消費者線程就能獲取到該節(jié)點上的數(shù)據(jù)了。

如果是生產(chǎn)者插入元素時，如果隊列為空，就生成一個數(shù)據(jù)節(jié)點入隊，然后立即返回，并不會阻塞。

雙端阻塞隊列

雙端阻塞隊列的實現(xiàn)是 LinkedBlockingDeque, 它和 LinkedBlockingQueue 的實現(xiàn)類似，只是一個這個是雙端。

單端非阻塞隊列

實現(xiàn)是 ConcurrentLinkedQueue, 從隊尾插入數(shù)據(jù)，從隊頭取出數(shù)據(jù)。均是通過CAS來更新head, tail的指向，所以不會阻塞線程，但是如果并發(fā)太激烈會導(dǎo)致CPU空轉(zhuǎn)。

雙端非阻塞隊列

實現(xiàn)是 ConcurrentLinkedDeque。和 ConcurrentLinkedQueue 實現(xiàn)類似，只是它是雙端隊列。

通過上面的分析還可以知道，阻塞就用到了鎖，會阻塞線程。

非阻塞使用的是CAS的方式，不會阻塞線程，但是在并發(fā)很大的情況下，會使得CPU使用率過高。

使用隊列時，需要格外注意隊列是否支持有界(所謂有界指的是內(nèi)部的隊列是否有容量限制)。實際工作中，一般都不建議使用無界的隊列，因為數(shù)據(jù)量大了之后很容易導(dǎo)致OOM。

上面提到的這些 Queue 中，只有 ArrayBlockingQueue 和 LinkedBlockingQueue 是支持有界的，所以在使用其他無界隊列時，一定要充分考慮是否存在導(dǎo)致 OOM 的隱患。

寫到最后

分析了這些容器的特點以及實現(xiàn)，我相信大家一定有自己的選擇了。