CouncurrentHashMap源码解析
注：本文源碼是JDK8的版本，與之前的版本有較大差異
jdk1.7的源碼：https://www.cnblogs.com/chengxiao/p/6842045.html
ConcurrentHashMap是conccurrent家族中的一個(gè)類，由于它可以高效地支持并發(fā)操作，以及被廣泛使用，經(jīng)典的開源框架Spring的底層數(shù)據(jù)結(jié)構(gòu)就是使用ConcurrentHashMap實(shí)現(xiàn)的。與同是線程安全的老大哥HashTable相比，它已經(jīng)更勝一籌，因此它的鎖更加細(xì)化，而不是像HashTable一樣為幾乎每個(gè)方法都添加了synchronized鎖，這樣的鎖無疑會(huì)影響到性能。
本文的分析的源碼是JDK8的版本，與JDK6的版本有很大的差異。實(shí)現(xiàn)線程安全的思想也已經(jīng)完全變了，它摒棄了Segment（鎖段）的概念，而是啟用了一種全新的方式實(shí)現(xiàn),利用CAS算法。它沿用了與它同時(shí)期的HashMap版本的思想，底層依然由“數(shù)組”+鏈表+紅黑樹的方式思想，但是為了做到并發(fā)，又增加了很多輔助的類，例如TreeBin，Traverser等對(duì)象內(nèi)部類。
1 重要的屬性
首先來看幾個(gè)重要的屬性，與HashMap相同的就不再介紹了，這里重點(diǎn)解釋一下sizeCtl這個(gè)屬性。可以說它是ConcurrentHashMap中出鏡率很高的一個(gè)屬性，因?yàn)樗且粋€(gè)控制標(biāo)識(shí)符，在不同的地方有不同用途，而且它的取值不同，也代表不同的含義。
- 負(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是對(duì)應(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)識(shí)量。
負(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)
2 重要的內(nèi)部類
2.1 Node
Node是最核心的內(nèi)部類，它包裝了key-value鍵值對(duì)，所有插入ConcurrentHashMap的數(shù)據(jù)都包裝在這里面。它與HashMap中的定義很相似，但是但是有一些差別它對(duì)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;//帶有同步鎖的value
volatile 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類很相似，但是有一些差別
它對(duì)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)鏈表長(zhǎng)度過長(zhǎng)的時(shí)候，會(huì)轉(zhuǎn)換為TreeNode。但是與HashMap不相同的是，它并不是直接轉(zhuǎn)換為紅黑樹，而是把這些結(jié)點(diǎn)包裝成TreeNode放在TreeBin對(duì)象中，由TreeBin完成對(duì)紅黑樹的包裝。而且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對(duì)象，而不是TreeNode對(duì)象，這是與HashMap的區(qū)別。另外這個(gè)類還帶有了讀寫鎖。
這里僅貼出它的構(gòu)造方法。可以看到在構(gòu)造TreeBin節(jié)點(diǎn)時(shí)，僅僅指定了它的hash值為TREEBIN常量，這也就是個(gè)標(biāo)識(shí)為。同時(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;
else
xp.right = x;
r = balanceInsertion(r, x);
break;
}
}
}
}
this.root = r;
assert checkInvariants(root);
}
2.5?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 nodes
outer: 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;
}
else
return e.find(h, k);
}
if ((e = e.next) == null)
return null;
}
}
}
}
3 Unsafe與CAS
在ConcurrentHashMap中，隨處可以看到U, 大量使用了U.compareAndSwapXXX的方法，這個(gè)方法是利用一個(gè)CAS算法實(shí)現(xiàn)無鎖化的修改值的操作，他可以大大降低鎖代理的性能消耗。這個(gè)算法的基本思想就是不斷地去比較當(dāng)前內(nèi)存中的變量值與你指定的一個(gè)變量值是否相等，如果相等，則接受你指定的修改的值，否則拒絕你的操作。因?yàn)楫?dāng)前線程中的值已經(jīng)不是最新的值，你的修改很可能會(huì)覆蓋掉其他線程修改的結(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è)原子操作，用于對(duì)指定位置的節(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算法中，會(huì)比較內(nèi)存中的值與你指定的這個(gè)值是否相等，如果相等才接受你的修改，否則拒絕你的修改
//因此當(dāng)前線程中的值并不是最新的值，這種修改可能會(huì)覆蓋掉其他線程的修改結(jié)果 有點(diǎn)類似于SVN
static 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);
}
4 初始化方法initTable
對(duì)于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表示有其他線程正在進(jìn)行初始化操作，把線程掛起。對(duì)于table的初始化工作，只能有一個(gè)線程在進(jìn)行。
if ((sc = sizeCtl) < 0)
Thread.yield(); // lost initialization race; just spin
else 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;
}
5 擴(kuò)容方法 transfer
當(dāng)ConcurrentHashMap容量不足的時(shí)候，需要對(duì)table進(jìn)行擴(kuò)容。這個(gè)方法的基本思想跟HashMap是很像的，但是由于它是支持并發(fā)擴(kuò)容的，所以要復(fù)雜的多。原因是它支持多線程進(jìn)行擴(kuò)容操作，而并沒有加鎖。我想這樣做的目的不僅僅是為了滿足concurrent的要求，而是希望利用并發(fā)處理去減少擴(kuò)容帶來的時(shí)間影響。因?yàn)樵跀U(kuò)容的時(shí)候，總是會(huì)涉及到從一個(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è)地方在后面會(huì)有提到；
- 第二個(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)，就把對(duì)應(yīng)點(diǎn)的值set為forward，另一個(gè)線程看到forward，就向后遍歷。這樣交叉就完成了復(fù)制工作。而且還很好的解決了線程安全的問題。 這個(gè)方法的設(shè)計(jì)實(shí)在是讓我膜拜。
/**
* 一個(gè)過渡的table表 只有在擴(kuò)容的時(shí)候才會(huì)使用
*/
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 range
if (nextTab == null) { // initiating
try {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];//構(gòu)造一個(gè)nextTable對(duì)象 它的容量是原來的兩倍
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = 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 nextTab
for (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í)對(duì)象nextTable
nextTable = 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 processed
else {
//節(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);
else
hn = 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;
}
//對(duì)TreeBin對(duì)象進(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;
else
loTail.next = p;
loTail = p;
++lc;
}
else {
if ((p.prev = hiTail) == null)
hi = p;
else
hiTail.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;
}
}
}
}
}
}
6 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è)線程正在對(duì)ConcurrentHashMap進(jìn)行擴(kuò)容操作，當(dāng)前線程也要進(jìn)入擴(kuò)容的操作中。這個(gè)擴(kuò)容的操作之所以能被檢測(cè)到，是因?yàn)閠ransfer方法中在空結(jié)點(diǎn)上插入forward節(jié)點(diǎn)，如果檢測(cè)到需要插入的位置被forward節(jié)點(diǎn)占有，就幫助進(jìn)行擴(kuò)容；
如果檢測(cè)到要插入的節(jié)點(diǎn)是非空且不是forward節(jié)點(diǎn)，就對(duì)這個(gè)節(jié)點(diǎn)加鎖，這樣就保證了線程安全。盡管這個(gè)有一些影響效率，但是還是會(huì)比hashTable的synchronized要好得多。
整體流程就是首先定義不允許key或value為null的情況放入? 對(duì)于每一個(gè)放入的值，首先利用spread方法對(duì)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)以后鏈表長(zhǎng)度大于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為null
if (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為空的話，初始化table
if (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值相同 則修改對(duì)應(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) {
//如果鏈表長(zhǎng)度已經(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ù)量+1
addCount(1L, binCount);
return null;
}
6.1 helpTransfer方法
這是一個(gè)協(xié)助擴(kuò)容的方法。這個(gè)方法被調(diào)用的時(shí)候，當(dāng)前ConcurrentHashMap一定已經(jīng)有了nextTable對(duì)象，首先拿到這個(gè)nextTable對(duì)象，調(diào)用transfer方法。回看上面的transfer方法可以看到，當(dāng)本線程進(jìn)入擴(kuò)容方法的時(shí)候會(huì)直接進(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è)方法用于將過長(zhǎng)的鏈表轉(zhuǎn)換為TreeBin對(duì)象。但是他并不是直接轉(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對(duì)象 把所有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;
else
tl.next = p;
tl = p;
}
//在原來index的位置 用TreeBin替換掉原來的Node對(duì)象
setTabAt(tab, index, new TreeBin<K,V>(hd));
}
}
}
}
}
7 get方法
get方法比較簡(jiǎn)單，給定一個(gè)key來確定value的時(shí)候，必須滿足兩個(gè)條件? key相同? hash值相同，對(duì)于節(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;
//否則遍歷鏈表 找到對(duì)應(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;
}
8 Size相關(guān)的方法
對(duì)于ConcurrentHashMap來說，這個(gè)table里到底裝了多少東西其實(shí)是個(gè)不確定的數(shù)量，因?yàn)椴豢赡茉谡{(diào)用size()方法的時(shí)候像GC的“stop the world”一樣讓其他線程都停下來讓你去統(tǒng)計(jì)，因此只能說這個(gè)數(shù)量是個(gè)估計(jì)值。對(duì)于這個(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的值，檢測(cè)是否進(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=null
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
s = sumCount();
}
}
}
（未完待續(xù)。。。）
from:https://blog.csdn.net/u010723709/article/details/48007881
總結(jié)
以上是生活随笔為你收集整理的CouncurrentHashMap源码解析的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： java集合（6）：TreeMap源码分
- 下一篇： 面试必备：LinkedHashMap源码