原文鏈接：http://www.jianshu.com/p/7f42ba895a64

前言

在Java1.5中，并發編程大師Doug Lea給我們帶來了concurrent包，而該包中提供的ConcurrentHashMap是線程安全并且高效的HashMap，本節我們就來研究下ConcurrentHashMap是如何保證線程安全的同時又能高效的操作。

1.為何用ConcurrentHashMap

在并發編程中使用HashMap可能會導致死循環，而使用線程安全的HashTable效率又低下。

線程不安全的HashMap

在多線程環境下，使用HashMap進行put操作會引起死循環，導致CPU利用率接近100%，所以在并發情況下不能使用HashMap，如以下代碼會導致死循環：

final HashMap<String, String> map = new HashMap<String, String>(2); Thread t = new Thread(new Runnable() {@Overridepublic void run() {for (int i = 0; i < 10000; i++) {new Thread(new Runnable() {@Overridepublic void run() {map.put(UUID.randomUUID().toString(), "");}}, "moon" + i).start();}} }, "ftf"); t.start(); t.join();

HashMap在并發執行put操作是會引起死循環，是因為多線程會導致HashMap的Entry鏈表形成環形數據結構，一旦形成環形數據結構，Entry的next節點永遠不為空，就會產生死循環。

效率低下的HashTable

HashTable使用synchronized來保證線程的安全，但是在線程競爭激烈的情況下HashTable的效率非常低下。當一個線程訪問HashTable的同步方法，其他方法訪問HashTable的同步方法時，會進入阻塞或者輪詢狀態。如果線程1使用put進行元素添加，線程2不但不能用put方法添加于元素同是也無法用get方法來獲取元素，所以競爭越激烈效率越低。

ConcurrentHashMap的鎖分段技術

HashTable容器在競爭激烈的并發環境效率低下的原因是所有訪問HashTable的線程都必須競爭同一把鎖，假如容器有多把鎖，每一把鎖用于鎖住容器中一部分數據，那么多線程訪問容器里不同數據段的數據時，線程間就不會存在鎖競爭，從而可以有效提高并發訪問率，這就是ConcurrentHashMap的鎖分段技術。將數據分成一段一段的存儲，然后給每一段數據配一把鎖，當一個線程占用鎖訪問其中一段數據的時候，其他段的數據也能被其他線程訪問。

2.Java1.6的ConcurrentHashMap的結構

首先來看看 Java1.6中ConcurrentHashMap的類圖：

ConcurrentHashMap是由Segment數組結構和HashEntry數組結構組成。Segment是一種可重入鎖ReentrantLock，在ConcurrentHashMap里扮演鎖的角色，HashEntry則用于存儲鍵值對數據。一個ConcurrentHashMap里包含一個Segment數組，Segment的結構和HashMap類似，是一種數組和鏈表結構， 一個Segment里包含一個HashEntry數組，每個HashEntry是一個鏈表結構的元素， 每個Segment守護者一個HashEntry數組里的元素,當對HashEntry數組的數據進行修改時，必須首先獲得它對應的Segment鎖。

3.java1.8的ConcurrentHashMap源碼分析

重要的內部類

從Java1.7 版本開始 ConcurrentHashMap 不再采用 Segment 實現，而是改用 Node，Node 是一個鏈表的結構，每個節點可以引用到下一個節點(next)。

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

• TreeNode類
  樹節點類，另外一個核心的數據結構。 當鏈表長度過長的時候，會轉換為TreeNode。
  但是與HashMap不相同的是，它并不是直接轉換為紅黑樹，而是把這些結點包裝成TreeNode放在TreeBin對象中，由TreeBin完成對紅黑樹的包裝。
  而且TreeNode在ConcurrentHashMap繼承自Node類，而并非HashMap中的集成自LinkedHashMap.Entry

• TreeBin
  這個類并不負責包裝用戶的key、value信息，而是包裝的很多TreeNode節點。它代替了TreeNode的根節點，也就是說在實際的ConcurrentHashMap“數組”中，存放的是TreeBin對象，而不是TreeNode對象，這是與HashMap的區別。

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

構造函數

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 binsinitialCapacity = concurrencyLevel; // as estimated threadslong size = (long)(1.0 + (long)initialCapacity / loadFactor);int cap = (size >= (long)MAXIMUM_CAPACITY) ?MAXIMUM_CAPACITY : tableSizeFor((int)size);this.sizeCtl = cap;}

Java1.8版本的 ConcurrentHashMap 在構造函數中不會初始化 Node 數組，而是第一次 put 操作的時候初始化。
整個 Map 第一次 put 的時候，map 中用于存放數據的 Node[] 還是null。

Unsafe與CAS

在ConcurrentHashMap中，大量使用了U.compareAndSwapXXX的方法，這個方法是利用一個CAS算法實現無鎖化的修改值的操作，他可以大大降低鎖代理的性能消耗。這個算法的基本思想就是不斷地去比較當前內存中的變量值與你指定的一個變量值是否相等，如果相等，則接受你指定的修改的值，否則拒絕你的操作。因為當前線程中的值已經不是最新的值，你的修改很可能會覆蓋掉其他線程修改的結果。這一點與樂觀鎖，SVN的思想是比較類似的。
unsafe代碼塊控制了一些屬性的修改工作，比如最常用的SIZECTL 。 在這一版本的concurrentHashMap中，大量應用來的CAS方法進行變量、屬性的修改工作。 利用CAS進行無鎖操作，可以大大提高性能。

初始化函數initTable

調用ConcurrentHashMap的構造方法僅僅是設置了一些參數而已，而整個table的初始化是在向ConcurrentHashMap中插入元素的時候發生的。如調用put、computeIfAbsent、compute、merge等方法的時候，調用時機是檢查table==null。
初始化方法主要應用了關鍵屬性sizeCtl 如果這個值 < 0，表示其他線程正在進行初始化，就放棄這個操作。
在這也可以看出ConcurrentHashMap的初始化只能由一個線程完成。如果獲得了初始化權限，就用CAS方法將sizeCtl置為-1，防止其他線程進入。初始化數組后，將sizeCtl的值改為0.75*n

sizeCtl含義
1.負數代表正在進行初始化或擴容操作
2.-1代表正在初始化
3.-N 表示有N-1個線程正在進行擴容操作
4.正數或0代表hash表還沒有被初始化，這個數值表示初始化或下一次進行擴容的大小，這一點類似于擴容閾值的概念。還后面可以看到，它的值始終是當前ConcurrentHashMap容量的0.75倍，這與loadfactor是對應的。

/** * 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 spin else 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; }

擴容方法transfer

支持多線程進行擴容操作，并沒有加鎖 ，這樣做的目的不僅僅是為了滿足concurrent的要求，而是希望利用并發處理去減少擴容帶來的時間影響。
單線程擴容的大體思想就是遍歷、復制的過程。首先根據運算得到需要遍歷的次數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倍，完成擴容。
多線程遍歷節點，處理了一個節點，就把對應點的值set為forward，另一個線程看到forward，就向后繼續遍歷，再加上給節點上鎖的機制，就完成了多線程的控制。這樣交叉就完成了復制工作。而且還很好的解決了線程安全的問題。

/*** 一個過渡的table表 只有在擴容的時候才會使用*/private transient volatile Node<K,V>[] nextTable;/*** Moves and/or copies the nodes in each bin to new table. See* above for explanation.*/private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {int n = tab.length, stride;if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)stride = MIN_TRANSFER_STRIDE; // subdivide rangeif (nextTab == null) { // initiatingtry {@SuppressWarnings("unchecked")Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];//構造一個nextTable對象 它的容量是原來的兩倍nextTab = nt;} catch (Throwable ex) { // try to cope with OOMEsizeCtl = Integer.MAX_VALUE;return;}nextTable = nextTab;transferIndex = n;}int nextn = nextTab.length;ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);//構造一個連節點指針 用于標志位boolean advance = true;//并發擴容的關鍵屬性 如果等于true 說明這個節點已經處理過boolean finishing = false; // to ensure sweep before committing nextTabfor (int i = 0, bound = 0;;) {Node<K,V> f; int fh;//這個while循環體的作用就是在控制i-- 通過i--可以依次遍歷原hash表中的節點while (advance) {int nextIndex, nextBound;if (--i >= bound || finishing)advance = false;else if ((nextIndex = transferIndex) <= 0) {i = -1;advance = false;}else if (U.compareAndSwapInt(this, TRANSFERINDEX, nextIndex,nextBound = (nextIndex > stride ?nextIndex - stride : 0))) {bound = nextBound;i = nextIndex - 1;advance = false;}}if (i < 0 || i >= n || i + n >= nextn) {int sc;if (finishing) {//如果所有的節點都已經完成復制工作 就把nextTable賦值給table 清空臨時對象nextTablenextTable = null;table = nextTab;sizeCtl = (n << 1) - (n >>> 1);//擴容閾值設置為原來容量的1.5倍 依然相當于現在容量的0.75倍return;}//利用CAS方法更新這個擴容閾值，在這里面sizectl值減一，說明新加入一個線程參與到擴容操作if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)return;finishing = advance = true;i = n; // recheck before commit}}//如果遍歷到的節點為空 則放入ForwardingNode指針else if ((f = tabAt(tab, i)) == null)advance = casTabAt(tab, i, null, fwd);//如果遍歷到ForwardingNode節點 說明這個點已經被處理過了 直接跳過 這里是控制并發擴容的核心else if ((fh = f.hash) == MOVED)advance = true; // already processedelse {//節點上鎖synchronized (f) {if (tabAt(tab, i) == f) {Node<K,V> ln, hn;//如果fh>=0 證明這是一個Node節點if (fh >= 0) {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);elsehn = new Node<K,V>(ph, pk, pv, hn);}//在nextTable的i位置上插入一個鏈表setTabAt(nextTab, i, ln);//在nextTable的i+n的位置上插入另一個鏈表setTabAt(nextTab, i + n, hn);//在table的i位置上插入forwardNode節點 表示已經處理過該節點setTabAt(tab, i, fwd);//設置advance為true 返回到上面的while循環中 就可以執行i--操作advance = true;}//對TreeBin對象進行處理 與上面的過程類似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;elseloTail.next = p;loTail = p;++lc;}else {if ((p.prev = hiTail) == null)hi = p;elsehiTail.next = p;hiTail = p;++hc;}}//如果擴容后已經不再需要tree的結構 反向轉換為鏈表結構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;//在nextTable的i位置上插入一個鏈表 setTabAt(nextTab, i, ln);//在nextTable的i+n的位置上插入另一個鏈表setTabAt(nextTab, i + n, hn);//在table的i位置上插入forwardNode節點 表示已經處理過該節點setTabAt(tab, i, fwd);//設置advance為true 返回到上面的while循環中 就可以執行i--操作advance = true;}}}}}}

put函數

put方法依然沿用HashMap的put方法的思想，根據hash值計算這個新插入的點在table中的位置i，如果i位置是空的，直接放進去，否則進行判斷，如果i位置是樹節點，按照樹的方式插入新的節點，否則把i插入到鏈表的末尾。ConcurrentHashMap中依然沿用這個思想，有一個最重要的不同點就是ConcurrentHashMap不允許key或value為null值。另外由于涉及到多線程，put方法就要復雜一點。在多線程中可能有以下兩個情況：

如果一個或多個線程正在對ConcurrentHashMap進行擴容操作，當前線程也要進入擴容的操作中。這個擴容的操作之所以能被檢測到，是因為transfer方法中在空結點上插入forward節點，如果檢測到需要插入的位置被forward節點占有，就幫助進行擴容。

如果檢測到要插入的節點是非空且不是forward節點，就對這個節點加鎖，這樣就保證了線程安全。盡管這個有一些影響效率，但是還是會比hashTable的synchronized要好得多。

整體流程就是首先定義不允許key或value為null的情況放入 對于每一個放入的值，首先利用spread方法對key的hashcode進行一次hash計算，由此來確定這個值在table中的位置。如果這個位置是空的，那么直接放入，而且不需要加鎖操作。
如果這個位置存在結點，說明發生了hash碰撞，首先判斷這個節點的類型。如果是鏈表節點（fh>0）,則得到的結點就是hash值相同的節點組成的鏈表的頭節點。需要依次向后遍歷確定這個新加入的值所在位置。如果遇到hash值與key值都與新加入節點是一致的情況，則只需要更新value值即可。否則依次向后遍歷，直到鏈表尾插入這個結點。 如果加入這個節點以后鏈表長度大于8，就把這個鏈表轉換成紅黑樹。如果這個節點的類型已經是樹節點的話，直接調用樹節點的插入方法進行插入新的值。

final V putVal(K key, V value, boolean onlyIfAbsent) {//不允許 key或value為null 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;// 第一次 put 操作的時候初始化，如果table為空的話，初始化table if (tab == null || (n = tab.length) == 0)tab = initTable();//根據hash值計算出在table里面的位置 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {// 根據對應的key hash 到具體的索引，如果該索引對應的 Node 為 null，則采用 CAS 操作更新整個 table// 如果這個位置沒有值 ，直接放進去，不需要加鎖 if (casTabAt(tab, i, null,new Node<K,V>(hash, key, value, null)))break; // no lock when adding to empty bin}//當遇到表連接點時，需要進行整合表的操作 else if ((fh = f.hash) == MOVED)tab = helpTransfer(tab, f);else {V oldVal = null;// 結點上鎖，只是對鏈表頭結點作鎖操作synchronized (f) {if (tabAt(tab, i) == f) {//fh > 0 說明這個節點是一個鏈表的節點 不是樹的節點 if (fh >= 0) {binCount = 1;//在這里遍歷鏈表所有的結點 for (Node<K,V> e = f;; ++binCount) {K ek;//如果hash值和key值相同 則修改對應結點的value值 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) {// 如果這個鏈表結點達到了臨界值8，那么把這個鏈表轉換成紅黑樹if (binCount >= TREEIFY_THRESHOLD)treeifyBin(tab, i);if (oldVal != null)return oldVal;break;}}}//將當前ConcurrentHashMap的元素數量+1，table的擴容是在這里發生的addCount(1L, binCount);return null;}

協助擴容函數helpTransfer

這是一個協助擴容的方法。這個方法被調用的時候，當前ConcurrentHashMap一定已經有了nextTable對象，首先拿到這個nextTable對象，調用上面講到的transfer方法來進行擴容。

final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {Node<K,V>[] nextTab; int sc;if (tab != null && (f instanceof ForwardingNode) &&(nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {int rs = resizeStamp(tab.length);//計算一個操作校驗碼while (nextTab == nextTable && table == tab &&(sc = sizeCtl) < 0) {if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||sc == rs + MAX_RESIZERS || transferIndex <= 0)break;if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {transfer(tab, nextTab);break;}}return nextTab;}return table; }

紅黑樹轉換

在putVal函數中，treeifyBin是在鏈表長度達到一定閾值（8）后轉換成紅黑樹的函數。 但是并不是直接轉換，而是進行一次容量判斷，如果容量沒有達到轉換的要求，直接進行擴容操作并返回；如果滿足條件才將鏈表的結構轉換為TreeBin ，這與HashMap不同的是，它并沒有把TreeNode直接放入紅黑樹，而是利用了TreeBin這個小容器來封裝所有的TreeNode。

private final void treeifyBin(Node<K,V>[] tab, int index) {Node<K,V> b; int n, sc;if (tab != null) {if ((n = tab.length) < MIN_TREEIFY_CAPACITY)tryPresize(n << 1);else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {synchronized (b) {if (tabAt(tab, index) == b) {TreeNode<K,V> hd = null, tl = null;for (Node<K,V> e = b; e != null; e = e.next) {TreeNode<K,V> p =new TreeNode<K,V>(e.hash, e.key, e.val,null, null);if ((p.prev = tl) == null)hd = p;elsetl.next = p;tl = p;}setTabAt(tab, index, new TreeBin<K,V>(hd));}}}} }

get方法

給定一個key來確定value的時候，必須滿足兩個條件 key相同 hash值相同，對于節點可能在鏈表或樹上的情況，需要分別去查找。

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());//根據hash值確定節點位置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;}//如果eh<0 說明這個節點在樹上 直接尋找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; }

總結

以上是生活随笔為你收集整理的ConcurrentHashMap的实现原理和源码分析的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。