(本文使用的源碼都是基于JDK1.8的)

(基于JDK1.6的版本可以參考這篇文章：https://blog.csdn.net/a745233700/article/details/83120464)

ConcurrentHashMap是conccurrent家族中的一個(gè)類，由于它可以高效地支持并發(fā)操作，以及被廣泛使用，經(jīng)典的開源框架Spring的底層數(shù)據(jù)結(jié)構(gòu)就是使用ConcurrentHashMap實(shí)現(xiàn)的。與同是線程安全的老大哥HashTable相比，它已經(jīng)更勝一籌，因此它的鎖更加細(xì)化，而不是像HashTable一樣為幾乎每個(gè)方法都添加了synchronized鎖，這樣的鎖無疑會影響到性能。

本文的分析的源碼是JDK8的版本，與JDK6的版本有很大的差異。實(shí)現(xiàn)線程安全的思想也已經(jīng)完全變了，它摒棄了Segment（鎖段）的概念，而是啟用了一種全新的方式實(shí)現(xiàn),利用CAS算法。它沿用了與它同時(shí)期的HashMap版本的思想，底層依然由“數(shù)組”+鏈表+紅黑樹的方式思想，但是為了做到并發(fā)，又增加了很多輔助的類，例如TreeBin，Traverser等對象內(nèi)部類。

---

?

一、重要的屬性：

首先來看幾個(gè)重要的屬性，與HashMap相同的就不再介紹了，這里重點(diǎn)解釋一下sizeCtl這個(gè)屬性。可以說它是ConcurrentHashMap中出鏡率很高的一個(gè)屬性，因?yàn)樗且粋€(gè)控制標(biāo)識符，在不同的地方有不同用途，而且它的取值不同，也代表不同的含義。

• 負(fù)數(shù)代表正在進(jìn)行初始化或擴(kuò)容操作
• -1代表正在初始化
• -N 表示有N-1個(gè)線程正在進(jìn)行擴(kuò)容操作
• 正數(shù)或0代表hash表還沒有被初始化，這個(gè)數(shù)值表示初始化或下一次進(jìn)行擴(kuò)容的大小，這一點(diǎn)類似于擴(kuò)容閾值的概念。還后面可以看到，它的值始終是當(dāng)前ConcurrentHashMap容量的0.75倍，這與loadfactor是對應(yīng)的。

/*** 盛裝Node元素的數(shù)組 它的大小是2的整數(shù)次冪* 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表初始化或擴(kuò)容時(shí)的一個(gè)控制位標(biāo)識量。負(fù)數(shù)代表正在進(jìn)行初始化或擴(kuò)容操作-1代表正在初始化-N 表示有N-1個(gè)線程正在進(jìn)行擴(kuò)容操作正數(shù)或0代表hash表還沒有被初始化，這個(gè)數(shù)值表示初始化或下一次進(jìn)行擴(kuò)容的大小*/private transient volatile int sizeCtl; // 以下兩個(gè)是用來控制擴(kuò)容的時(shí)候 單線程進(jìn)入的變量/*** 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，表示這是一個(gè)forwardNode節(jié)點(diǎn)static final int TREEBIN = -2; // hash值是-2 表示這時(shí)一個(gè)TreeBin節(jié)點(diǎn)

---

?

二、重要的內(nèi)部類：

2.1、Node：

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

static class Node<K,V> implements Map.Entry<K,V> {final int hash;final K key;volatile V val;//帶有同步鎖的valuevolatile Node<K,V> next;//帶有同步鎖的next指針Node(int hash, K key, V val, Node<K,V> next) {this.hash = hash;this.key = key;this.val = val;this.next = next;}public final K getKey() { return key; }public final V getValue() { return val; }public final int hashCode() { return key.hashCode() ^ val.hashCode(); }public final String toString(){ return key + "=" + val; }//不允許直接改變value的值public final V setValue(V value) {throw new UnsupportedOperationException();}public final boolean equals(Object o) {Object k, v, u; Map.Entry<?,?> e;return ((o instanceof Map.Entry) &&(k = (e = (Map.Entry<?,?>)o).getKey()) != null &&(v = e.getValue()) != null &&(k == key || k.equals(key)) &&(v == (u = val) || v.equals(u)));}/*** Virtualized support for map.get(); overridden in subclasses.*/Node<K,V> find(int h, Object k) {Node<K,V> e = this;if (k != null) {do {K ek;if (e.hash == h &&((ek = e.key) == k || (ek != null && k.equals(ek))))return e;} while ((e = e.next) != null);}return null;}}這個(gè)Node內(nèi)部類與HashMap中定義的Node類很相似，但是有一些差別它對value和next屬性設(shè)置了volatile同步鎖它不允許調(diào)用setValue方法直接改變Node的value域它增加了find方法輔助map.get()方法

---

2.2、TreeNode：

樹節(jié)點(diǎn)類，另外一個(gè)核心的數(shù)據(jù)結(jié)構(gòu)。當(dāng)鏈表長度過長的時(shí)候，會轉(zhuǎn)換為TreeNode。但是與HashMap不相同的是，它并不是直接轉(zhuǎn)換為紅黑樹，而是把這些結(jié)點(diǎn)包裝成TreeNode放在TreeBin對象中，由TreeBin完成對紅黑樹的包裝。而且TreeNode在ConcurrentHashMap集成自Node類，而并非HashMap中的集成自LinkedHashMap.Entry<K,V>類，也就是說TreeNode帶有next指針，這樣做的目的是方便基于TreeBin的訪問。

---

2.3、TreeBin：

這個(gè)類并不負(fù)責(zé)包裝用戶的key、value信息，而是包裝的很多TreeNode節(jié)點(diǎn)。它代替了TreeNode的根節(jié)點(diǎn)，也就是說在實(shí)際的ConcurrentHashMap“數(shù)組”中，存放的是TreeBin對象，而不是TreeNode對象，這是與HashMap的區(qū)別。另外這個(gè)類還帶有了讀寫鎖。

這里僅貼出它的構(gòu)造方法。可以看到在構(gòu)造TreeBin節(jié)點(diǎn)時(shí)，僅僅指定了它的hash值為TREEBIN常量，這也就是個(gè)標(biāo)識為。同時(shí)也看到我們熟悉的紅黑樹構(gòu)造方法。

/*** Creates bin with initial set of nodes headed by b.*/TreeBin(TreeNode<K,V> b) {super(TREEBIN, null, null, null);this.first = b;TreeNode<K,V> r = null;for (TreeNode<K,V> x = b, next; x != null; x = next) {next = (TreeNode<K,V>)x.next;x.left = x.right = null;if (r == null) {x.parent = null;x.red = false;r = x;}else {K k = x.key;int h = x.hash;Class<?> kc = null;for (TreeNode<K,V> p = r;;) {int dir, ph;K pk = p.key;if ((ph = p.hash) > h)dir = -1;else if (ph < h)dir = 1;else if ((kc == null &&(kc = comparableClassFor(k)) == null) ||(dir = compareComparables(kc, k, pk)) == 0)dir = tieBreakOrder(k, pk);TreeNode<K,V> xp = p;if ((p = (dir <= 0) ? p.left : p.right) == null) {x.parent = xp;if (dir <= 0)xp.left = x;elsexp.right = x;r = balanceInsertion(r, x);break;}}}}this.root = r;assert checkInvariants(root);}

---

2.4、ForwardingNode：

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

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

---

?

三、Unsafe與CAS：

在ConcurrentHashMap中，隨處可以看到U, 大量使用了U.compareAndSwapXXX的方法，這個(gè)方法是利用一個(gè)CAS算法實(shí)現(xiàn)無鎖化的修改值的操作，他可以大大降低鎖代理的性能消耗。這個(gè)算法的基本思想就是不斷地去比較當(dāng)前內(nèi)存中的變量值與你指定的一個(gè)變量值是否相等，如果相等，則接受你指定的修改的值，否則拒絕你的操作。因?yàn)楫?dāng)前線程中的值已經(jīng)不是最新的值，你的修改很可能會覆蓋掉其他線程修改的結(jié)果。這一點(diǎn)與樂觀鎖，SVN的思想是比較類似的。

3.1、unsafe靜態(tài)塊：

unsafe代碼塊控制了一些屬性的修改工作，比如最常用的SIZECTL 。? 在這一版本的concurrentHashMap中，大量應(yīng)用來的CAS方法進(jìn)行變量、屬性的修改工作。? 利用CAS進(jìn)行無鎖操作，可以大大提高性能。

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);}}

3.2、三個(gè)核心方法：

ConcurrentHashMap定義了三個(gè)原子操作，用于對指定位置的節(jié)點(diǎn)進(jìn)行操作。正是這些原子操作保證了ConcurrentHashMap的線程安全。

@SuppressWarnings("unchecked")//獲得在i位置上的Node節(jié)點(diǎn)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算法設(shè)置i位置上的Node節(jié)點(diǎn)。之所以能實(shí)現(xiàn)并發(fā)是因?yàn)樗付嗽瓉磉@個(gè)節(jié)點(diǎn)的值是多少//在CAS算法中，會比較內(nèi)存中的值與你指定的這個(gè)值是否相等，如果相等才接受你的修改，否則拒絕你的修改//因此當(dāng)前線程中的值并不是最新的值，這種修改可能會覆蓋掉其他線程的修改結(jié)果 有點(diǎn)類似于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方法設(shè)置節(jié)點(diǎn)位置的值static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);}

---

?

四、初始化方法initTable：

對于ConcurrentHashMap來說，調(diào)用它的構(gòu)造方法僅僅是設(shè)置了一些參數(shù)而已。而整個(gè)table的初始化是在向ConcurrentHashMap中插入元素的時(shí)候發(fā)生的。如調(diào)用put、computeIfAbsent、compute、merge等方法的時(shí)候，調(diào)用時(shí)機(jī)是檢查table==null。

初始化方法主要應(yīng)用了關(guān)鍵屬性sizeCtl 如果這個(gè)值〈0，表示其他線程正在進(jìn)行初始化，就放棄這個(gè)操作。在這也可以看出ConcurrentHashMap的初始化只能由一個(gè)線程完成。如果獲得了初始化權(quán)限，就用CAS方法將sizeCtl置為-1，防止其他線程進(jìn)入。初始化數(shù)組后，將sizeCtl的值改為0.75*n
?

/*** 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<0表示有其他線程正在進(jìn)行初始化操作，把線程掛起。對于table的初始化工作，只能有一個(gè)線程在進(jìn)行。if ((sc = sizeCtl) < 0)Thread.yield(); // lost initialization race; just spinelse if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {//利用CAS方法把sizectl的值置為-1 表示本線程正在進(jìn)行初始化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);//相當(dāng)于0.75*n 設(shè)置一個(gè)擴(kuò)容的閾值}} finally {sizeCtl = sc;}break;}}return tab;}

---

?

五、擴(kuò)容方法transfer：

當(dāng)ConcurrentHashMap容量不足的時(shí)候，需要對table進(jìn)行擴(kuò)容。這個(gè)方法的基本思想跟HashMap是很像的，但是由于它是支持并發(fā)擴(kuò)容的，所以要復(fù)雜的多。原因是它支持多線程進(jìn)行擴(kuò)容操作，而并沒有加鎖。我想這樣做的目的不僅僅是為了滿足concurrent的要求，而是希望利用并發(fā)處理去減少擴(kuò)容帶來的時(shí)間影響。因?yàn)樵跀U(kuò)容的時(shí)候，總是會涉及到從一個(gè)“數(shù)組”到另一個(gè)“數(shù)組”拷貝的操作，如果這個(gè)操作能夠并發(fā)進(jìn)行，那真真是極好的了。

整個(gè)擴(kuò)容操作分為兩個(gè)部分:

• 第一部分是構(gòu)建一個(gè)nextTable,它的容量是原來的兩倍，這個(gè)操作是單線程完成的。這個(gè)單線程的保證是通過RESIZE_STAMP_SHIFT這個(gè)常量經(jīng)過一次運(yùn)算來保證的，這個(gè)地方在后面會有提到；
• 第二個(gè)部分就是將原來table中的元素復(fù)制到nextTable中，這里允許多線程進(jìn)行操作

先來看一下單線程是如何完成的：

它的大體思想就是遍歷、復(fù)制的過程。首先根據(jù)運(yùn)算得到需要遍歷的次數(shù)i，然后利用tabAt方法獲得i位置的元素：

• 如果這個(gè)位置為空，就在原table中的i位置放入forwardNode節(jié)點(diǎn)，這個(gè)也是觸發(fā)并發(fā)擴(kuò)容的關(guān)鍵點(diǎn)；
• 如果這個(gè)位置是Node節(jié)點(diǎn)（fh>=0），如果它是一個(gè)鏈表的頭節(jié)點(diǎn)，就構(gòu)造一個(gè)反序鏈表，把他們分別放在nextTable的i和i+n的位置上;
• 如果這個(gè)位置是TreeBin節(jié)點(diǎn)（fh<0），也做一個(gè)反序處理，并且判斷是否需要untreefi，把處理的結(jié)果分別放在nextTable的i和i+n的位置上;
• 遍歷過所有的節(jié)點(diǎn)以后就完成了復(fù)制工作，這時(shí)讓nextTable作為新的table，并且更新sizeCtl為新容量的0.75倍 ，完成擴(kuò)容。

再看一下多線程是如何完成的：
在代碼的69行有一個(gè)判斷，如果遍歷到的節(jié)點(diǎn)是forward節(jié)點(diǎn)，就向后繼續(xù)遍歷，再加上給節(jié)點(diǎn)上鎖的機(jī)制，就完成了多線程的控制。多線程遍歷節(jié)點(diǎn)，處理了一個(gè)節(jié)點(diǎn)，就把對應(yīng)點(diǎn)的值set為forward，另一個(gè)線程看到forward，就向后遍歷。這樣交叉就完成了復(fù)制工作。而且還很好的解決了線程安全的問題。 這個(gè)方法的設(shè)計(jì)實(shí)在是讓我膜拜。

/*** 一個(gè)過渡的table表 只有在擴(kuò)容的時(shí)候才會使用*/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];//構(gòu)造一個(gè)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);//構(gòu)造一個(gè)連節(jié)點(diǎn)指針 用于標(biāo)志位boolean advance = true;//并發(fā)擴(kuò)容的關(guān)鍵屬性 如果等于true 說明這個(gè)節(jié)點(diǎn)已經(jīng)處理過boolean finishing = false; // to ensure sweep before committing nextTabfor (int i = 0, bound = 0;;) {Node<K,V> f; int fh;//這個(gè)while循環(huán)體的作用就是在控制i-- 通過i--可以依次遍歷原h(huán)ash表中的節(jié)點(diǎn)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) {//如果所有的節(jié)點(diǎn)都已經(jīng)完成復(fù)制工作 就把nextTable賦值給table 清空臨時(shí)對象nextTablenextTable = null;table = nextTab;sizeCtl = (n << 1) - (n >>> 1);//擴(kuò)容閾值設(shè)置為原來容量的1.5倍 依然相當(dāng)于現(xiàn)在容量的0.75倍return;}//利用CAS方法更新這個(gè)擴(kuò)容閾值，在這里面sizectl值減一，說明新加入一個(gè)線程參與到擴(kuò)容操作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}}//如果遍歷到的節(jié)點(diǎn)為空 則放入ForwardingNode指針else if ((f = tabAt(tab, i)) == null)advance = casTabAt(tab, i, null, fwd);//如果遍歷到ForwardingNode節(jié)點(diǎn) 說明這個(gè)點(diǎn)已經(jīng)被處理過了 直接跳過 這里是控制并發(fā)擴(kuò)容的核心else if ((fh = f.hash) == MOVED)advance = true; // already processedelse {//節(jié)點(diǎn)上鎖synchronized (f) {if (tabAt(tab, i) == f) {Node<K,V> ln, hn;//如果fh>=0 證明這是一個(gè)Node節(jié)點(diǎn)if (fh >= 0) {int runBit = fh & n;//以下的部分在完成的工作是構(gòu)造兩個(gè)鏈表 一個(gè)是原鏈表 另一個(gè)是原鏈表的反序排列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位置上插入一個(gè)鏈表setTabAt(nextTab, i, ln);//在nextTable的i+n的位置上插入另一個(gè)鏈表setTabAt(nextTab, i + n, hn);//在table的i位置上插入forwardNode節(jié)點(diǎn) 表示已經(jīng)處理過該節(jié)點(diǎn)setTabAt(tab, i, fwd);//設(shè)置advance為true 返回到上面的while循環(huán)中 就可以執(zhí)行i--操作advance = true;}//對TreeBin對象進(jìn)行處理 與上面的過程類似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;//構(gòu)造正序和反序兩個(gè)鏈表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;}}//如果擴(kuò)容后已經(jīng)不再需要tree的結(jié)構(gòu) 反向轉(zhuǎn)換為鏈表結(jié)構(gòu)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位置上插入一個(gè)鏈表 setTabAt(nextTab, i, ln);//在nextTable的i+n的位置上插入另一個(gè)鏈表setTabAt(nextTab, i + n, hn);//在table的i位置上插入forwardNode節(jié)點(diǎn) 表示已經(jīng)處理過該節(jié)點(diǎn)setTabAt(tab, i, fwd);//設(shè)置advance為true 返回到上面的while循環(huán)中 就可以執(zhí)行i--操作advance = true;}}}}}}

---

?

六、put()方法：（尾插法）

前面的所有的介紹其實(shí)都為這個(gè)方法做鋪墊。ConcurrentHashMap最常用的就是put和get兩個(gè)方法。現(xiàn)在來介紹put方法，這個(gè)put方法依然沿用HashMap的put方法的思想，根據(jù)hash值計(jì)算這個(gè)新插入的點(diǎn)在table中的位置i，如果i位置是空的，直接放進(jìn)去，否則進(jìn)行判斷，如果i位置是樹節(jié)點(diǎn)，按照樹的方式插入新的節(jié)點(diǎn)，否則把i插入到鏈表的末尾。ConcurrentHashMap中依然沿用這個(gè)思想，有一個(gè)最重要的不同點(diǎn)就是ConcurrentHashMap不允許key或value為null值。另外由于涉及到多線程，put方法就要復(fù)雜一點(diǎn)。在多線程中可能有以下兩個(gè)情況。

如果一個(gè)或多個(gè)線程正在對ConcurrentHashMap進(jìn)行擴(kuò)容操作，當(dāng)前線程也要進(jìn)入擴(kuò)容的操作中。這個(gè)擴(kuò)容的操作之所以能被檢測到，是因?yàn)閠ransfer方法中在空結(jié)點(diǎn)上插入forward節(jié)點(diǎn)，如果檢測到需要插入的位置被forward節(jié)點(diǎn)占有，就幫助進(jìn)行擴(kuò)容；

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

整體流程就是首先定義不允許key或value為null的情況放入，對于每一個(gè)放入的值，首先利用spread方法對key的hashcode進(jìn)行一次hash計(jì)算，由此來確定這個(gè)值在table中的位置。

• 如果這個(gè)位置是空的，那么直接放入，而且不需要加鎖操作。
• 如果這個(gè)位置存在結(jié)點(diǎn)，說明發(fā)生了hash碰撞，首先判斷這個(gè)節(jié)點(diǎn)的類型。如果是鏈表節(jié)點(diǎn)（fh>0）,則得到的結(jié)點(diǎn)就是hash值相同的節(jié)點(diǎn)組成的鏈表的頭節(jié)點(diǎn)。需要依次向后遍歷確定這個(gè)新加入的值所在位置。如果遇到hash值與key值都與新加入節(jié)點(diǎn)是一致的情況，則只需要更新value值即可。否則依次向后遍歷，直到鏈表尾插入這個(gè)結(jié)點(diǎn)。??如果加入這個(gè)節(jié)點(diǎn)以后鏈表長度大于8，就把這個(gè)鏈表轉(zhuǎn)換成紅黑樹。如果這個(gè)節(jié)點(diǎn)的類型已經(jīng)是樹節(jié)點(diǎn)的話，直接調(diào)用樹節(jié)點(diǎn)的插入方法進(jìn)行插入新的值。

public V put(K key, V value) {return putVal(key, value, false);}/** Implementation for put and putIfAbsent */final V putVal(K key, V value, boolean onlyIfAbsent) {//不允許 key或value為nullif (key == null || value == null) throw new NullPointerException();//計(jì)算hash值int hash = spread(key.hashCode());int binCount = 0;//死循環(huán) 何時(shí)插入成功 何時(shí)跳出for (Node<K,V>[] tab = table;;) {Node<K,V> f; int n, i, fh;//如果table為空的話，初始化tableif (tab == null || (n = tab.length) == 0)tab = initTable();//根據(jù)hash值計(jì)算出在table里面的位置 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {//如果這個(gè)位置沒有值 ，直接放進(jìn)去，不需要加鎖if (casTabAt(tab, i, null,new Node<K,V>(hash, key, value, null)))break; // no lock when adding to empty bin}//當(dāng)遇到表連接點(diǎn)時(shí)，需要進(jìn)行整合表的操作else if ((fh = f.hash) == MOVED)tab = helpTransfer(tab, f);else {V oldVal = null;//結(jié)點(diǎn)上鎖 這里的結(jié)點(diǎn)可以理解為hash值相同組成的鏈表的頭結(jié)點(diǎn)synchronized (f) {if (tabAt(tab, i) == f) {//fh〉0 說明這個(gè)節(jié)點(diǎn)是一個(gè)鏈表的節(jié)點(diǎn) 不是樹的節(jié)點(diǎn)if (fh >= 0) {binCount = 1;//在這里遍歷鏈表所有的結(jié)點(diǎn)for (Node<K,V> e = f;; ++binCount) {K ek;//如果hash值和key值相同 則修改對應(yīng)結(jié)點(diǎn)的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;//如果遍歷到了最后一個(gè)結(jié)點(diǎn)，那么就證明新的節(jié)點(diǎn)需要插入 就把它插入在鏈表尾部if ((e = e.next) == null) {pred.next = new Node<K,V>(hash, key,value, null);break;}}}//如果這個(gè)節(jié)點(diǎn)是樹節(jié)點(diǎn)，就按照樹的方式插入值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) {//如果鏈表長度已經(jīng)達(dá)到臨界值8 就需要把鏈表轉(zhuǎn)換為樹結(jié)構(gòu)if (binCount >= TREEIFY_THRESHOLD)treeifyBin(tab, i);if (oldVal != null)return oldVal;break;}}}//將當(dāng)前ConcurrentHashMap的元素?cái)?shù)量+1addCount(1L, binCount);return null;}

---

6.1、helpTransfer()方法：

這是一個(gè)協(xié)助擴(kuò)容的方法。這個(gè)方法被調(diào)用的時(shí)候，當(dāng)前ConcurrentHashMap一定已經(jīng)有了nextTable對象，首先拿到這個(gè)nextTable對象，調(diào)用transfer方法。回看上面的transfer方法可以看到，當(dāng)本線程進(jìn)入擴(kuò)容方法的時(shí)候會直接進(jìn)入復(fù)制階段。

/*** Helps transfer if a resize is in progress.*/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);//計(jì)算一個(gè)操作校驗(yàn)碼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;}

---

6.2、treeifyBin()方法：

這個(gè)方法用于將過長的鏈表轉(zhuǎn)換為TreeBin對象。但是他并不是直接轉(zhuǎn)換，而是進(jìn)行一次容量判斷，如果容量沒有達(dá)到轉(zhuǎn)換的要求，直接進(jìn)行擴(kuò)容操作并返回；如果滿足條件才鏈表的結(jié)構(gòu)抓換為TreeBin ，這與HashMap不同的是，它并沒有把TreeNode直接放入紅黑樹，而是利用了TreeBin這個(gè)小容器來封裝所有的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)//如果table.length<64 就擴(kuò)大一倍 返回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;//構(gòu)造了一個(gè)TreeBin對象 把所有Node節(jié)點(diǎn)包裝成TreeNode放進(jìn)去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);//這里只是利用了TreeNode封裝 而沒有利用TreeNode的next域和parent域if ((p.prev = tl) == null)hd = p;elsetl.next = p;tl = p;}//在原來index的位置 用TreeBin替換掉原來的Node對象setTabAt(tab, index, new TreeBin<K,V>(hd));}}}}}

---

?

七、get()方法：

get方法比較簡單，給定一個(gè)key來確定value的時(shí)候，必須滿足兩個(gè)條件? key相同? hash值相同，對于節(jié)點(diǎn)可能在鏈表或樹上的情況，需要分別去查找。

public V get(Object key) {Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;//計(jì)算hash值int h = spread(key.hashCode());//根據(jù)hash值確定節(jié)點(diǎn)位置if ((tab = table) != null && (n = tab.length) > 0 &&(e = tabAt(tab, (n - 1) & h)) != null) {//如果搜索到的節(jié)點(diǎn)key與傳入的key相同且不為null,直接返回這個(gè)節(jié)點(diǎn) if ((eh = e.hash) == h) {if ((ek = e.key) == key || (ek != null && key.equals(ek)))return e.val;}//如果eh<0 說明這個(gè)節(jié)點(diǎn)在樹上 直接尋找else if (eh < 0)return (p = e.find(h, key)) != null ? p.val : null;//否則遍歷鏈表 找到對應(yīng)的值并返回while ((e = e.next) != null) {if (e.hash == h &&((ek = e.key) == key || (ek != null && key.equals(ek))))return e.val;}}return null;}

---

?

八、Size相關(guān)的方法：

對于ConcurrentHashMap來說，這個(gè)table里到底裝了多少東西其實(shí)是個(gè)不確定的數(shù)量，因?yàn)椴豢赡茉谡{(diào)用size()方法的時(shí)候像GC的“stop the world”一樣讓其他線程都停下來讓你去統(tǒng)計(jì)，因此只能說這個(gè)數(shù)量是個(gè)估計(jì)值。對于這個(gè)估計(jì)值，ConcurrentHashMap也是大費(fèi)周章才計(jì)算出來的。

8.1、輔助定義：

為了統(tǒng)計(jì)元素個(gè)數(shù)，ConcurrentHashMap定義了一些變量和一個(gè)內(nèi)部類。

/*** A padded cell for distributing counts. Adapted from LongAdder* and Striped64. See their internal docs for explanation.*/@sun.misc.Contended static final class CounterCell {volatile long value;CounterCell(long x) { value = x; }}/******************************************/ /*** 實(shí)際上保存的是hashmap中的元素個(gè)數(shù) 利用CAS鎖進(jìn)行更新但它并不用返回當(dāng)前hashmap的元素個(gè)數(shù) */private transient volatile long baseCount;/*** Spinlock (locked via CAS) used when resizing and/or creating CounterCells.*/private transient volatile int cellsBusy;/*** Table of counter cells. When non-null, size is a power of 2.*/private transient volatile CounterCell[] counterCells;

---

8.2 mappingCount與Size方法：

mappingCount與size方法的類似? 從Java工程師給出的注釋來看，應(yīng)該使用mappingCount代替size方法 兩個(gè)方法都沒有直接返回basecount 而是統(tǒng)計(jì)一次這個(gè)值，而這個(gè)值其實(shí)也是一個(gè)大概的數(shù)值，因此可能在統(tǒng)計(jì)的時(shí)候有其他線程正在執(zhí)行插入或刪除操作。

public int size() {long n = sumCount();return ((n < 0L) ? 0 :(n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :(int)n);}/*** Returns the number of mappings. This method should be used* instead of {@link #size} because a ConcurrentHashMap may* contain more mappings than can be represented as an int. The* value returned is an estimate; the actual count may differ if* there are concurrent insertions or removals.** @return the number of mappings* @since 1.8*/public long mappingCount() {long n = sumCount();return (n < 0L) ? 0L : n; // ignore transient negative values}final long sumCount() {CounterCell[] as = counterCells; CounterCell a;long sum = baseCount;if (as != null) {for (int i = 0; i < as.length; ++i) {if ((a = as[i]) != null)sum += a.value;//所有counter的值求和}}return sum;}

---

8.3 addCount方法：

在put方法結(jié)尾處調(diào)用了addCount方法，把當(dāng)前ConcurrentHashMap的元素個(gè)數(shù)+1這個(gè)方法一共做了兩件事,更新baseCount的值，檢測是否進(jìn)行擴(kuò)容。

private final void addCount(long x, int check) {CounterCell[] as; long b, s;//利用CAS方法更新baseCount的值 if ((as = counterCells) != null ||!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {CounterCell a; long v; int m;boolean uncontended = true;if (as == null || (m = as.length - 1) < 0 ||(a = as[ThreadLocalRandom.getProbe() & m]) == null ||!(uncontended =U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {fullAddCount(x, uncontended);return;}if (check <= 1)return;s = sumCount();}//如果check值大于等于0 則需要檢驗(yàn)是否需要進(jìn)行擴(kuò)容操作if (check >= 0) {Node<K,V>[] tab, nt; int n, sc;while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&(n = tab.length) < MAXIMUM_CAPACITY) {int rs = resizeStamp(n);//if (sc < 0) {if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||transferIndex <= 0)break;//如果已經(jīng)有其他線程在執(zhí)行擴(kuò)容操作if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))transfer(tab, nt);}//當(dāng)前線程是唯一的或是第一個(gè)發(fā)起擴(kuò)容的線程 此時(shí)nextTable=nullelse if (U.compareAndSwapInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) + 2))transfer(tab, null);s = sumCount();}}}

?

?

?

---

版權(quán)聲明：
作者：小小旭GISer?
原文地址：https://blog.csdn.net/u010723709/article/details/48007881

?

與50位技術(shù)專家面對面20年技術(shù)見證，附贈技術(shù)全景圖

總結(jié)

以上是生活随笔為你收集整理的Java集合篇：ConcurrentHashMap详解（JDK1.8）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。