原文作者：熱愛可抵漫長歲月原文鏈接：https://juejin.cn/post/7063655099465203742Java 容器在我們平常項目中使用實在是太頻繁了，比

原文作者：熱愛可抵漫長歲月
原文鏈接：https://juejin.cn/post/7063655099465203742

Java 容器在我們平常項目中使用實在是太頻繁了，比如 ArrayList、HashMap、HashSet 這三個，但有一點需要我們注意，這三個容器在并發環境下是線程不安全的。
所以我們在并發環境下如果要使用容器的話，java.util.concurrent 包下為我們提供了一些線程安全的容器，并且還能在保證線程安全的條件下性能也不差。如 ConcurrentHashMap、CopyOnWriteArrayList、CopyOnWriteArraySet 等。

1、HahMap 線程不安全

解決方法：n（1）使用 Hashtable 替代 HashMapn（2）Collections.synchronizedMap()n（3）使用 ConcurrentHashMap 代替 HashMap（推薦使用）

數據丟失問題

先看源碼：

假如兩個線程同時進入 if ((p = tab[i = (n - 1) & hash]) == null) 這一行，假設對應的位置為 null。

這也就意味著兩個線程可以同時進入 if 包含的代碼塊（線程上下文切換），恰好兩個 put 的位置是一樣的（如 key 相等或者哈希沖突，哈希沖突正常情況下應該用拉鏈法解決）。結果就是后一個線程的數據會覆蓋前一個線程的數據。

也就是數據丟失問題。

可見性問題

可見性：當一個線程操作這個容器的時候，該操作需要對另外的線程都可見，也就是其他線程都能感知到本次操作。

HashMap 是無法做到的。

死鏈問題

在 JDK 1.7 的時候是采用鏈表頭插法的，這種情況下擴容的時候會造成鏈表死循環。

JDK 1.7 源碼如下：

void transfer(Entry[] newTable, boolean rehash) {n int newCapacity = newTable.length;n for (Entry<K,V> e : table) {n while(null != e) {n Entry<K,V> next = e.next;n if (rehash) {n e.hash = null == e.key ? 0 : hash(e.key);n }n int i = indexFor(e.hash, newCapacity);n e.next = newTable[i];n newTable[i] = e;n e = next;n }n }n}

• e 和 next 都是局部變量，用來指向當前節點和下一個節點
• 線程 1（綠色）的臨時變量 e 和 next 剛引用了這倆節點，還未來得及移動節點，發生了線程切換，由線程 2（藍色）完成擴容和遷移

線程 2 擴容完成，由于頭插法，鏈表順序顛倒。但線程 1 的臨時變量 e 和 next 還引用了這倆節點，還要再來一遍遷移

第一次循環

• 循環接著線程切換前運行，注意此時 e 指向的是節點 a，next 指向的是節點 b
• e 頭插 a 節點，注意圖中畫了兩份 a 節點，但事實上只有一個（為了不讓箭頭特別亂畫了兩份）
• 當循環結束是 e 會指向 next 也就是 b 節點

第二次循環

• next 指向了節點 a
• e 頭插節點 b
• 當循環結束時，e 指向 next 也就是節點 a

第三次循環

• next 指向了 null
• e 頭插節點 a，a 的 next 指向了 b（之前 a.next 一直是 null），b 的 next 指向 a，死鏈已成
• 當循環結束時，e 指向 next 也就是 null，因此第四次循環時會正常退出

2、ConcurrentHashMap 分析

面試題：ConcurrentHashMap 在 JDK 1.7 和 JDK 1.8 中的結構分別是什么？它們有什么相同點和不同點？

JDK 1.7

整體結構圖：

可以看出，在 ConcurrentHashMap 內部進行了 Segment 分段，Segment 繼承了 ReentrantLock，可以理解為一把鎖，各個 Segment 之間都是相互獨立上鎖的，互不影響。

相比于 Hashtable 每次操作都需要把整個對象鎖住而言，大大提高了并發效率。因為它的鎖與鎖之間是獨立的，而不是整個對象只有一把鎖。

每個 Segment 的底層數據結構與 HashMap 類似，仍然是數組和鏈表組成的拉鏈法結構。默認有 16 個 Segment，所以最多可以同時支持 16 個線程并發操作（操作分別分布在不同的 Segment 上）。16 這個默認值可以在初始化的時候設置為其他值，但是一旦確認初始化以后，是不可以擴容的。

JDK 1.8

整體結構圖：

圖中的節點有三種類型：

• 第一種是最簡單的，空著的位置代表當前還沒有元素來填充。
• 第二種就是和 HashMap 非常類似的拉鏈法結構，在每一個槽中會首先填入第一個節點，但是后續如果計算出相同的 Hash 值，就用鏈表的形式往后進行延伸。
• 第三種結構就是紅黑樹結構，這是 JDK 1.7 的 ConcurrentHashMap 中所沒有的結構，了解過 HashMap 的話應該知道這個。

當鏈表長度大于 8，數組長度 >= 64 時，鏈表會轉化為紅黑樹。

紅黑樹特點：

紅黑樹是每個節點都帶有顏色屬性的二叉查找樹，顏色為紅色或黑色，紅黑樹的本質是對二叉查找樹 BST 的一種平衡策略，我們可以理解為是一種平衡二叉查找樹，查找效率高，會自動平衡，防止極端不平衡從而影響查找效率的情況發生。

查找時間復雜度為：O(logN)，因為如果鏈表在某些極端情況下太長的話，查找時間復雜度為 O(N)，這也就是引入紅黑樹的原因，主要是為了性能。

節點顏色規定：

• 每個節點要么是紅色，要么是黑色，但根節點永遠是黑色的。
• 紅色節點不能連續，也就是說，紅色節點的子和父都不能是紅色的。
• 從任一節點到其每個葉子節點的路徑都包含相同數量的黑色節點。

JDK 1.8 源碼分析

Node 節點

（1）ConcurrentHashMap 的 Node 節點源碼：

static class Node<K,V> implements Map.Entry<K,V> {n final int hash;n final K key;n volatile V val;n volatile Node<K,V> next;n}

（2）HashMap 的 Node 節點源碼：

static class Node<K,V> implements Map.Entry<K,V> {n final int hash;n final K key;n V value;n Node<K,V> next;n}

很明顯，在 value 和 next 屬性上加了 volatile 關鍵字。為了保證可見性。

put 方法

源碼如下：

// 實際上是調用了 putVal 方法npublic V put(K key, V value) {n return putVal(key, value, false);n}nn// 真正的添加元素邏輯實現nfinal V putVal(K key, V value, boolean onlyIfAbsent) {n if (key == null || value == null) throw new NullPointerException();n // 計算 hash 值，方法源碼寫在下面了，和 HashMap 有點不一樣n int hash = spread(key.hashCode());n int binCount = 0;n for (Node<K,V>[] tab = table;;) {n Node<K,V> f; int n, i, fh;n // 哈希桶為空，進行相應的初始化過程n if (tab == null || (n = tab.length) == 0)n tab = initTable();n // 找到哈希值對應的哈希桶下標n else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {n // 以 CAS 的方式放入新的值，方法源碼寫在下面了n if (casTabAt(tab, i, null,n new Node<K,V>(hash, key, value, null)))n break; // no lock when adding to empty binn }n // hash 值等于 MOVED，代表在擴容n else if ((fh = f.hash) == MOVED)n tab = helpTransfer(tab, f);n // 對應哈希桶下標有值的情況n else {n V oldVal = null;n // 用 synchronized 鎖住當前位置，保證并發安全，synchronized 在 JDK 1.6 后做了優化n synchronized (f) {n if (tabAt(tab, i) == f) {n // 如果是鏈表n if (fh >= 0) {n binCount = 1;n // 遍歷鏈表n for (Node<K,V> e = f;; ++binCount) {n K ek;n if (e.hash == hash &&n ((ek = e.key) == key ||n (ek != null && key.equals(ek)))) {n oldVal = e.val;n if (!onlyIfAbsent)n e.val = value;n break;n }n Node<K,V> pred = e;n // 遍歷了整個鏈表也沒有發現相同的值，說明之前不存在該值，添加到鏈表尾部n if ((e = e.next) == null) {n pred.next = new Node<K,V>(hash, key,n value, null);n break;n }n }n }n // 如果是紅黑樹n else if (f instanceof TreeBin) {n Node<K,V> p;n binCount = 2;n // 調用 putTreeVal 方法往紅黑樹里增加數據n if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,n value)) != null) {n oldVal = p.val;n if (!onlyIfAbsent)n p.val = value;n }n }n }n }n if (binCount != 0) {n // 檢查是否滿足條件并把鏈表轉換為紅黑樹的形式，默認的 TREEIFY_THRESHOLD 閾值是 8n if (binCount >= TREEIFY_THRESHOLD)n treeifyBin(tab, i);n // putVal 的返回是添加前的舊值，所以返回 oldValn if (oldVal != null)n return oldVal;n break;n }n }n }n addCount(1L, binCount);n return null;n}nnstatic final int spread(int h) {n return (h ^ (h >>> 16)) & HASH_BITS;n}nnstatic final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,n Node<K,V> c, Node<K,V> v) {n return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);n}

get 方法

public V get(Object key) {n Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;n int h = spread(key.hashCode());n if ((tab = table) != null && (n = tab.length) > 0 &&n (e = tabAt(tab, (n - 1) & h)) != null) {n if ((eh = e.hash) == h) {n if ((ek = e.key) == key || (ek != null && key.equals(ek)))n return e.val;n }n else if (eh < 0)n return (p = e.find(h, key)) != null ? p.val : null;n while ((e = e.next) != null) {n if (e.hash == h &&n ((ek = e.key) == key || (ek != null && key.equals(ek))))n return e.val;n }n }n return null;n}

JDK 1.7 和 JDK 1.8 對比

數據結構

JDK 1.7：Segment 分段鎖

JDK 1.8：數組 + 鏈表 + 紅黑樹

并發度

JDK 1.7：每個 Segment 獨立加鎖，最大并發個數就是 Segment 的個數，默認是 16。

JDK 1.8：鎖粒度更細，理想情況下 table 數組元素的個數（也就是數組長度）就是其支持并發的最大個數，并發度比之前有提高。

保證并發安全

JDK 1.7：采用 Segment 分段鎖來保證安全，而 Segment 是繼承自 ReentrantLock。

JDK 1.8：采用 Node + CAS + synchronized 保證線程安全。

遇到 Hash 碰撞

JDK 1.7：在 Hash 沖突時，會使用拉鏈法，也就是鏈表的形式。

JDK 1.8：先使用拉鏈法，當鏈表長度大于 8，數組長度 >= 64 時，鏈表會轉化為紅黑樹，來提高查找效率。

查詢時間復雜度

JDK 1.7：遍歷鏈表的時間復雜度是 O(N)，n 為鏈表長度。

JDK 1.8：如果變成遍歷紅黑樹，那么時間復雜度降低為 O(logN)，n 為樹的節點個數。

與 Hashtable 的區別

實現線程安全的方式不同

Hashtable 是使用 synchronized 來保證線程安全的。

ConcurrentHashMap 是使用 CAS + synchronized + volatile 來保證線程安全的。

性能不同

Hashtable 把整個對象給鎖了，而 ConcurrentHashMap 只鎖單個節點，所以并發性能比 Hashtable 好。

迭代時修改的不同

Hashtable（包括 HashMap）不允許在迭代期間修改內容，否則會拋出 ConcurrentModificationException 異常，其原理是檢測 modCount 變量，Hashtable 的 next() 方法源碼如下：

public T next() {n if (modCount != expectedModCount)n throw new ConcurrentModificationException();n return nextElement();n}

ConcurrentHashMap 即便在迭代期間修改內容，也不會拋出 ConcurrentModificationException。

3、CopyOnWriteArrayList 分析

從 JDK 1.5 開始，Java 并發包里提供了使用 CopyOnWrite 機制實現的并發容器 CopyOnWriteArrayList 作為主要的并發 List，CopyOnWrite 的并發集合還包括 CopyOnWriteArraySet，其底層正是利用 CopyOnWriteArrayList 實現的。
核心就是讀寫分離。

適用場景

• 讀操作可以盡可能的快，而寫即使慢一些也沒關系

比如，有些系統級別的信息，往往只需要加載或者修改很少的次數，但是會被系統內所有模塊頻繁的訪問。

• 讀多寫少

讀寫規則

• 讀寫鎖的規則

讀寫鎖的思想是：讀讀共享、其他都互斥（寫寫互斥、讀寫互斥、寫讀互斥）。

• 對讀寫鎖規則的升級

CopyOnWriteArrayList 的思想比讀寫鎖的思想又更進一步。為了將讀取的性能發揮到極致，CopyOnWriteArrayList 讀取是完全不用加鎖的，更厲害的是，寫入也不會阻塞讀取操作，也就是說你可以在寫入的同時進行讀取，只有寫入和寫入之間需要進行同步，也就是不允許多個寫入同時發生，但是在寫入發生時允許讀取同時發生。這樣一來，讀操作的性能就會大幅度提升。

特點

• CopyOnWrite的含義

CopyOnWrite 的意思是說，當容器需要被修改的時候，不直接修改當前容器，而是先將當前容器進行 Copy，復制出一個新的容器，然后修改新的容器，完成修改之后，再將原容器的引用指向新的容器。這樣就完成了整個修改過程。

CopyOnWriteArrayList 利用了不變性原理，因為容器每次修改都是創建新副本，所以對于舊容器來說，其實是不可變的，也是線程安全的，無需進一步的同步操作。所以可以對 CopyOnWrite 容器進行并發的讀，而不需要加鎖，因為當前容器不會添加任何元素，也不會有修改。

核心思想就是讀寫分離。

• 迭代期間允許修改集合內容

ArrayList 在迭代期間如果修改集合的內容，會拋出 ConcurrentModificationException 異常。根據 modCount 的值是否改變來判斷。

CopyOnWriteArrayList 的迭代器一旦被建立之后，如果往之前的 CopyOnWriteArrayList 對象中去新增元素，在迭代器中既不會顯示出元素的變更情況，同時也不會報錯。

缺點

• 內存占用問題

因為 CopyOnWrite 的寫時復制機制，所以在進行寫操作的時候，內存里會同時駐扎兩個對象的內存，這一點會占用額外的內存空間。

• 在元素較多或者復雜的情況下，復制的開銷很大

復制過程不僅會占用雙倍內存，還需要消耗 CPU 等資源，會降低整體性能。

• 數據一致性問題

由于 CopyOnWrite 容器的修改是先修改副本，所以這次修改對于其他線程來說，并不是實時能看到的，只有在修改完之后才能體現出來。如果你希望寫入的的數據馬上能被其他線程看到，CopyOnWrite 容器是不適用的。

源碼分析

數據結構

// 非公平鎖nfinal transient ReentrantLock lock = new ReentrantLock();n// 底層就是一個數組，加了 volatile 關鍵字來保證可見性nprivate transient volatile Object[] array;nn// 獲取數組nfinal Object[] getArray() {n return array;n}nn// 設置數組nfinal void setArray(Object[] a) {n array = a;n}nn// 空構造方法npublic CopyOnWriteArrayList() {n setArray(new Object[0]);n}

add 方法

public boolean add(E e) {n final ReentrantLock lock = this.lock;n lock.lock();n try {n // 獲取原數組n Object[] elements = getArray();n // 獲取原數組長度n int len = elements.length;n // 復制一個新數組n Object[] newElements = Arrays.copyOf(elements, len + 1);n // 將新元素添加到新數組n newElements[len] = e;n // 設置新數組為 array 的引用n setArray(newElements);n return true;n } finally {n lock.unlock();n }n}

CopyOnWrite 的思想：寫操作是在原來容器的拷貝上進行的，并且在讀取數據的時候不會鎖住 list。而且可以看到，如果對容器拷貝操作·的過程中有新的讀線程進來，那么讀到的還是舊的數據，因為在那個時候對象的引用還沒有被更改。

get 方法

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

可以發現，get 操作并沒有上鎖。

迭代器 COWIterator 類

private COWIterator(Object[] elements, int initialCursor) {n // 迭代器的游標n cursor = initialCursor;n // 數組的快照n snapshot = elements;n}

之后的迭代器所有的操作都基于 snapshot 數組進行的，比如：

public E next() {n if (! hasNext())n throw new NoSuchElementException();n return (E) snapshot[cursor++];n}