承接上文
在上一篇文章的時(shí)候党瓮，已經(jīng)基本解釋了JDK1.7版本的ConcurrentHashMap的核心代碼屋吨，可見：Java源碼分析專題系列之【ConcurrentHashMap】深入淺出的源碼分析（JDK1.7版本）慰毅，接下來我們要研究一下目前非常重要的JDK1.8版本的ConcurrentHashMap泉蝌，這是目前我們最應(yīng)該學(xué)習(xí)的技術(shù)源碼之一银酬。
前提概要
ConcurrentHashMap是concurrent家族中的一個(gè)類榄棵，由于它可以高效地支持并發(fā)操作凝颇，以及被廣泛使用，經(jīng)典的開源框架Spring的底層數(shù)據(jù)結(jié)構(gòu)就是使用ConcurrentHashMap實(shí)現(xiàn)的疹鳄。
與同是線程安全的老大哥HashTable相比拧略，它已經(jīng)更勝一籌，因此它的鎖更加細(xì)化瘪弓，而不是像HashTable一樣為幾乎每個(gè)方法都添加了synchronized鎖垫蛆，這樣的鎖無疑會(huì)影響到性能。
原理簡(jiǎn)介
本文的分析的源碼是JDK8的版本，與JDK7的版本有很大的差異月褥。實(shí)現(xiàn)線程安全的思想也已經(jīng)完全變了弛随，它摒棄了Segment（鎖段）的概念，而是啟用了一種全新的方式實(shí)現(xiàn)宁赤，利用CAS算法舀透。
它沿用了與它同時(shí)期(JDK1.8)的HashMap版本的思想，底層依然由“數(shù)組”+鏈表+紅黑樹的方式思想决左，但是為了做到并發(fā)愕够，又增加了很多輔助的類，例如TreeBin佛猛，Traverser等對(duì)象內(nèi)部類惑芭。
重要的屬性
首先來看幾個(gè)重要的屬性，與HashMap相同的就不再介紹了继找，這里重點(diǎn)解釋一下sizeCtl這個(gè)屬性遂跟。可以說它是ConcurrentHashMap中出鏡率很高的一個(gè)屬性婴渡，因?yàn)樗且粋€(gè)控制標(biāo)識(shí)符幻锁，在不同的地方有不同用途，而且它的取值不同边臼，也代表不同的含義哄尔。
sizeCtl
- ** 負(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)
重要的內(nèi)部類
Node（鏈表）
Node是最核心的內(nèi)部類，它包裝了key-value鍵值對(duì)半哟，所有插入ConcurrentHashMap的數(shù)據(jù)都包裝在這里面酬滤。
如下所示：
static class Node<K,V> implements Map.Entry<K,V> {
final int hash; //hash值
final K key; //鍵值key
volatile V val;//帶有sync鎖的value
volatile Node<K,V> next;//帶有sync鎖的next指針
Node(int hash, K key, V val, Node<K,V> next) {
this.hash = hash; // hash
this.key = key; // key
this.val = val; // value
this.next = next; //下一個(gè)節(jié)點(diǎn)的引用
}
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();
}
// 比較兩個(gè)node是否相等
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)) && // 再比 key
(v == (u = val) || v.equals(u))); // 再比 value值
}
/**
* 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的sync鎖
它不允許調(diào)用setValue方法直接改變Node的value域
它增加了find方法輔助map.get()方法
TreeNode（樹節(jié)點(diǎn)）
樹節(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的訪問年栓。
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)造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; // 包裝進(jìn)去作為first節(jié)點(diǎn)
TreeNode<K,V> r = null;
//初始化根節(jié)點(diǎn)
for (TreeNode<K,V> x = b, next; x != null; x = next) {
next = (TreeNode<K,V>)x.next; //傳入的節(jié)點(diǎn)的next賦值為當(dāng)前treebin節(jié)點(diǎn)的next節(jié)點(diǎn)打月，作為數(shù)據(jù)同步使用
x.left = x.right = null; //將左子樹和右子樹都是null
if (r == null) {
x.parent = null;//父節(jié)點(diǎn)為空
x.red = false; // 不屬于紅子樹節(jié)點(diǎn)
r = x; //將紅節(jié)點(diǎn)队腐，臨時(shí)存儲(chǔ)起來，用于下次循環(huán)使用
}
else {
K k = x.key; //如果不是根節(jié)點(diǎn)則就是下一個(gè)子樹節(jié)點(diǎn)（屬于next值）
int h = x.hash; // 如果不是根節(jié)點(diǎn)就取出hash值
Class<?> kc = null;
for (TreeNode<K,V> p = r;;) {
int dir, ph; //
K pk = p.key;//獲取根節(jié)點(diǎn)的key值
if ((ph = p.hash) > h) // 獲取hash值判斷是否屬于小于根節(jié)點(diǎn)hash值
dir = -1; //屬于左子樹奏篙，
else if (ph < h) // 否則屬于右子樹
dir = 1;
else if ((kc == null &&
(kc = comparableClassFor(k)) == null) ||
(dir = compareComparables(kc, k, pk)) == 0)
// 當(dāng)hash值相等的時(shí)候柴淘，比較class類以及比較key值
dir = tieBreakOrder(k, pk);
TreeNode<K,V> xp = p;
if ((p = (dir <= 0) ? p.left : p.right) == null) {
// 判斷比較結(jié)果是否小于零，且左右子樹都為null的情況下
x.parent = xp;// 將父節(jié)點(diǎn)進(jìn)行賦值
if (dir <= 0)
xp.left = x; // 進(jìn)行判斷是否屬于左子樹（<=0）
else
xp.right = x;// 進(jìn)行判斷是否屬于右子樹（>0）
// 進(jìn)行平衡插入機(jī)制（其實(shí)就是建立平衡關(guān)系）
r = balanceInsertion(r, x);
break;
}
}
}
}
this.root = r; // 根節(jié)點(diǎn)
assert checkInvariants(root);
}
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;
}
// 循環(huán)遍歷查找相關(guān)的對(duì)應(yīng)hash和key值的
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;
// 如果無法定位或者定位到null則直接返回
if (k == null || tab == null || (n = tab.length) == 0 ||
(e = tabAt(tab, (n - 1) & h)) == null)
return null;
for (;;) {
int eh; K ek;
// 遍歷查找對(duì)比hash值和key值，從而進(jìn)行分析是否相等话原。
if ((eh = e.hash) == h &&
((ek = e.key) == k || (ek != null && k.equals(ek))))
// 查找到后相等直接返回
return e;
//如果hash值為-1 或者 -2
if (eh < 0) {
// 如果是forwardingNode 則就是代表是-1
if (e instanceof ForwardingNode) {
//直接獲取到下一個(gè)表夕吻。
tab = ((ForwardingNode<K,V>)e).nextTable;
continue outer;
}
else
return e.find(h, k);
}
//直到為null，直接返回
if ((e = e.next) == null)
return null;
}
}
}
}
Unsafe與CAS
在ConcurrentHashMap中繁仁，隨處可以看到Unsafe, 大量使用了Unsafe.compareAndSwapXXX的方法涉馅，這個(gè)方法是利用一個(gè)CAS算法實(shí)現(xiàn)無鎖化的修改值的操作，他可以大大降低鎖代理的性能消耗黄虱。
這個(gè)算法的基本思想就是不斷地去比較當(dāng)前內(nèi)存中的變量值與你指定的一個(gè)變量值是否相等稚矿，如果相等，則接受你指定的修改的值，否則拒絕你的操作晤揣。
因?yàn)楫?dāng)前線程中的值已經(jīng)不是最新的值桥爽，你的修改很可能會(huì)覆蓋掉其他線程修改的結(jié)果。這一點(diǎn)與樂觀鎖昧识，SVN的思想是比較類似的钠四。
unsafe靜態(tài)塊
unsafe代碼塊控制了一些屬性的修改工作，比如最常用的SIZECTL 跪楞。 在這一版本的concurrentHashMap中形导，大量應(yīng)用來的CAS方法進(jìn)行變量、屬性的修改工作习霹。 利用CAS進(jìn)行無鎖操作朵耕，可以大大提高性能。
// 各個(gè)字段屬性的偏移量淋叶，可以通過偏移量+首地址獲取到對(duì)應(yīng)的數(shù)據(jù)對(duì)象
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);
}
}
三個(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);
}
初始化方法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;
}
擴(kuò)容方法 transfer
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ò)容属瓣。
再看一下多線程是如何完成的：
如果遍歷到的節(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;
}
}
}
}
}
}
put方法
ConcurrentHashMap最常用的就是put和get兩個(gè)方法
根據(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;
}
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;
}
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));
}
}
}
}
}
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;
}
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ì)算出來的鹃共。
輔助定義
為了統(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;
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;
}
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();
}
}
}
