Java併發（十七）：ConcurrentHashMap
阿新 • • 發佈：2018-12-18
先做總結：
1、HashMap HashTable ConcurrentHashMap
HashMap：執行緒不安全
HashTable：執行緒安全，每個方法都加了 synchronized 修飾。類似 Collections.synchronizedMap(hashMap)
對讀寫加鎖，獨佔式，一個執行緒在讀時其他執行緒必須等待，吞吐量較低，效能較為低下。
ConcurrentHashMap：利用CAS+Synchronized來保證併發的安全性。資料結構同HashMap。
2、ConcurrentHashMap如何實現執行緒安全？
（1）get()方法使用tabAt(Node<K, V>[], int)方法
呼叫Unsafe的native方法 getObjectVolatile(Object obj, long offset);
// 獲取obj物件中offset偏移地址對應的object型field的值，支援volatile load語義，即：讓快取中的資料失效，重新從主記憶體載入資料
（2）put()方法
①需要獲取陣列上的Node時同樣使用tabAt()方法
②設定陣列上Node是使用casTabAt() 方法，
casTabAt()呼叫Unsafe的native方法compareAndSwapObject()，CAS操作
③雜湊衝突之後，需要操作改hash值對應的連結串列/紅黑樹，此時synchronized(該連結串列第一個Node)
保證執行緒安全的基礎上，減小了鎖的粒度。
3、執行緒安全的容器只能保證自身的資料不被破壞，但無法保證業務的行為是否正確。
public static void demo1() { final Map<String, Integer> count = new ConcurrentHashMap<>(); final CountDownLatch endLatch = new CountDownLatch(2); Runnable task = new Runnable() { @Overridepublic void run() { for (int i = 0; i < 5; i++) { Integer value = count.get("a"); if (null == value) { count.put("a", 1); } else { count.put("a", value + 1); } } endLatch.countDown(); } }; new Thread(task).start(); new Thread(task).start(); try { endLatch.await(); System.out.println(count); } catch (Exception e) { e.printStackTrace(); } }
demo1是兩個執行緒操作ConcurrentHashMap，意圖將value變為10。但是，因為多個執行緒用相同的key呼叫時，很可能會覆蓋相互的結果，造成記錄的次數比實際出現的次數少。
當然可以用鎖解決這個問題，但是也可以使用ConcurrentMap定義的方法：
V putIfAbsent(K key, V value) 如果key對應的value不存在，則put進去，返回null。否則不put，返回已存在的value。 boolean remove(Object key, Object value) 如果key對應的值是value，則移除K-V，返回true。否則不移除，返回false。 boolean replace(K key, V oldValue, V newValue) 如果key對應的當前值是oldValue，則替換為newValue，返回true。否則不替換，返回false。
修改：
public static void demo1() { final Map<String, Integer> count = new ConcurrentHashMap<>(); final CountDownLatch endLatch = new CountDownLatch(2); Runnable task = new Runnable() { @Override public void run() { Integer oldValue, newValue; for (int i = 0; i < 5; i++) { while (true) { oldValue = count.get("a"); if (null == oldValue) { newValue = 1; if (count.putIfAbsent("a", newValue) == null) { break; } } else { newValue = oldValue + 1; if (count.replace("a", oldValue, newValue)) { break; } } } } endLatch.countDown(); } }; new Thread(task).start(); new Thread(task).start(); try { endLatch.await(); System.out.println(count); } catch (Exception e) { e.printStackTrace(); } }
由於ConcurrentMap中不能儲存value為null的值，所以需要處理不存在和已存在兩種情況，不過可以使用AtomicInteger來替代。
public static void demo1() { final Map<String, AtomicInteger> count = new ConcurrentHashMap<>(); final CountDownLatch endLatch = new CountDownLatch(2); Runnable task = new Runnable() { @Override public void run() { AtomicInteger oldValue; for (int i = 0; i < 5; i++) { oldValue = count.get("a"); if (null == oldValue) { AtomicInteger zeroValue = new AtomicInteger(0); oldValue = count.putIfAbsent("a", zeroValue); if (null == oldValue) { oldValue = zeroValue; } } oldValue.incrementAndGet(); } endLatch.countDown(); } }; new Thread(task).start(); new Thread(task).start(); try { endLatch.await(); System.out.println(count); } catch (Exception e) { e.printStackTrace(); } }
一、屬性
// 最大容量：2^30=1073741824 private static final int MAXIMUM_CAPACITY = 1 << 30; // 預設初始值，必須是2的幕數 private static final int DEFAULT_CAPACITY = 16; // static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8; // private static final int DEFAULT_CONCURRENCY_LEVEL = 16; // private static final float LOAD_FACTOR = 0.75f; // 連結串列轉紅黑樹閥值,> 8 連結串列轉換為紅黑樹 static final int TREEIFY_THRESHOLD = 8; //樹轉連結串列閥值，小於等於6（tranfer時，lc、hc=0兩個計數器分別++記錄原bin、新binTreeNode數量，<=UNTREEIFY_THRESHOLD 則untreeify(lo)） static final int UNTREEIFY_THRESHOLD = 6; // static final int MIN_TREEIFY_CAPACITY = 64; // private static final int MIN_TRANSFER_STRIDE = 16; // private static int RESIZE_STAMP_BITS = 16; // 2^15-1，help resize的最大執行緒數 private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1; // 32-16=16，sizeCtl中記錄size大小的偏移量 private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS; // forwarding nodes的hash值 static final int MOVED = -1; // 樹根節點的hash值 static final int TREEBIN = -2; // ReservationNode的hash值 static final int RESERVED = -3; // 可用處理器數量 static final int NCPU = Runtime.getRuntime().availableProcessors();
幾個很重要的概念：
（1）table：用來存放Node節點資料的，預設為null，預設大小為16的陣列，每次擴容時大小總是2的冪次方；
（2）nextTable：擴容時新生成的資料，陣列為table的兩倍；
（3）Node：節點，儲存key-value的資料結構；
（4）ForwardingNode：一個特殊的Node節點，hash值為-1，其中儲存nextTable的引用。只有table發生擴容的時候，ForwardingNode才會發揮作用，作為一個佔位符放在table中表示當前節點為null或則已經被移動
（5）sizeCtl：控制識別符號，用來控制table初始化和擴容操作的，在不同的地方有不同的用途，其值也不同，所代表的含義也不同
- 負數代表正在進行初始化或擴容操作
- -1代表正在初始化
- -N 表示有N-1個執行緒正在進行擴容操作
- 正數或0代表hash表還沒有被初始化，這個數值表示初始化或下一次進行擴容的大小
二、構造
public ConcurrentHashMap() { } public ConcurrentHashMap(int initialCapacity) { if (initialCapacity < 0) throw new IllegalArgumentException(); int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY : tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1)); this.sizeCtl = cap; } public ConcurrentHashMap(Map<? extends K, ? extends V> m) { this.sizeCtl = DEFAULT_CAPACITY; putAll(m); } public ConcurrentHashMap(int initialCapacity, float loadFactor) { this(initialCapacity, loadFactor, 1); } public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) { if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0) throw new IllegalArgumentException(); if (initialCapacity < concurrencyLevel) // Use at least as many bins initialCapacity = concurrencyLevel; // as estimated threads long size = (long)(1.0 + (long)initialCapacity / loadFactor); int cap = (size >= (long)MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : tableSizeFor((int)size); this.sizeCtl = cap; }
初始化： initTable()
private final Node<K,V>[] initTable() { Node<K,V>[] tab; int sc; while ((tab = table) == null || tab.length == 0) { // 初始化的"功勞"被其他執行緒"搶去"了 if ((sc = sizeCtl) < 0) Thread.yield(); // lost initialization race; just spin // CAS 一下，將 sizeCtl 設定為 -1，代表搶到了鎖 else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { try { if ((tab = table) == null || tab.length == 0) { // DEFAULT_CAPACITY 預設初始容量是 16 int n = (sc > 0) ? sc : DEFAULT_CAPACITY; // 初始化陣列，長度為 16 或初始化時提供的長度 Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; // 將這個陣列賦值給 table，table 是 volatile 的 table = tab = nt; // 如果 n 為 16 的話，那麼這裡 sc = 12 // 其實就是 0.75 * n sc = n - (n >>> 2); } } finally { // 設定 sizeCtl 為 sc，我們就當是 12 吧 sizeCtl = sc; } break; } } return tab; }
三、put()
public V put(K key, V value) { return putVal(key, value, false); } final V putVal(K key, V value, boolean onlyIfAbsent) { if (key == null || value == null) throw new NullPointerException(); // 得到 hash 值 int hash = spread(key.hashCode()); // 用於記錄相應連結串列的長度 int binCount = 0; for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; // 如果陣列"空"，進行陣列初始化 if (tab == null || (n = tab.length) == 0) // 初始化陣列，後面會詳細介紹 tab = initTable(); // 找該 hash 值對應的陣列下標，得到第一個節點 f else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { // 如果陣列該位置為空， // 用一次 CAS 操作將這個新值放入其中即可，這個 put 操作差不多就結束了，可以拉到最後面了 // 如果 CAS 失敗，那就是有併發操作，進到下一個迴圈就好了 if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) break; // no lock when adding to empty bin } // hash 居然可以等於 MOVED，這個需要到後面才能看明白，不過從名字上也能猜到，肯定是因為在擴容 else if ((fh = f.hash) == MOVED) // 幫助資料遷移，這個等到看完資料遷移部分的介紹後，再理解這個就很簡單了 tab = helpTransfer(tab, f); else { // 到這裡就是說，f 是該位置的頭結點，而且不為空 V oldVal = null; // 獲取陣列該位置的頭結點的監視器鎖 synchronized (f) { if (tabAt(tab, i) == f) { if (fh >= 0) { // 頭結點的 hash 值大於 0，說明是連結串列 // 用於累加，記錄連結串列的長度 binCount = 1; // 遍歷連結串列 for (Node<K,V> e = f;; ++binCount) { K ek; // 如果發現了"相等"的 key，判斷是否要進行值覆蓋，然後也就可以 break 了 if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { oldVal = e.val; if (!onlyIfAbsent) e.val = value; break; } // 到了連結串列的最末端，將這個新值放到連結串列的最後面 Node<K,V> pred = e; if ((e = e.next) == null) { pred.next = new Node<K,V>(hash, key, value, null); break; } } } else if (f instanceof TreeBin) { // 紅黑樹 Node<K,V> p; binCount = 2; // 呼叫紅黑樹的插值方法插入新節點 if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) != null) { oldVal = p.val; if (!onlyIfAbsent) p.val = value; } } } } if (binCount != 0) { // 判斷是否要將連結串列轉換為紅黑樹，臨界值和 HashMap 一樣，也是 8 if (binCount >= TREEIFY_THRESHOLD) // 這個方法和 HashMap 中稍微有一點點不同，那就是它不是一定會進行紅黑樹轉換， // 如果當前陣列的長度小於 64，那麼會選擇進行陣列擴容，而不是轉換為紅黑樹 // 具體原始碼我們就不看了，擴容部分後面說 treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } // addCount(1L, binCount); return null; }
按照上面的原始碼，我們可以確定put整個流程如下：
- 判空；ConcurrentHashMap的key、value都不允許為null
- 計算hash。利用方法計算hash值。
- 遍歷table，進行節點插入操作，過程如下：
- 如果table為空，則表示ConcurrentHashMap還沒有初始化，則進行初始化操作：initTable()
- 根據hash值獲取節點的位置i，若該位置為空，則直接插入，這個過程是不需要加鎖的。計算f位置：i=(n – 1) & hash
- 如果檢測到fh = f.hash == -1，則f是ForwardingNode節點，表示有其他執行緒正在進行擴容操作，則幫助執行緒一起進行擴容操作
- 如果f.hash >= 0 表示是連結串列結構，則遍歷連結串列，如果存在當前key節點則替換value，否則插入到連結串列尾部。如果f是TreeBin型別節點，則按照紅黑樹的方法更新或者增加節點
- 若連結串列長度 > TREEIFY_THRESHOLD(預設是8)，則將連結串列轉換為紅黑樹結構
- 呼叫addCount方法，ConcurrentHashMap的size + 1
這裡整個put操作已經完成。
四、get()
public V get(Object key) { Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek; // 計算hash int h = spread(key.hashCode()); if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) { // 搜尋到的節點key與傳入的key相同且不為null,直接返回這個節點 if ((eh = e.hash) == h) { if ((ek = e.key) == key || (ek != null && key.equals(ek))) return e.val; } // 樹 else if (eh < 0) return (p = e.find(h, key)) != null ? p.val : null; // 連結串列，遍歷 while ((e = e.next) != null) { if (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek)))) return e.val; } } return null; }
get操作：
- 計算hash值
- 判斷table是否為空，如果為空，直接返回null
- 根據hash值獲取table中的Node節點（tabAt(tab, (n – 1) & h)），然後根據連結串列或者樹形方式找到相對應的節點，返回其value值。
五、擴容
// 首先要說明的是，方法引數 size 傳進來的時候就已經翻了倍了 private final void tryPresize(int size) { // c：size 的 1.5 倍，再加 1，再往上取最近的 2 的 n 次方。 int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY : tableSizeFor(size + (size >>> 1) + 1); int sc; while ((sc = sizeCtl) >= 0) { Node<K,V>[] tab = table; int n; // 這個 if 分支和之前說的初始化陣列的程式碼基本上是一樣的，在這裡，我們可以不用管這塊程式碼 if (tab == null || (n = tab.length) == 0) { n = (sc > c) ? sc : c; if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { try { if (table == tab) { @SuppressWarnings("unchecked") Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; table = nt; sc = n - (n >>> 2); // 0.75 * n } } finally { sizeCtl = sc; } } } else if (c <= sc || n >= MAXIMUM_CAPACITY) break; else if (tab == table) { // 我沒看懂 rs 的真正含義是什麼，不過也關係不大 int rs = resizeStamp(n); if (sc < 0) { Node<K,V>[] nt; if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || (nt = nextTable) == null || transferIndex <= 0) break; // 2. 用 CAS 將 sizeCtl 加 1，然後執行 transfer 方法 // 此時 nextTab 不為 null if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) transfer(tab, nt); } // 1. 將 sizeCtl 設定為 (rs << RESIZE_STAMP_SHIFT) + 2) // 我是沒看懂這個值真正的意義是什麼？不過可以計算出來的是，結果是一個比較大的負數 // 呼叫 transfer 方法，此時 nextTab 引數為 null else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)) transfer(tab, null); } } }
這個方法的核心在於 sizeCtl 值的操作，首先將其設定為一個負數，然後執行 transfer(tab, null)，再下一個迴圈將 sizeCtl 加 1，並執行 transfer(tab, nt)，之後可能是繼續 sizeCtl 加 1，並執行 transfer(tab, nt)。
所以，可能的操作就是執行 1 次 transfer(tab, null) + 多次 transfer(tab, nt)，這裡怎麼結束迴圈的需要看完 transfer 原始碼才清楚。
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) { int n = tab.length, stride; // stride 在單核下直接等於 n，多核模式下為 (n>>>3)/NCPU，最小值是 16 // stride 可以理解為”步長“，有 n 個位置是需要進行遷移的， // 將這 n 個任務分為多個任務包，每個任務包有 stride 個任務 if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) stride = MIN_TRANSFER_STRIDE; // subdivide range // 如果 nextTab 為 null，先進行一次初始化 // 前面我們說了，外圍會保證第一個發起遷移的執行緒呼叫此方法時，引數 nextTab 為 null // 之後參與遷移的執行緒呼叫此方法時，nextTab 不會為 null if (nextTab == null) { try { // 容量翻倍 Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1]; nextTab = nt; } catch (Throwable ex) { // try to cope with OOME sizeCtl = Integer.MAX_VALUE; return; } // nextTable 是 ConcurrentHashMap 中的屬性 nextTable = nextTab; // transferIndex 也是 ConcurrentHashMap 的屬性，用於控制遷移的位置 transferIndex = n; } int nextn = nextTab.length; // ForwardingNode 翻譯過來就是正在被遷移的 Node // 這個構造方法會生成一個Node，key、value 和 next 都為 null，關鍵是 hash 為 MOVED // 後面我們會看到，原陣列中位置 i 處的節點完成遷移工作後， // 就會將位置 i 處設定為這個 ForwardingNode，用來告訴其他執行緒該位置已經處理過了 // 所以它其實相當於是一個標誌。 ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab); // advance 指的是做完了一個位置的遷移工作，可以準備做下一個位置的了 boolean advance = true; boolean finishing = false; // to ensure sweep before committing nextTab /* * 下面這個 for 迴圈，最難理解的在前面，而要看懂它們，應該先看懂後面的，然後再倒回來看 * */ // i 是位置索引，bound 是邊界，注意是從後往前 for (int i = 0, bound = 0;;) { Node<K,V> f; int fh; // 下面這個 while 真的是不好理解 // advance 為 true 表示可以進行下一個位置的遷移了 // 簡單理解結局：i 指向了 transferIndex，bound 指向了 transferIndex-stride while (advance) { int nextIndex, nextBound; if (--i >= bound || finishing) advance = false; // 將 transferIndex 值賦給 nextIndex // 這裡 transferIndex 一旦小於等於 0，說明原陣列的所有位置都有相應的執行緒去處理了 else if ((nextIndex = transferIndex) <= 0) { i = -1; advance = false; } else if (U.compareAndSwapInt (this, TRANSFERINDEX, nextIndex, nextBound = (nextIndex > stride ? nextIndex - stride : 0))) { // 看括號中的程式碼，nextBound 是這次遷移任務的邊界，注意，是從後往前 bound = nextBound; i = nextIndex - 1; advance = false; } } if (i < 0 || i >= n || i + n >= nextn) { int sc; if (finishing) { // 所有的遷移操作已經完成 nextTable = null; // 將新的 nextTab 賦值給 table 屬性，完成遷移 table = nextTab; // 重新計算 sizeCtl：n 是原陣列長度，所以 sizeCtl 得出的值將是新陣列長度的 0.75 倍 sizeCtl = (n << 1) - (n >>> 1); return; } // 之前我們說過，sizeCtl 在遷移前會設定為 (rs << RESIZE_STAMP_SHIFT) + 2 // 然後，每有一個執行緒參與遷移就會將 sizeCtl 加 1， // 這裡使用 CAS 操作對 sizeCtl 進行減 1，代表做完了屬於自己的任務 if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { // 任務結束，方法退出 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) return; // 到這裡，說明 (sc - 2) == resizeStamp(n) << RESIZE_STAMP_SHIFT， // 也就是說，所有的遷移任務都做完了，也就會進入到上面的 if(finishing){} 分支了 finishing = advance = true; i = n; // recheck before commit } } // 如果位置 i 處是空的，沒有任何節點，那麼放入剛剛初始化的 ForwardingNode ”空節點“ else if ((f = tabAt(tab, i)) == null) advance = casTabAt(tab, i, null, fwd); // 該位置處是一個 ForwardingNode，代表該位置已經遷移過了 else if ((fh = f.hash) == MOVED) advance = true; // already processed else { // 對陣列該位置處的結點加鎖，開始處理陣列該位置處的遷移工作 synchronized (f) { if (tabAt(tab, i) == f) { Node<K,V> ln, hn; // 頭結點的 hash 大於 0，說明是連結串列的 Node 節點 if (fh >= 0) { // 下面這一塊和 Java7 中的 ConcurrentHashMap 遷移是差不多的， // 需要將連結串列一分為二， // 找到原連結串列中的 lastRun，然後 lastRun 及其之後的節點是一起進行遷移的 // lastRun 之前的節點需要進行克隆，然後分到兩個連結串列中 int runBit = fh & n; Node<K,V> lastRun = f; for (Node<K,V> p = f.next; p != null; p = p.next) { int b = p.hash & n; if (b != runBit) { runBit = b; lastRun = p; } } if (runBit == 0) { ln = lastRun; hn = null; } else { hn = lastRun; ln = null; } for (Node<K,V> p = f; p != lastRun; p = p.next) { int ph = p.hash; K pk = p.key; V pv = p.val; if ((ph & n) == 0) ln = new Node<K,V>(ph, pk, pv, ln); else hn = new Node<K,V>(ph, pk, pv, hn); } // 其中的一個連結串列放在新陣列的位置 i setTabAt(nextTab, i, ln); // 另一個連結串列放在新陣列的位置 i+n setTabAt(nextTab, i + n, hn); // 將原陣列該位置處設定為 fwd，代表該位置已經處理完畢， // 其他執行緒一旦看到該位置的 hash 值為 MOVED，就不會進行遷移了 setTabAt(tab, i, fwd); // advance 設定為 true，代表該位置已經遷移完畢 advance = true; } else if (f instanceof TreeBin) { // 紅黑樹的遷移 TreeBin<K,V> t = (TreeBin<K,V>)f; TreeNode<K,V> lo = null, loTail = null; TreeNode<K,V> hi = null, hiTail = null; int lc = 0, hc = 0; for (Node<K,V> e = t.first; e != null; e = e.next) { int h = e.hash; TreeNode<K,V> p = new TreeNode<K,V> (h, e.key, e.val, null, null); if ((h & n) == 0) { if ((p.prev = loTail) == null) lo = p; else loTail.next = p; loTail = p; ++lc; } else { if ((p.prev = hiTail) == null) hi = p; else hiTail.next = p; hiTail = p; ++hc; } } // 如果一分為二後，節點數少於 8，那麼將紅黑樹轉換回連結串列 ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) : (hc != 0) ? new TreeBin<K,V>(lo) : t; hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) : (lc != 0) ? new TreeBin<K,V>(hi) : t; // 將 ln 放置在新陣列的位置 i setTabAt(nextTab, i, ln); // 將 hn 放置在新陣列的位置 i+n setTabAt(nextTab, i + n, hn); // 將原陣列該位置處設定為 fwd，代表該位置已經處理完畢， // 其他執行緒一旦看到該位置的 hash 值為 MOVED，就不會進行遷移了 setTabAt(tab, i, fwd); // advance 設定為 true，代表該位置已經遷移完畢 advance = true; } } } } } }
參考資料 / 相關推薦：
【死磕Java併發】—–J.U.C之Java併發容器：ConcurrentHashMap
Java7/8 中的 HashMap 和 ConcurrentHashMap 全解析