第二章 ConcurrentHashMap原始碼解析
阿新 • • 發佈：2019-01-18
注：在看這篇文章之前，如果對HashMap的層不清楚的話，建議先去看看HashMap原始碼解析。
1、對於ConcurrentHashMap需要掌握以下幾點
- Map的建立：ConcurrentHashMap()
- 往Map中新增鍵值對：即put(Object key, Object value)方法
- 獲取Map中的單個物件：即get(Object key)方法
- 刪除Map中的物件：即remove(Object key)方法
- 判斷物件是否存在於Map中：containsKey(Object key)
- 遍歷Map中的物件：即keySet().iterator()，在實際中更常用的是增強型的for迴圈去做遍歷
2、ConcurrentHashMap的建立
注：在往下看之前，心裡先有這樣一個映像：ConcurrentHashMap的資料結構：一個指定個數的Segment陣列，陣列中的每一個元素Segment相當於一個HashTable。
2.1、使用方法：
Map<String, Object> map = new ConcurrentHashMap<String, Object>();
2.2、原始碼：
ConcurrentHashMap相關屬性：
/** * 用於分段 */ // 根據這個數來計算segment的個數,segment的個數是僅小於這個數且是2的幾次方的一個數（ssize）View Codestatic final int DEFAULT_CONCURRENCY_LEVEL = 16; // 最大的分段（segment）數(2的16次方) static final int MAX_SEGMENTS = 1 << 16; /** * 用於HashEntry */ // 預設的用於計算Segment陣列中的每一個segment的HashEntry[]的容量，但是並不是每一個segment的HashEntry[]的容量 static final int DEFAULT_INITIAL_CAPACITY = 16;// 預設的載入因子（用於resize） static final float DEFAULT_LOAD_FACTOR = 0.75f; // 用於計算Segment陣列中的每一個segment的HashEntry[]的最大容量（2的30次方） static final int MAXIMUM_CAPACITY = 1 << 30; /** * segments陣列 * 每一個segment元素都看做是一個HashTable */ final Segment<K, V>[] segments; /** * 用於擴容 */ final int segmentMask;// 用於根據給定的key的hash值定位到一個Segment final int segmentShift;// 用於根據給定的key的hash值定位到一個Segment
Segment類（ConcurrentHashMap的內部類）
/** * 一個特殊的HashTable */ static final class Segment<K, V> extends ReentrantLock implements Serializable { private static final long serialVersionUID = 2249069246763182397L; transient volatile int count;// 該Segment中的包含的所有HashEntry中的key-value的個數 transient int modCount;// 併發標記 /* * 元素個數超出了這個值就擴容 threshold==(int)(capacity * loadFactor) * 值得注意的是，只是當前的Segment擴容，所以這是Segment自己的一個變數，而不是ConcurrentHashMap的 */ transient int threshold; transient volatile HashEntry<K, V>[] table;// 連結串列陣列 final float loadFactor; /** * 這裡要注意一個很不好的程式設計習慣，就是小寫l，容易與數字1混淆，所以最好不要用小寫l，可以改為大寫L */ Segment(int initialCapacity, float lf) { loadFactor = lf;//每個Segment的載入因子 setTable(HashEntry.<K, V> newArray(initialCapacity)); } /** * 建立一個Segment陣列，容量為i */ @SuppressWarnings("unchecked") static final <K, V> Segment<K, V>[] newArray(int i) { return new Segment[i]; } /** * Sets table to new HashEntry array. Call only while holding lock or in * constructor. */ void setTable(HashEntry<K, V>[] newTable) { threshold = (int) (newTable.length * loadFactor);// 設定擴容值 table = newTable;// 設定連結串列陣列 }View Code
說明：只列出了Segement的全部屬性和建立ConcurrentHashMap時所用到的方法。
HashEntry類（ConcurrentHashMap的內部類）
/** * Segment中的HashEntry節點 類比HashMap中的Entry節點 */ static final class HashEntry<K, V> { final K key;// 鍵 final int hash;//hash值 volatile V value;// 實現執行緒可見性 final HashEntry<K, V> next;// 下一個HashEntry HashEntry(K key, int hash, HashEntry<K, V> next, V value) { this.key = key; this.hash = hash; this.next = next; this.value = value; } /* * 建立HashEntry陣列，容量為傳入的i */ @SuppressWarnings("unchecked") static final <K, V> HashEntry<K, V>[] newArray(int i) { return new HashEntry[i]; } }View Code
ConcurrentHashMap(int initialCapacity,float loadFactor,int concurrencyLevel)
1 /** 2 * 建立ConcurrentHashMap 3 * @param initialCapacity 用於計算Segment陣列中的每一個segment的HashEntry[]的容量， 但是並不是每一個segment的HashEntry[]的容量 4 * @param loadFactor 5 * @param concurrencyLevel 用於計算Segment陣列的大小（可以傳入不是2的幾次方的數，但是根據下邊的計算，最終segment陣列的大小ssize將是2的幾次方的數） 6 * 7 * 步驟： 8 * 這裡以預設的無參構造器引數為例，initialCapacity==16，loadFactor==0.75f，concurrencyLevel==16 9 * 1)檢查各引數是否符合要求 10 * 2)根據concurrencyLevel(16),計算Segment[]的容量ssize(16)與擴容移位條件sshift(4) 11 * 3)根據sshift與ssize計算將來用於定位到相應Segment的引數segmentShift與segmentMask 12 * 4)根據ssize建立Segment[]陣列，容量為ssize(16) 13 * 5)根據initialCapacity(16)與ssize計算用於計算HashEntry[]容量的引數c(1) 14 * 6)根據c計算HashEntry[]的容量cap(1) 15 * 7)根據cap與loadFactor(0.75)為每一個Segment[i]都例項化一個Segment 16 * 8)每一個Segment的例項化都做下面這些事兒： 17 * 8.1)為當前的Segment初始化其loadFactor為傳入的loadFactor(0.75) 18 * 8.2)建立一個HashEntry[]，容量為傳入的cap(1) 19 * 8.3)根據創建出來的HashEntry的容量(1)和初始化的loadFactor(0.75)，計算擴容因子threshold(0) 20 * 8.4)初始化Segment的table為剛剛創建出來的HashEntry 21 */ 22 public ConcurrentHashMap(int initialCapacity,float loadFactor,int concurrencyLevel) { 23 // 檢查引數情況 24 if (loadFactor <= 0f || initialCapacity < 0 || concurrencyLevel <= 0) 25 throw new IllegalArgumentException(); 26 27 if (concurrencyLevel > MAX_SEGMENTS) 28 concurrencyLevel = MAX_SEGMENTS; 29 30 /** 31 * 找一個能夠正好小於concurrencyLevel的數（這個數必須是2的幾次方的數） 32 * eg.concurrencyLevel==16==>sshift==4,ssize==16 33 * 當然，如果concurrencyLevel==15也是上邊這個結果 34 */ 35 int sshift = 0; 36 int ssize = 1;// segment陣列的長度 37 while (ssize < concurrencyLevel) { 38 ++sshift; 39 ssize <<= 1;// ssize=ssize*2 40 } 41 42 segmentShift = 32 - sshift;// eg.segmentShift==32-4=28 用於根據給定的key的hash值定位到一個Segment 43 segmentMask = ssize - 1;// eg.segmentMask==16-1==15 用於根據給定的key的hash值定位到一個Segment 44 this.segments = Segment.newArray(ssize);// 構造出了Segment[ssize]陣列 eg.Segment[16] 45 46 /* 47 * 下面將為segment陣列中新增Segment元素 48 */ 49 if (initialCapacity > MAXIMUM_CAPACITY) 50 initialCapacity = MAXIMUM_CAPACITY; 51 int c = initialCapacity / ssize;// eg.initialCapacity==16,c==16/16==1 52 if (c * ssize < initialCapacity)// eg.initialCapacity==17,c==17/16=1,這時1*16<17,所以c=c+1==2 53 ++c;// 為了少執行這一句，最好將initialCapacity設定為2的幾次方 54 int cap = 1;// 每一個Segment中的HashEntry[]的初始化容量 55 while (cap < c) 56 cap <<= 1;// 建立容量 57 58 for (int i = 0; i < this.segments.length; ++i) 59 // 這一塊this.segments.length就是ssize，為了不去計算這個值，可以直接改成i<ssize 60 this.segments[i] = new Segment<K, V>(cap, loadFactor); 61 }View Code
注意：這個方法裡邊我在頭部所寫的註釋非常重要，在這塊註釋寫明瞭：
- 每一個引數的作用
- 整個ConcurrentHashMap的一個建立步驟（以預設的引數值為例）
ConcurrentHashMap()
/** * 建立ConcurrentHashMap */ public ConcurrentHashMap() { this(DEFAULT_INITIAL_CAPACITY, // 16 DEFAULT_LOAD_FACTOR, // 0.75f DEFAULT_CONCURRENCY_LEVEL);// 16 }View Code
該方法呼叫了上邊的三參構造器。
五點注意：
- 傳入的concurrencyLevel只是用於計算Segment陣列的大小（可以傳入不是2的幾次方的數，但是根據下邊的計算，最終segment陣列的大小ssize將是2的幾次方的數），並非真正的Segment陣列的大小
- 傳入的initialCapacity只是用於計算Segment陣列中的每一個segment的HashEntry[]的容量， 但是並不是每一個segment的HashEntry[]的容量，而每一個HashEntry[]的容量不是2的幾次方
- 非常值得注意的是，在預設情況下，創建出的HashEntry[]陣列的容量為1，並不是傳入的initialCapacity(16)，證實了上一點；而每一個Segment的擴容因子threshold，一開始算出來是0，即開始put第一個元素就要擴容，不太理解JDK為什麼這樣做。
- 想要在初始化時擴大HashEntry[]的容量，可以指定initialCapacity引數，且指定時最好指定為2的幾次方的一個數，這樣的話，在程式碼執行中可能會少執行一句"c++"，具體參看三參構造器的註釋
- 對於Concurrenthashmap的擴容而言，只會擴當前的Segment，而不是整個Concurrenthashmap中的所有Segment都擴
兩點改進：
在Concurrenthashmap的構造過程中，相對於JDK的程式碼，有兩點改進：
- 在遍歷Segment陣列為每一個數組元素例項化的時候，可以直接寫作i<ssize，而不必在每次迴圈時都去計算this.segments.length，JDK程式碼如下，可以按照程式碼中的註釋做優化
for (int i = 0; i < this.segments.length; ++i) // 這一塊this.segments.length就是ssize，為了不去計算這個值，可以直接改成i<ssize this.segments[i] = new Segment<K, V>(cap, loadFactor);
View Code - 另外一個，就是在程式中，儘量少用小寫"l"，容易與數字"1"混淆，要麼不用"l"，若要用的話，就用大寫"L"，JDK程式碼如下，可參照註釋進行優化：
/** * 這裡要注意一個很不好的程式設計習慣，就是小寫l，容易與數字1混淆，所以最好不要用小寫l，可以改為大寫L */ Segment(int initialCapacity, float lf) { loadFactor = lf;//每個Segment的載入因子 setTable(HashEntry.<K, V> newArray(initialCapacity)); }
View Code
一個疑問：
- 在預設情況下，創建出的HashEntry[]陣列的容量為1，而每一個Segment的擴容因子threshold，一開始算出來是0，即開始put第一個元素就要擴容，不太理解JDK為什麼這樣做。在我們實際開發中，其實空間有的是，所以我們一般會採用"以適當的空間換取時間"的方式，所以我們會適當的擴大HashEntry[]，以確保在put資料的時候儘量減少擴容才對，但是JDK這樣做到底是為了什麼？是為了減少空間嗎？還是我本身的理解就有問題？求大神指點！！！
- 注意我上邊說的適當容量，是因為如果容量設的太大，可能會導致某個HashEntry[i]中的HashEntry連結串列過長，進而影響查詢的效率，容量設的太小的話，有需要不斷擴容，影響插入效率。
3、put(Object key, Object value)
上述方法，若新增已有key的key-value對，則新值覆蓋舊值。
putIfAbsent(K key, V value)：若新增已有key的key-value對，直接返回舊值，則新值相當於沒有新增。
使用方法：
map.put("hello", "world");
原始碼：
ConcurrentHashMap的put(Object key, Object value)方法
/** * 將key-value放入map * 注意：key和value都不可以為空 * 步驟： * 1)計算key.hashCode()的hash值 * 2)根據hash值定位到某個Segment * 3)呼叫Segment的put()方法 * Segment的put()方法: * 1)上鎖 * 2)從主記憶體中讀取key-value對個數count * 3)count+1如果大於threshold，執行rehash() * 4)計算將要插入的HashEntry[]的下標index * 5)獲取HashEntry的頭節點HashEntry[index]-->first * 6)從頭結點開始遍歷整個HashEntry連結串列， * 6.1)若找到與key和hash相同的節點，則判斷onlyIfAbsent如果為false，新值覆蓋舊值，返回舊值；如果為true，則直接返回舊值（相當於不新增重複key的元素） * 6.2)若沒有找到與key和hash相同的節點，則建立新節點HashEntry，並將之前的有節點作為新節點的next，即將新節點放入鏈頭，然後將新節點賦值給HashEntry[index],將count強制寫入主記憶體，最後返回null */ public V put(K key, V value) { if (key == null || value == null) throw new NullPointerException(); int hash = hash(key.hashCode());//計算key.hashCode()的hash值 /** * 根據hash值定位到某個Segment，呼叫Segment的put()方法 */ return segmentFor(hash).put(key, hash, value, false); }View Code
注意：
- key和value都不可為null，這一點與HashMap不同，但是從程式碼來看，並沒有判斷key為空的情況，這一段程式碼在哪裡呢？為了可讀性，建議將判斷的地方改為如下程式碼
if (key == null || value == null) throw new NullPointerException();
View Code - 註釋部分寫明瞭整個插入流程，詳細的流程步驟見程式碼，這裡列出大致流程
- 根據key獲取key.hashCode的hash值
- 根據hash值算出將要插入的Segment
- 根據hash值與Segment中的HashEntry的容量-1按位與獲取將要插入的HashEntry的index
- 若HashEntry[index]中的HashEntry連結串列有與插入元素相同的key和hash值，根據onlyIfAbsent決定是否替換舊值
- 若沒有相同的key和hash，直接返回將新節點插入鏈頭，原來的頭節點設為新節點的next（採用的方式與HashMap一致，都是HashEntry替換的方法）
Segment的put(K key, int hash, V value, boolean onlyIfAbsent)
/** * 往當前segment中新增key-value * 注意： * 1)onlyIfAbsent-->false如果有舊值存在,新值覆蓋舊值,返回舊值；true如果有舊值存在,則直接返回舊值,相當於不新增元素(不可新增重複key的元素) * 2)ReentrantLock的用法 * 3)volatile只能配合鎖去使用才能實現原子性 */ V put(K key, int hash, V value, boolean onlyIfAbsent) { lock();//加鎖：ReentrantLock try { int c = count;//當前Segment中的key-value對（注意：由於count是volatile型的，所以讀的時候工作記憶體會從主記憶體重新載入count值） if (c++ > threshold) // 需要擴容 rehash();//擴容 HashEntry<K, V>[] tab = table; int index = hash & (tab.length - 1);//按位與獲取陣列下標：與HashMap相同 HashEntry<K, V> first = tab[index];//獲取相應的HashEntry[i]中的頭節點 HashEntry<K, V> e = first; //一直遍歷到與插入節點的hash和key相同的節點e；若沒有，最後e==null while (e != null && (e.hash != hash || !key.equals(e.key))) e = e.next; V oldValue;//舊值 if (e != null) {//table中已經有與將要插入節點相同hash和key的節點 oldValue = e.value;//獲取舊值 if (!onlyIfAbsent) e.value = value;//false 覆蓋舊值 true的話，就不新增元素了 } else {//table中沒有與將要插入節點相同hash或key的節點 oldValue = null; ++modCount; tab[index] = new HashEntry<K, V>(key, hash, first, value);//將頭節點作為新節點的next，所以新加入的元素也是新增在鏈頭 count = c; //設定key-value對（注意：由於count是volatile型的，所以寫的時候工作記憶體會立即向主記憶體重新寫入count值） } return oldValue; } finally { unlock();//手工釋放鎖 } }View Code
注意：在註釋中已經寫明瞭，這裡還是要寫一下
- onlyIfAbsent-->false如果有舊值存在,新值覆蓋舊值,返回舊值；true如果有舊值存在,則直接返回舊值,相當於不新增元素
- ReentrantLock的用法：必須手工釋放鎖。可實現Synchronized的效果，原子性。
- volatile需要配合鎖去使用才能實現原子性，否則在多執行緒操作的情況下依然不夠用，在程式中，count變數（當前Segment中的key-value對個數）通過volatile修飾，實現記憶體可見性（關於記憶體可見性以後會仔細去記錄，這裡列出大概的一個流程）在有鎖保證了原子性的情況下
- 當我們讀取count變數的時候，會強制從主記憶體中讀取count的最新值
- 當我們對count變數進行賦值之後，會強制將最新的count值刷到主記憶體中去
- 通過以上兩點，我們可以保證在高併發的情況下，執行這段流程的執行緒可以讀取到最新值
- 在這裡的ReentrantLock與volatile結合的用法值得我們學習
補：volatile的介紹見《附2 volatile》，連結如下：
hash(int h)
/** * 對key.hashCode()進行hash計算 * @param h key.hashCode() */ private static int hash(int h) { // Spread bits to regularize both segment and index locations, // using variant of single-word Wang/Jenkins hash. h += (h << 15) ^ 0xffffcd7d; h ^= (h >>> 10); h += (h << 3); h ^= (h >>> 6); h += (h << 2) + (h << 14); return h ^ (h >>> 16); }View Code
segmentFor(int hash)
/** * 根據給定的key的hash值定位到一個Segment * @param hash */ final Segment<K, V> segmentFor(int hash) { return segments[(hash >>> segmentShift) & segmentMask]; }View Code
注意：hash(int h)與segmentFor(int hash)這兩個方法應該會盡量將key的hash值打散，從而保證儘可能多的同時在多個Segment上進行put操作，而不是在同一個Segment上執行多個put操作，這樣之後，在同一個Segment中，要儘可能的保證向HashEntry[]的不同元素上進行put，而不是向同一個元素上一直put，以上兩個函式究竟是怎樣保證實現這樣的將hash打散的效果呢？求大神指點啊！！！
rehash()
JDK的實現程式碼：
/** * 步驟： * 需要注意的是：同一個桶下邊的HashEntry連結串列中的每一個元素的hash值不一定相同，只是hash&(table.length-1)的結果相同 * 1)建立一個新的HashEntry陣列，容量為舊陣列的二倍 * 2)計算新的threshold * 3)遍歷舊陣列的每一個元素，對於每一個元素 * 3.1)根據頭節點e重新計算將要存入的新陣列的索引idx * 3.2)若整個連結串列只有一個節點e，則是直接將e賦給newTable[idx]即可 * 3.3)若整個連結串列還有其他節點，先算出最後一個節點lastRun的位置lastIdx，並將最後一個節點賦值給newTable[lastIdx] * 3.4)最後將從頭節點開始到最後一個節點之前的所有節點計算其將要儲存的索引k，然後建立新節點，將新節點賦給newTable[k]，並將之前newTable[k]上存在的節點作為新節點的下一節點 */ void rehash() { HashEntry<K, V>[] oldTable = table; int oldCapacity = oldTable.length; if (oldCapacity >= MAXIMUM_CAPACITY) return; HashEntry<K, V>[] newTable = HashEntry.newArray(oldCapacity << 1);//擴容為原來二倍 threshold = (int) (newTable.length * loadFactor);//計算新的擴容臨界值 int sizeMask = newTable.length - 1; for (int i = 0; i < oldCapacity; i++) { // We need to guarantee that any existing reads of old Map can // proceed. So we cannot yet null out each bin. HashEntry<K, V> e = oldTable[i];//頭節點 if (e != null) { HashEntry<K, V> next = e.next; int idx = e.hash & sizeMask;//重新按位與計算將要存放的新陣列中的索引 if (next == null)//如果是隻有一個頭節點，只需將頭節點設定到newTable[idx]即可 newTable[idx] = e; else { // Reuse trailing consecutive sequence at same slot HashEntry<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 all remaining nodes for (HashEntry<K, V> p = e; p != lastRun; p = p.next) { int k = p.hash & sizeMask; HashEntry<K, V> n = newTable[k];//獲取newTable[k]已經存在的HashEntry，並將此HashEntry賦給n //建立新節點，並將之前的n作為新節點的下一節點 newTable[k] = new HashEntry<K, V>(p.key, p.hash, n,p.value); } } } } table = newTable; }View Code
個人感覺JDK的實現方式比較拖沓，改造後的程式碼如下，如有問題，請指出！！！
我對其進行改造後的實現程式碼：
/** * 步驟： * 需要注意的是：同一個桶下邊的HashEntry連結串列中的每一個元素的hash值不一定相同，只是hash&(table.length-1)的結果相同 * 1)建立一個新的HashEntry陣列，容量為舊陣列的二倍 * 2)計算新的threshold * 3)遍歷舊陣列的每一個元素，對於每一個元素（即一個連結串列） * 3.1)獲取頭節點e * 3.2)從頭節點開始到最後一個節點（null之前的那個節點）的所有節點計算其將要儲存的索引k，然後建立新節點，將新節點賦給newTable[k]，並將之前newTable[k]上存在的節點作為新節點的下一節點 */ void rehash() { HashEntry<K, V>[] oldTable = table; int oldCapacity = oldTable.length; if (oldCapacity >= MAXIMUM_CAPACITY) return; HashEntry<K, V>[] newTable = HashEntry.newArray(oldCapacity << 1);//擴容為原來二倍 threshold = (int) (newTable.length * loadFactor);//計算新的擴容臨界值 int sizeMask = newTable.length - 1; for (int i = 0; i < oldCapacity; i++) {//遍歷每一個數組元素 // We need to guarantee that any existing reads of old Map can // proceed. So we cannot yet null out each bin. HashEntry<K, V> e = oldTable[i];//頭節點 if (e != null) { for (HashEntry<K, V> p = e; p != null; p = p.next) {//遍歷陣列元素中的連結串列 int k = p.hash & sizeMask; HashEntry<K, V> n = newTable[k];//獲取newTable[k]已經存在的HashEntry，並將此HashEntry賦給n //建立新節點，並將之前的n作為新節點的下一節點 newTable[k] = new HashEntry<K, V>(p.key, p.hash, n,p.value); } } } table = newTable; }View Code
注意點：
- 同一個桶下邊的HashEntry連結串列中的每一個元素的hash值不一定相同，只是index = hash&(table.length-1)的結果相同，當table.length發生變化時，同一個桶下各個HashEntry算出來的index會不同。
總結：ConcurrentHashMap基於concurrencyLevel劃分出多個Segment來儲存key-value，這樣的話put的時候只鎖住當前的Segment，可以避免put的時候鎖住整個map，從而減少了併發時的阻塞現象。
4、get(Object key)
使用方法：
map.get("hello");
原始碼：
ConcurrentHashMap的get(Object key)
/** * 根據key獲取value * 步驟： * 1)根據key獲取hash值 * 2)根據hash值找到相應的Segment * 呼叫Segment的get(Object key, int hash) * 3)根據hash值找出HashEntry陣列中的索引index，並返回HashEntry[index] * 4)遍歷整個HashEntry[index]連結串列，找出hash和key與給定引數相等的HashEntry，例如e， * 4.1)如沒找到e，返回null * 4.2)如找到e，獲取e.value * 4.2.1)如果e.value!=null，直接返回 * 4.2.2)如果e.value==null,則先加鎖，等併發的put操作將value設定成功後，再返回value值 */ public V get(Object key) { int hash = hash(key.hashCode()); return segmentFor(hash).get(key, hash); }View Code
Segment的get(Object key, int hash)
/** * 根據key和hash值獲取value */ V get(Object key, int hash) { if (count != 0) { // read-volatile HashEntry<K, V> e = getFirst(hash);//找到HashEntry[index] while (e != null) {//遍歷整個連結串列 if (e.hash == hash && key.equals(e.key)) { V v = e.value; if (v != null) return v; /* * 如果V等於null,有可能是當下的這個HashEntry剛剛被建立，value屬性還沒有設定成功， * 這時候我們讀到是該HashEntry的value的預設值null,所以這裡加鎖，等待put結束後，返回value值 */ return readValueUnderLock(e); } e = e.next; } } <