ConcurrentHashMap原理深入分析

併發程式設計實踐中，ConcurrentHashMap是一個經常被使用的資料結構，相比於Hashtable以及Collections.synchronizedMap()，ConcurrentHashMap線上程安全的基礎上提供了更好的寫併發能力，但同時降低了對讀一致性的要求（這點好像CAP理論啊 O(∩_∩)O）。ConcurrentHashMap的設計與實現非常精巧，大量的利用了volatile，final，CAS等lock-free技術來減少鎖競爭對於效能的影響，無論對於Java併發程式設計的學習還是Java記憶體模型的理解，ConcurrentHashMap的設計以及原始碼都值得非常仔細的閱讀與揣摩。

這篇日誌記錄了自己對ConcurrentHashMap的一些總結，由於JDK6，7，8中實現都不同，需要分開闡述在不同版本中的ConcurrentHashMap。

之前已經在ConcurrentHashMap原理分析中解釋了ConcurrentHashMap的原理，主要是從程式碼的角度來闡述是原始碼是如何寫的，本文仍然從原始碼出發，挑選個人覺得重要的點（會用紅色標註）再次進行回顧，以及闡述ConcurrentHashMap的一些注意點。

1. JDK6與JDK7中的實現

1.1 設計思路

ConcurrentHashMap採用了分段鎖的設計，只有在同一個分段內才存在競態關係，不同的分段鎖之間沒有鎖競爭。相比於對整個Map加鎖的設計，分段鎖大大的提高了高併發環境下的處理能力。但同時，由於不是對整個Map加鎖，導致一些需要掃描整個Map的方法（如size(), containsValue()）需要使用特殊的實現，另外一些方法（如clear()）甚至放棄了對一致性的要求（ConcurrentHashMap是弱一致性的，具體請檢視

ConcurrentHashMap能完全替代HashTable嗎？）。

ConcurrentHashMap中的分段鎖稱為Segment，它即類似於HashMap（JDK7與JDK8中HashMap的實現）的結構，即內部擁有一個Entry陣列，陣列中的每個元素又是一個連結串列；同時又是一個ReentrantLock（Segment繼承了ReentrantLock）。ConcurrentHashMap中的HashEntry相對於HashMap中的Entry有一定的差異性：HashEntry中的value以及next都被volatile修飾，這樣在多執行緒讀寫過程中能夠保持它們的可見性，程式碼如下：

12345 static final class HashEntry<K,V> {final int hash;final K key;volatile V value;volatile HashEntry<K,V> next;

1.2 併發度（Concurrency Level）

併發度可以理解為程式執行時能夠同時更新ConccurentHashMap且不產生鎖競爭的最大執行緒數，實際上就是ConcurrentHashMap中的分段鎖個數，即Segment[]的陣列長度。ConcurrentHashMap預設的併發度為16，但使用者也可以在建構函式中設定併發度。當用戶設定併發度時，ConcurrentHashMap會使用大於等於該值的最小2冪指數作為實際併發度（假如使用者設定併發度為17，實際併發度則為32）。執行時通過將key的高n位（n = 32 – segmentShift）和併發度減1（segmentMask）做位與運算定位到所在的Segment。segmentShift與segmentMask都是在構造過程中根據concurrency level被相應的計算出來。

如果併發度設定的過小，會帶來嚴重的鎖競爭問題；如果併發度設定的過大，原本位於同一個Segment內的訪問會擴散到不同的Segment中，CPU cache命中率會下降，從而引起程式效能下降。（文件的說法是根據你併發的執行緒數量決定，太多會導效能降低）

1.3 建立分段鎖

和JDK6不同，JDK7中除了第一個Segment之外，剩餘的Segments採用的是延遲初始化的機制：每次put之前都需要檢查key對應的Segment是否為null，如果是則呼叫ensureSegment()以確保對應的Segment被建立。

ensureSegment可能在併發環境下被呼叫，但與想象中不同，ensureSegment並未使用鎖來控制競爭，而是使用了Unsafe物件的getObjectVolatile()提供的原子讀語義結合CAS來確保Segment建立的原子性。程式碼段如下：

123456789 if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))== null) { // recheckSegment<K,V> s = new Segment<K,V>(lf, threshold, tab);while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))== null) {if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))break;}}

1.4 put/putIfAbsent/putAll

和JDK6一樣，ConcurrentHashMap的put方法被代理到了對應的Segment（定位Segment的原理之前已經描述過）中。與JDK6不同的是，JDK7版本的ConcurrentHashMap在獲得Segment鎖的過程中，做了一定的優化 - 在真正申請鎖之前，put方法會通過tryLock()方法嘗試獲得鎖，在嘗試獲得鎖的過程中會對對應hashcode的連結串列進行遍歷，如果遍歷完畢仍然找不到與key相同的HashEntry節點，則為後續的put操作提前建立一個HashEntry。當tryLock一定次數後仍無法獲得鎖，則通過lock申請鎖。

需要注意的是，由於在併發環境下，其他執行緒的put，rehash或者remove操作可能會導致連結串列頭結點的變化，因此在過程中需要進行檢查，如果頭結點發生變化則重新對錶進行遍歷。而如果其他執行緒引起了連結串列中的某個節點被刪除，即使該變化因為是非原子寫操作（刪除節點後連結後續節點呼叫的是Unsafe.putOrderedObject()，該方法不提供原子寫語義）可能導致當前執行緒無法觀察到，但因為不影響遍歷的正確性所以忽略不計。

之所以在獲取鎖的過程中對整個連結串列進行遍歷，主要目的是希望遍歷的連結串列被CPU cache所快取，為後續實際put過程中的連結串列遍歷操作提升效能。

在獲得鎖之後，Segment對連結串列進行遍歷，如果某個HashEntry節點具有相同的key，則更新該HashEntry的value值，否則新建一個HashEntry節點，將它設定為連結串列的新head節點並將原頭節點設為新head的下一個節點。新建過程中如果節點總數（含新建的HashEntry）超過threshold，則呼叫rehash()方法對Segment進行擴容，最後將新建HashEntry寫入到陣列中。

put方法中，連結新節點的下一個節點（HashEntry.setNext()）以及將連結串列寫入到陣列中（setEntryAt()）都是通過Unsafe的putOrderedObject()方法來實現，這裡並未使用具有原子寫語義的putObjectVolatile()的原因是：JMM會保證獲得鎖到釋放鎖之間所有物件的狀態更新都會在鎖被釋放之後更新到主存，從而保證這些變更對其他執行緒是可見的。

1.5 rehash

相對於HashMap的resize，ConcurrentHashMap的rehash原理類似，但是Doug Lea為rehash做了一定的優化，避免讓所有的節點都進行復制操作：由於擴容是基於2的冪指來操作，假設擴容前某HashEntry對應到Segment中陣列的index為i，陣列的容量為capacity，那麼擴容後該HashEntry對應到新陣列中的index只可能為i或者i+capacity，因此大多數HashEntry節點在擴容前後index可以保持不變。基於此，rehash方法中會定位第一個後續所有節點在擴容後index都保持不變的節點，然後將這個節點之前的所有節點重排即可。這部分程式碼如下：

1234567891011121314151617181920212223242526272829303132333435363738394041424344 private void rehash(HashEntry<K,V> node) {HashEntry<K,V>[] oldTable = table;int oldCapacity = oldTable.length;int newCapacity = oldCapacity << 1;threshold = (int)(newCapacity * loadFactor);HashEntry<K,V>[] newTable =(HashEntry<K,V>[]) new HashEntry[newCapacity];int sizeMask = newCapacity - 1;for (int i = 0; i < oldCapacity ; i++) {HashEntry<K,V> e = oldTable[i];if (e != null) {HashEntry<K,V> next = e.next;int idx = e.hash & sizeMask;if (next == null) // Single node on listnewTable[idx] = e;else { // Reuse consecutive sequence at same slotHashEntry<K,V> lastRun = e;int lastIdx = idx;for (HashEntry<K,V> last = next;last != null;last = last.next) {int k = last.hash & sizeMask;if (k != lastIdx) {lastIdx = k;lastRun = last;}}newTable[lastIdx] = lastRun;// Clone remaining nodesfor (HashEntry<K,V> p = e; p != lastRun; p = p.next) {V v = p.value;int h = p.hash;int k = h & sizeMask;HashEntry<K,V> n = newTable[k];newTable[k] = new HashEntry<K,V>(h, p.key, v, n);}}}}int nodeIndex = node.hash & sizeMask; // add the new nodenode.setNext(newTable[nodeIndex]);newTable[nodeIndex] = node;table = newTable;}

1.6 remove

和put類似，remove在真正獲得鎖之前，也會對連結串列進行遍歷以提高快取命中率。

1.7 get與containsKey

get與containsKey兩個方法幾乎完全一致：他們都沒有使用鎖，而是通過Unsafe物件的getObjectVolatile()方法提供的原子讀語義，來獲得Segment以及對應的連結串列，然後對連結串列遍歷判斷是否存在key相同的節點以及獲得該節點的value。但由於遍歷過程中其他執行緒可能對連結串列結構做了調整，因此get和containsKey返回的可能是過時的資料，這一點是ConcurrentHashMap在弱一致性上的體現。如果要求強一致性，那麼必須使用Collections.synchronizedMap()方法。

1.8 size、containsValue

這些方法都是基於整個ConcurrentHashMap來進行操作的，他們的原理也基本類似：首先不加鎖迴圈執行以下操作：迴圈所有的Segment（通過Unsafe的getObjectVolatile()以保證原子讀語義），獲得對應的值以及所有Segment的modcount之和。如果連續兩次所有Segment的modcount和相等，則過程中沒有發生其他執行緒修改ConcurrentHashMap的情況，返回獲得的值。

當迴圈次數超過預定義的值時，這時需要對所有的Segment依次進行加鎖，獲取返回值後再依次解鎖。值得注意的是，加鎖過程中要強制建立所有的Segment，否則容易出現其他執行緒建立Segment並進行put，remove等操作。程式碼如下：

123	`for(int` `j =0; j < segments.length; ++j)ensureSegment(j).lock();// force creation`

一般來說，應該避免在多執行緒環境下使用size和containsValue方法。

注1：modcount在put, replace, remove以及clear等方法中都會被修改。

注2：對於containsValue方法來說，如果在迴圈過程中發現匹配value的HashEntry，則直接返回true。

最後，與HashMap不同的是，ConcurrentHashMap並不允許key或者value為null，按照Doug Lea的說法，這麼設計的原因是在ConcurrentHashMap中，一旦value出現null，則代表HashEntry的key/value沒有對映完成就被其他執行緒所見，需要特殊處理。在JDK6中，get方法的實現中就有一段對HashEntry.value == null的防禦性判斷。但Doug Lea也承認實際執行過程中，這種情況似乎不可能發生（參考：http://cs.oswego.edu/pipermail/concurrency-interest/2011-March/007799.html）。

2. JDK8中的實現

ConcurrentHashMap在JDK8中進行了巨大改動，很需要通過原始碼來再次學習下Doug Lea的實現方法。

它摒棄了Segment（鎖段）的概念，而是啟用了一種全新的方式實現,利用CAS演算法。它沿用了與它同時期的HashMap版本的思想，底層依然由“陣列”+連結串列+紅黑樹的方式思想(JDK7與JDK8中HashMap的實現)，但是為了做到併發，又增加了很多輔助的類，例如TreeBin，Traverser等物件內部類。

2.1 重要的屬性

首先來看幾個重要的屬性，與HashMap相同的就不再介紹了，這裡重點解釋一下sizeCtl這個屬性。可以說它是ConcurrentHashMap中出鏡率很高的一個屬性，因為它是一個控制識別符號，在不同的地方有不同用途，而且它的取值不同，也代表不同的含義。

負數代表正在進行初始化或擴容操作

-1代表正在初始化

-N 表示有N-1個執行緒正在進行擴容操作

正數或0代表hash表還沒有被初始化，這個數值表示初始化或下一次進行擴容的大小，這一點類似於擴容閾值的概念。還後面可以看到，它的值始終是當前ConcurrentHashMap容量的0.75倍，這與loadfactor是對應的。

12345678910111213141516171819202122232425262728293031323334353637 /*** 盛裝Node元素的陣列它的大小是2的整數次冪* Size is always a power of two. Accessed directly by iterators.*/transient volatile Node<K,V>[] table;/*** Table initialization and resizing control. When negative, the* table is being initialized or resized: -1 for initialization,* else -(1 + the number of active resizing threads). Otherwise,* when table is null, holds the initial table size to use upon* creation, or 0 for default. After initialization, holds the* next element count value upon which to resize the table.hash表初始化或擴容時的一個控制位標識量。負數代表正在進行初始化或擴容操作-1代表正在初始化-N 表示有N-1個執行緒正在進行擴容操作正數或0代表hash表還沒有被初始化，這個數值表示初始化或下一次進行擴容的大小*/private transient volatile int sizeCtl; // 以下兩個是用來控制擴容的時候單執行緒進入的變數/*** The number of bits used for generation stamp in sizeCtl.* Must be at least 6 for 32bit arrays.*/private static int RESIZE_STAMP_BITS = 16;/*** The bit shift for recording size stamp in sizeCtl.*/private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;/** Encodings for Node hash fields. See above for explanation.*/static final int MOVED = -1; // hash值是-1，表示這是一個forwardNode節點static final int TREEBIN = -2; // hash值是-2 表示這時一個TreeBin節點

2.2 重要的類

2.2.1 Node

Node是最核心的內部類，它包裝了key-value鍵值對，所有插入ConcurrentHashMap的資料都包裝在這裡面。它與HashMap中的定義很相似，但是但是有一些差別它對value和next屬性設定了volatile同步鎖(與JDK7的Segment相同)，它不允許呼叫setValue方法直接改變Node的value域，它增加了find方法輔助map.get()方法。

2.2.2 TreeNode

樹節點類，另外一個核心的資料結構。當連結串列長度過長的時候，會轉換為TreeNode。但是與HashMap不相同的是，它並不是直接轉換為紅黑樹，而是把這些結點包裝成TreeNode放在TreeBin物件中，由TreeBin完成對紅黑樹的包裝。而且TreeNode在ConcurrentHashMap整合自Node類，而並非HashMap中的整合自LinkedHashMap.Entry<K,V>類，也就是說TreeNode帶有next指標，這樣做的目的是方便基於TreeBin的訪問。

2.2.3 TreeBin

這個類並不負責包裝使用者的key、value資訊，而是包裝的很多TreeNode節點。它代替了TreeNode的根節點，也就是說在實際的ConcurrentHashMap“陣列”中，存放的是TreeBin物件，而不是TreeNode物件，這是與HashMap的區別。另外這個類還帶有了讀寫鎖。

這裡僅貼出它的構造方法。可以看到在構造TreeBin節點時，僅僅指定了它的hash值為TREEBIN常量，這也就是個標識為。同時也看到我們熟悉的紅黑樹構造方法

2.2.4 ForwardingNode

一個用於連線兩個table的節點類。它包含一個nextTable指標，用於指向下一張表。而且這個節點的key value next指標全部為null，它的hash值為-1. 這裡面定義的find的方法是從nextTable裡進行查詢節點，而不是以自身為頭節點進行查詢。

123456789101112131415161718192021222324252627282930313233343536 /*** A node inserted at head of bins during transfer operations.*/static final class ForwardingNode<K,V> extends Node<K,V> {final Node<K,V>[] nextTable;ForwardingNode(Node<K,V>[] tab) {super(MOVED, null, null, null);this.nextTable = tab;}Node<K,V> find(int h, Object k) {// loop to avoid arbitrarily deep recursion on forwarding nodesouter: for (Node<K,V>[] tab = nextTable;;) {Node<K,V> e; int n;if (k == null || tab == null || (n = tab.length) == 0 ||(e = tabAt(tab, (n - 1) & h)) == null)return null;for (;;) {int eh; K ek;if ((eh = e.hash) == h &&((ek = e.key) == k || (ek != null && k.equals(ek))))return e;if (eh < 0) {if (e instanceof ForwardingNode) {tab = ((ForwardingNode<K,V>)e).nextTable;continue outer;}elsereturn e.find(h, k);}if ((e = e.next) == null)return null;}}}}

2.3 Unsafe與CAS

在ConcurrentHashMap中，隨處可以看到U, 大量使用了U.compareAndSwapXXX的方法，這個方法是利用一個CAS演算法實現無鎖化的修改值的操作，他可以大大降低鎖代理的效能消耗。這個演算法的基本思想就是不斷地去比較當前記憶體中的變數值與你指定的一個變數值是否相等，如果相等，則接受你指定的修改的值，否則拒絕你的操作。因為當前執行緒中的值已經不是最新的值，你的修改很可能會覆蓋掉其他執行緒修改的結果。這一點與樂觀鎖，SVN的思想是比較類似的。

2.3.1 unsafe靜態塊

unsafe程式碼塊控制了一些屬性的修改工作，比如最常用的SIZECTL 。在這一版本的concurrentHashMap中，大量應用來的CAS方法進行變數、屬性的修改工作。利用CAS進行無鎖操作，可以大大提高效能。

12345678910111213141516171819202122232425262728293031323334 private static final sun.misc.Unsafe U;private static final long SIZECTL;private static final long TRANSFERINDEX;private static final long BASECOUNT;private static final long CELLSBUSY;private static final long CELLVALUE;private static final long ABASE;private static final int ASHIFT;static {try {U = sun.misc.Unsafe.getUnsafe();Class<?> k = ConcurrentHashMap.class;SIZECTL = U.objectFieldOffset(k.getDeclaredField("sizeCtl"));TRANSFERINDEX = U.objectFieldOffset(k.getDeclaredField("transferIndex"));BASECOUNT = U.objectFieldOffset(k.getDeclaredField("baseCount"));CELLSBUSY = U.objectFieldOffset(k.getDeclaredField("cellsBusy"));Class<?> ck = CounterCell.class;CELLVALUE = U.objectFieldOffset(ck.getDeclaredField("value"));Class<?> ak = Node[].class;ABASE = U.arrayBaseOffset(ak);int scale = U.arrayIndexScale(ak);if ((scale & (scale - 1)) != 0)throw new Error("data type scale not a power of two");ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);} catch (Exception e) {throw new Error(e);}}

2.3.2 三個核心方法

ConcurrentHashMap定義了三個原子操作，用於對指定位置的節點進行操作。正是這些原子操作保證了ConcurrentHashMap的執行緒安全。

123456789101112131415 //獲得在i位置上的Node節點static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);}//利用CAS演算法設定i位置上的Node節點。之所以能實現併發是因為他指定了原來這個節點的值是多少//在CAS演算法中，會比較記憶體中的值與你指定的這個值是否相等，如果相等才接受你的修改，否則拒絕你的修改//因此當前執行緒中的值並不是最新的值，這種修改可能會覆蓋掉其他執行緒的修改結果有點類似於SVNstatic final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,Node<K,V> c, Node<K,V> v) {return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);}//利用volatile方法設定節點位置的值static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);}

2.4 初始化方法initTable

對於ConcurrentHashMap來說，呼叫它的構造方法僅僅是設定了一些引數而已。而整個table的初始化是在向ConcurrentHashMap中插入元素的時候發生的。如呼叫put、computeIfAbsent、compute、merge等方法的時候，呼叫時機是檢查table==null。

初始化方法主要應用了關鍵屬性sizeCtl 如果這個值〈0，表示其他執行緒正在進行初始化，就放棄這個操作。在這也可以看出ConcurrentHashMap的初始化只能由一個執行緒完成。如果獲得了初始化許可權，就用CAS方法將sizeCtl置為-1，防止其他執行緒進入。初始化陣列後，將sizeCtl的值改為0.75*n。

1234567891011121314151617181920212223242526 /*** Initializes table, using the size recorded in sizeCtl.*/private final Node<K,V>[] initTable() {Node<K,V>[] tab; int sc;while ((tab = table) == null || tab.length == 0) {//sizeCtl表示有其他執行緒正在進行初始化操作，把執行緒掛起。對於table的初始化工作，只能有一個執行緒在進行。if ((sc = sizeCtl) < 0)Thread.yield(); // lost initialization race; just spinelse if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {//利用CAS方法把sizectl的值置為-1 表示本執行緒正在進行初始化try {if ((tab = table) == null || tab.length == 0) {int n = (sc > 0) ? sc : DEFAULT_CAPACITY;@SuppressWarnings("unchecked")Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];table = tab = nt;sc = n - (n >>> 2);//相當於0.75*n 設定一個擴容的閾值}} finally {sizeCtl = sc;}break;}}return tab;}

2.5 擴容方法 transfer

當ConcurrentHashMap容量不足的時候，需要對table進行擴容。這個方法的基本思想跟HashMap是很像的，但是由於它是支援併發擴容的，所以要複雜的多。原因是它支援多執行緒進行擴容操作，而並沒有加鎖。我想這樣做的目的不僅僅是為了滿足concurrent的要求，而是希望利用併發處理去減少擴容帶來的時間影響。因為在擴容的時候，總是會涉及到從一個“陣列”到另一個“陣列”拷貝的操作，如果這個操作能夠併發進行，那真真是極好的了。

整個擴容操作分為兩個部分

第一部分是構建一個nextTable,它的容量是原來的兩倍，這個操作是單執行緒完成的。這個單執行緒的保證是通過RESIZE_STAMP_SHIFT這個常量經過一次運算來保證的，這個地方在後面會有提到；
第二個部分就是將原來table中的元素複製到nextTable中，這裡允許多執行緒進行操作。

先來看一下單執行緒是如何完成的：

它的大體思想就是遍歷、複製的過程。首先根據運算得到需要遍歷的次數i，然後利用tabAt方法獲得i位置的元素：

如果這個位置為空，就在原table中的i位置放入forwardNode節點，這個也是觸發併發擴容的關鍵點；

如果這個位置是Node節點（fh>=0），如果它是一個連結串列的頭節點，就構造一個反序連結串列，把他們分別放在nextTable的i和i+n的位置上

如果這個位置是TreeBin節點（fh<0），也做一個反序處理，並且判斷是否需要untreefi，把處理的結果分別放在nextTable的i和i+n的位置上

遍歷過所有的節點以後就完成了複製工作，這時讓nextTable作為新的table，並且更新sizeCtl為新容量的0.75倍，完成擴容。

再看一下多執行緒是如何完成的：

在程式碼的69行有一個判斷，如果遍歷到的節點是forward節點，就向後繼續遍歷，再加上給節點上鎖的機制，就完成了多執行緒的控制。多執行緒遍歷節點，處理了一個節點，就把對應點的值set為forward，另一個執行緒看到forward，就向後遍歷。這樣交叉就完成了複製工作。而且還很好的解決了執行緒安全的問題。這個方法的設計實在是讓我膜拜。

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061