轉自：http://www.importnew.com/22007.html
並發編程實踐中，ConcurrentHashMap是一個經常被使用的數據結構，相比於Hashtable以及Collections.synchronizedMap()，ConcurrentHashMap在線程安全的基礎上提供了更好的寫並發能力，但同時降低了對讀一致性的要求（這點好像CAP理論啊 O(∩_∩)O）。ConcurrentHashMap的設計與實現非常精巧，大量的利用了volatile，final，CAS等lock-free技術來減少鎖競爭對於性能的影響，無論對於Java並發編程的學習還是Java內存模型的理解，ConcurrentHashMap的設計以及源碼都值得非常仔細的閱讀與揣摩。
這篇日志記錄了自己對ConcurrentHashMap的一些總結，由於JDK6，7，8中實現都不同，需要分開闡述在不同版本中的ConcurrentHashMap。
之前已經在ConcurrentHashMap原理分析中解釋了ConcurrentHashMap的原理，主要是從代碼的角度來闡述是源碼是如何寫的，本文仍然從源碼出發，挑選個人覺得重要的點（會用紅色標注）再次進行回顧，以及闡述ConcurrentHashMap的一些注意點。
1. JDK6與JDK7中的實現
1.1 設計思路
ConcurrentHashMap采用了分段鎖的設計，只有在同一個分段內才存在競態關系，不同的分段鎖之間沒有鎖競爭。相比於對整個Map加鎖的設計，分段鎖大大的提高了高並發環境下的處理能力。但同時，由於不是對整個Map加鎖，導致一些需要掃描整個Map的方法（如size(), containsValue()）需要使用特殊的實現，另外一些方法（如clear()）甚至放棄了對一致性的要求（ConcurrentHashMap是弱一致性的，具體請查看ConcurrentHashMap能完全替代HashTable嗎？）。
ConcurrentHashMap中的分段鎖稱為Segment，它即類似於HashMap（JDK7與JDK8中HashMap的實現）的結構，即內部擁有一個Entry數組，數組中的每個元素又是一個鏈表；同時又是一個ReentrantLock（Segment繼承了ReentrantLock）。ConcurrentHashMap中的HashEntry相對於HashMap中的Entry有一定的差異性：HashEntry中的value以及next都被volatile修飾，這樣在多線程讀寫過程中能夠保持它們的可見性，代碼如下：
|
1
2
3
4
5
|
static
final
class
HashEntry<K,V> {
final
int
hash;
final
K key;
volatile
V value;
volatile
HashEntry<K,V> next;
|
1.2 並發度（Concurrency Level）
並發度可以理解為程序運行時能夠同時更新ConccurentHashMap且不產生鎖競爭的最大線程數，實際上就是ConcurrentHashMap中的分段鎖個數，即Segment[]的數組長度。ConcurrentHashMap默認的並發度為16，但用戶也可以在構造函數中設置並發度。當用戶設置並發度時，ConcurrentHashMap會使用大於等於該值的最小2冪指數作為實際並發度（假如用戶設置並發度為17，實際並發度則為32）。運行時通過將key的高n位（n = 32 – segmentShift）和並發度減1（segmentMask）做位與運算定位到所在的Segment。segmentShift與segmentMask都是在構造過程中根據concurrency level被相應的計算出來。
如果並發度設置的過小，會帶來嚴重的鎖競爭問題；如果並發度設置的過大，原本位於同一個Segment內的訪問會擴散到不同的Segment中，CPU cache命中率會下降，從而引起程序性能下降。（文檔的說法是根據你並發的線程數量決定，太多會導性能降低）
1.3 創建分段鎖
和JDK6不同，JDK7中除了第一個Segment之外，剩余的Segments采用的是延遲初始化的機制：每次put之前都需要檢查key對應的Segment是否為null，如果是則調用ensureSegment()以確保對應的Segment被創建。
ensureSegment可能在並發環境下被調用，但與想象中不同，ensureSegment並未使用鎖來控制競爭，而是使用了Unsafe對象的getObjectVolatile()提供的原子讀語義結合CAS來確保Segment創建的原子性。代碼段如下：
|
1
2
3
4
5
6
7
8
9
|
if
((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
==
null
) {
// recheck
Segment<K,V> s =
new
Segment<K,V>(lf, threshold, tab);
while
((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
==
null
) {
if
(UNSAFE.compareAndSwapObject(ss, u,
null
, seg = s))
break
;
}
}
|
1.4 put/putIfAbsent/putAll
和JDK6一樣，ConcurrentHashMap的put方法被代理到了對應的Segment（定位Segment的原理之前已經描述過）中。與JDK6不同的是，JDK7版本的ConcurrentHashMap在獲得Segment鎖的過程中，做了一定的優化 - 在真正申請鎖之前，put方法會通過tryLock()方法嘗試獲得鎖，在嘗試獲得鎖的過程中會對對應hashcode的鏈表進行遍歷，如果遍歷完畢仍然找不到與key相同的HashEntry節點，則為后續的put操作提前創建一個HashEntry。當tryLock一定次數后仍無法獲得鎖，則通過lock申請鎖。
需要注意的是，由於在並發環境下，其他線程的put，rehash或者remove操作可能會導致鏈表頭結點的變化，因此在過程中需要進行檢查，如果頭結點發生變化則重新對表進行遍歷。而如果其他線程引起了鏈表中的某個節點被刪除，即使該變化因為是非原子寫操作（刪除節點后鏈接后續節點調用的是Unsafe.putOrderedObject()，該方法不提供原子寫語義）可能導致當前線程無法觀察到，但因為不影響遍歷的正確性所以忽略不計。
之所以在獲取鎖的過程中對整個鏈表進行遍歷，主要目的是希望遍歷的鏈表被CPU cache所緩存，為后續實際put過程中的鏈表遍歷操作提升性能。
在獲得鎖之后，Segment對鏈表進行遍歷，如果某個HashEntry節點具有相同的key，則更新該HashEntry的value值，否則新建一個HashEntry節點，將它設置為鏈表的新head節點並將原頭節點設為新head的下一個節點。新建過程中如果節點總數（含新建的HashEntry）超過threshold，則調用rehash()方法對Segment進行擴容，最后將新建HashEntry寫入到數組中。
put方法中，鏈接新節點的下一個節點（HashEntry.setNext()）以及將鏈表寫入到數組中（setEntryAt()）都是通過Unsafe的putOrderedObject()方法來實現，這里並未使用具有原子寫語義的putObjectVolatile()的原因是：JMM會保證獲得鎖到釋放鎖之間所有對象的狀態更新都會在鎖被釋放之后更新到主存，從而保證這些變更對其他線程是可見的。
1.5 rehash
相對於HashMap的resize，ConcurrentHashMap的rehash原理類似，但是Doug Lea為rehash做了一定的優化，避免讓所有的節點都進行復制操作：由於擴容是基於2的冪指來操作，假設擴容前某HashEntry對應到Segment中數組的index為i，數組的容量為capacity，那么擴容后該HashEntry對應到新數組中的index只可能為i或者i+capacity，因此大多數HashEntry節點在擴容前后index可以保持不變。基於此，rehash方法中會定位第一個后續所有節點在擴容后index都保持不變的節點，然后將這個節點之前的所有節點重排即可。這部分代碼如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
|
private
void
rehash(HashEntry<K,V> node) {
HashEntry<K,V>[] oldTable = table;
int
oldCapacity = oldTable.length;
int
newCapacity = oldCapacity <<
1
;
threshold = (
int
)(newCapacity * loadFactor);
HashEntry<K,V>[] newTable =
(HashEntry<K,V>[])
new
HashEntry[newCapacity];
int
sizeMask = newCapacity -
1
;
for
(
int
i =
0
; i < oldCapacity ; i++) {
HashEntry<K,V> e = oldTable[i];
if
(e !=
null
) {
HashEntry<K,V> next = e.next;
int
idx = e.hash & sizeMask;
if
(next ==
null
)
// Single node on list
newTable[idx] = e;
else
{
// Reuse consecutive sequence at same slot
HashEntry<K,V> lastRun = e;
int
lastIdx = idx;
for
(HashEntry<K,V> last = next;
last !=
null
;
last = last.next) {
int
k = last.hash & sizeMask;
if
(k != lastIdx) {
lastIdx = k;
lastRun = last;
}
}
newTable[lastIdx] = lastRun;
// Clone remaining nodes
for
(HashEntry<K,V> p = e; p != lastRun; p = p.next) {
V v = p.value;
int
h = p.hash;
int
k = h & sizeMask;
HashEntry<K,V> n = newTable[k];
newTable[k] =
new
HashEntry<K,V>(h, p.key, v, n);
}
}
}
}
int
nodeIndex = node.hash & sizeMask;
// add the new node
node.setNext(newTable[nodeIndex]);
newTable[nodeIndex] = node;
table = newTable;
}
|
1.6 remove
和put類似，remove在真正獲得鎖之前，也會對鏈表進行遍歷以提高緩存命中率。
1.7 get與containsKey
get與containsKey兩個方法幾乎完全一致：他們都沒有使用鎖，而是通過Unsafe對象的getObjectVolatile()方法提供的原子讀語義，來獲得Segment以及對應的鏈表，然后對鏈表遍歷判斷是否存在key相同的節點以及獲得該節點的value。但由於遍歷過程中其他線程可能對鏈表結構做了調整，因此get和containsKey返回的可能是過時的數據，這一點是ConcurrentHashMap在弱一致性上的體現。如果要求強一致性，那么必須使用Collections.synchronizedMap()方法。
1.8 size、containsValue
這些方法都是基於整個ConcurrentHashMap來進行操作的，他們的原理也基本類似：首先不加鎖循環執行以下操作：循環所有的Segment（通過Unsafe的getObjectVolatile()以保證原子讀語義），獲得對應的值以及所有Segment的modcount之和。如果連續兩次所有Segment的modcount和相等，則過程中沒有發生其他線程修改ConcurrentHashMap的情況，返回獲得的值。
當循環次數超過預定義的值時，這時需要對所有的Segment依次進行加鎖，獲取返回值后再依次解鎖。值得注意的是，加鎖過程中要強制創建所有的Segment，否則容易出現其他線程創建Segment並進行put，remove等操作。代碼如下：
|
1
2
3
|
for
(
int
j =
0
; j < segments.length; ++j)
ensureSegment(j).lock();
// force creation
|
一般來說，應該避免在多線程環境下使用size和containsValue方法。
注1：modcount在put, replace, remove以及clear等方法中都會被修改。
注2：對於containsValue方法來說，如果在循環過程中發現匹配value的HashEntry，則直接返回true。
最后，與HashMap不同的是，ConcurrentHashMap並不允許key或者value為null，按照Doug Lea的說法，這么設計的原因是在ConcurrentHashMap中，一旦value出現null，則代表HashEntry的key/value沒有映射完成就被其他線程所見，需要特殊處理。在JDK6中，get方法的實現中就有一段對HashEntry.value == null的防御性判斷。但Doug Lea也承認實際運行過程中，這種情況似乎不可能發生（參考：http://cs.oswego.edu/pipermail/concurrency-interest/2011-March/007799.html）。
2. JDK8中的實現
ConcurrentHashMap在JDK8中進行了巨大改動，很需要通過源碼來再次學習下Doug Lea的實現方法。
它摒棄了Segment（鎖段）的概念，而是啟用了一種全新的方式實現,利用CAS算法。它沿用了與它同時期的HashMap版本的思想，底層依然由“數組”+鏈表+紅黑樹的方式思想(JDK7與JDK8中HashMap的實現)，但是為了做到並發，又增加了很多輔助的類，例如TreeBin，Traverser等對象內部類。
2.1 重要的屬性
首先來看幾個重要的屬性，與HashMap相同的就不再介紹了，這里重點解釋一下sizeCtl這個屬性。可以說它是ConcurrentHashMap中出鏡率很高的一個屬性，因為它是一個控制標識符，在不同的地方有不同用途，而且它的取值不同，也代表不同的含義。
- 負數代表正在進行初始化或擴容操作
- -1代表正在初始化
- -N 表示有N-1個線程正在進行擴容操作
- 正數或0代表hash表還沒有被初始化，這個數值表示初始化或下一次進行擴容的大小，這一點類似於擴容閾值的概念。還后面可以看到，它的值始終是當前ConcurrentHashMap容量的0.75倍，這與loadfactor是對應的。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
|
/**
* 盛裝Node元素的數組 它的大小是2的整數次冪
* 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表初始化或擴容時的一個控制位標識量。
負數代表正在進行初始化或擴容操作
-1代表正在初始化
-N 表示有N-1個線程正在進行擴容操作
正數或0代表hash表還沒有被初始化，這個數值表示初始化或下一次進行擴容的大小
*/
private
transient
volatile
int
sizeCtl;
// 以下兩個是用來控制擴容的時候 單線程進入的變量
/**
* 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，表示這是一個forwardNode節點
static
final
int
TREEBIN = -
2
;
// hash值是-2 表示這時一個TreeBin節點
|
2.2 重要的類
2.2.1 Node
Node是最核心的內部類，它包裝了key-value鍵值對，所有插入ConcurrentHashMap的數據都包裝在這里面。它與HashMap中的定義很相似，但是但是有一些差別它對value和next屬性設置了volatile同步鎖(與JDK7的Segment相同)，它不允許調用setValue方法直接改變Node的value域，它增加了find方法輔助map.get()方法。
2.2.2 TreeNode
樹節點類，另外一個核心的數據結構。當鏈表長度過長的時候，會轉換為TreeNode。但是與HashMap不相同的是，它並不是直接轉換為紅黑樹，而是把這些結點包裝成TreeNode放在TreeBin對象中，由TreeBin完成對紅黑樹的包裝。而且TreeNode在ConcurrentHashMap集成自Node類，而並非HashMap中的集成自LinkedHashMap.Entry<K,V>類，也就是說TreeNode帶有next指針，這樣做的目的是方便基於TreeBin的訪問。
2.2.3 TreeBin
這個類並不負責包裝用戶的key、value信息，而是包裝的很多TreeNode節點。它代替了TreeNode的根節點，也就是說在實際的ConcurrentHashMap“數組”中，存放的是TreeBin對象，而不是TreeNode對象，這是與HashMap的區別。另外這個類還帶有了讀寫鎖。
這里僅貼出它的構造方法。可以看到在構造TreeBin節點時，僅僅指定了它的hash值為TREEBIN常量，這也就是個標識為。同時也看到我們熟悉的紅黑樹構造方法
2.2.4 ForwardingNode
一個用於連接兩個table的節點類。它包含一個nextTable指針，用於指向下一張表。而且這個節點的key value next指針全部為null，它的hash值為-1. 這里面定義的find的方法是從nextTable里進行查詢節點，而不是以自身為頭節點進行查找。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
|
/**
* 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
;
}
}
}
}
|
2.3 Unsafe與CAS
在ConcurrentHashMap中，隨處可以看到U, 大量使用了U.compareAndSwapXXX的方法，這個方法是利用一個CAS算法實現無鎖化的修改值的操作，他可以大大降低鎖代理的性能消耗。這個算法的基本思想就是不斷地去比較當前內存中的變量值與你指定的一個變量值是否相等，如果相等，則接受你指定的修改的值，否則拒絕你的操作。因為當前線程中的值已經不是最新的值，你的修改很可能會覆蓋掉其他線程修改的結果。這一點與樂觀鎖，SVN的思想是比較類似的。
2.3.1 unsafe靜態塊
unsafe代碼塊控制了一些屬性的修改工作，比如最常用的SIZECTL 。在這一版本的concurrentHashMap中，大量應用來的CAS方法進行變量、屬性的修改工作。利用CAS進行無鎖操作，可以大大提高性能。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
|
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);
}
}
|
2.3.2 三個核心方法
ConcurrentHashMap定義了三個原子操作，用於對指定位置的節點進行操作。正是這些原子操作保證了ConcurrentHashMap的線程安全。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
//獲得在i位置上的Node節點
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算法設置i位置上的Node節點。之所以能實現並發是因為他指定了原來這個節點的值是多少
//在CAS算法中，會比較內存中的值與你指定的這個值是否相等，如果相等才接受你的修改，否則拒絕你的修改
//因此當前線程中的值並不是最新的值，這種修改可能會覆蓋掉其他線程的修改結果 有點類似於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方法設置節點位置的值
static
final
<K,V>
void
setTabAt(Node<K,V>[] tab,
int
i, Node<K,V> v) {
U.putObjectVolatile(tab, ((
long
)i << ASHIFT) + ABASE, v);
}
|
2.4 初始化方法initTable
對於ConcurrentHashMap來說，調用它的構造方法僅僅是設置了一些參數而已。而整個table的初始化是在向ConcurrentHashMap中插入元素的時候發生的。如調用put、computeIfAbsent、compute、merge等方法的時候，調用時機是檢查table==null。
初始化方法主要應用了關鍵屬性sizeCtl 如果這個值〈0，表示其他線程正在進行初始化，就放棄這個操作。在這也可以看出ConcurrentHashMap的初始化只能由一個線程完成。如果獲得了初始化權限，就用CAS方法將sizeCtl置為-1，防止其他線程進入。初始化數組后，將sizeCtl的值改為0.75*n。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
/**
* 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;
}
|
2.5 擴容方法 transfer
當ConcurrentHashMap容量不足的時候，需要對table進行擴容。這個方法的基本思想跟HashMap是很像的，但是由於它是支持並發擴容的，所以要復雜的多。原因是它支持多線程進行擴容操作，而並沒有加鎖。我想這樣做的目的不僅僅是為了滿足concurrent的要求，而是希望利用並發處理去減少擴容帶來的時間影響。因為在擴容的時候，總是會涉及到從一個“數組”到另一個“數組”拷貝的操作，如果這個操作能夠並發進行，那真真是極好的了。
整個擴容操作分為兩個部分
- 第一部分是構建一個nextTable,它的容量是原來的兩倍，這個操作是單線程完成的。這個單線程的保證是通過RESIZE_STAMP_SHIFT這個常量經過一次運算來保證的，這個地方在后面會有提到；
- 第二個部分就是將原來table中的元素復制到nextTable中，這里允許多線程進行操作。
先來看一下單線程是如何完成的：
它的大體思想就是遍歷、復制的過程。首先根據運算得到需要遍歷的次數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倍 ，完成擴容。
再看一下多線程是如何完成的：
在代碼的69行有一個判斷，如果遍歷到的節點是forward節點，就向后繼續遍歷，再加上給節點上鎖的機制，就完成了多線程的控制。多線程遍歷節點，處理了一個節點，就把對應點的值set為forward，另一個線程看到forward，就向后遍歷。這樣交叉就完成了復制工作。而且還很好的解決了線程安全的問題。 這個方法的設計實在是讓我膜拜。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
|
/**
* 一個過渡的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 range
if
(nextTab ==
null
) {
// initiating
try
{
@SuppressWarnings
(
"unchecked"
)
Node<K,V>[] nt = (Node<K,V>[])
new
Node<?,?>[n <<
1
];
//構造一個nextTable對象 它的容量是原來的兩倍
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);
//構造一個連節點指針 用於標志位
boolean
advance =
true
;
//並發擴容的關鍵屬性 如果等於true 說明這個節點已經處理過
boolean
finishing =
false
;
// to ensure sweep before committing nextTab
for
(
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 清空臨時對象nextTable
nextTable =
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 processed
else
{
//節點上鎖
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);
else
hn =
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;
else
loTail.next = p;
loTail = p;
++lc;
}
else
{
if
((p.prev = hiTail) ==
null
)
hi = p;
else
hiTail.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
;
}
}
}
}
}
}
|
2.6 Put方法
前面的所有的介紹其實都為這個方法做鋪墊。ConcurrentHashMap最常用的就是put和get兩個方法。現在來介紹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，就把這個鏈表轉換成紅黑樹。如果這個節點的類型已經是樹節點的話，直接調用樹節點的插入方法進行插入新的值。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
|
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();
//計算hash值
int
hash = spread(key.hashCode());
int
binCount =
0
;
//死循環 何時插入成功 何時跳出
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();
//根據hash值計算出在table里面的位置
else
if
((f = tabAt(tab, i = (n -
1
) & hash)) ==
null
) {
//如果這個位置沒有值 ，直接放進去，不需要加鎖
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
;
//結點上鎖 這里的結點可以理解為hash值相同組成的鏈表的頭結點
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
addCount(1L, binCount);
return
null
;
}
|
我們可以發現JDK8中的實現也是鎖分離的思想，只是鎖住的是一個Node，而不是JDK7中的Segment，而鎖住Node之前的操作是無鎖的並且也是線程安全的，建立在之前提到的3個原子操作上。
2.6.1 helpTransfer方法
這是一個協助擴容的方法。這個方法被調用的時候，當前ConcurrentHashMap一定已經有了nextTable對象，首先拿到這個nextTable對象，調用transfer方法。回看上面的transfer方法可以看到，當本線程進入擴容方法的時候會直接進入復制階段。
2.6.2 treeifyBin方法
這個方法用於將過長的鏈表轉換為TreeBin對象。但是他並不是直接轉換，而是進行一次容量判斷，如果容量沒有達到轉換的要求，直接進行擴容操作並返回；如果滿足條件才鏈表的結構抓換為TreeBin ，這與HashMap不同的是，它並沒有把TreeNode直接放入紅黑樹，而是利用了TreeBin這個小容器來封裝所有的TreeNode.
2.7 get方法
get方法比較簡單，給定一個key來確定value的時候，必須滿足兩個條件 key相同 hash值相同，對於節點可能在鏈表或樹上的情況，需要分別去查找。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
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
;
}
|
2.8 Size相關的方法
對於ConcurrentHashMap來說，這個table里到底裝了多少東西其實是個不確定的數量，因為不可能在調用size()方法的時候像GC的“stop the world”一樣讓其他線程都停下來讓你去統計，因此只能說這個數量是個估計值。對於這個估計值，ConcurrentHashMap也是大費周章才計算出來的。
2.8.1 輔助定義
為了統計元素個數，ConcurrentHashMap定義了一些變量和一個內部類
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
/**
* 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; }
}
/******************************************/
/**
* 實際上保存的是hashmap中的元素個數 利用CAS鎖進行更新
但它並不用返回當前hashmap的元素個數
*/
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;
|
2.8.2 mappingCount與Size方法
mappingCount與size方法的類似 從Java工程師給出的注釋來看，應該使用mappingCount代替size方法 兩個方法都沒有直接返回basecount 而是統計一次這個值，而這個值其實也是一個大概的數值，因此可能在統計的時候有其他線程正在執行插入或刪除操作。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
|
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;
}
|
2.8.3 addCount方法
在put方法結尾處調用了addCount方法，把當前ConcurrentHashMap的元素個數+1這個方法一共做了兩件事,更新baseCount的值，檢測是否進行擴容。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
|
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 則需要檢驗是否需要進行擴容操作
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
;
//如果已經有其他線程在執行擴容操作
if
(U.compareAndSwapInt(
this
, SIZECTL, sc, sc +
1
))
transfer(tab, nt);
}
//當前線程是唯一的或是第一個發起擴容的線程 此時nextTable=null
else
if
(U.compareAndSwapInt(
this
, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) +
2
))
transfer(tab,
null
);
s = sumCount();
}
}
}
|
總結
JDK6,7中的ConcurrentHashmap主要使用Segment來實現減小鎖粒度，把HashMap分割成若干個Segment，在put的時候需要鎖住Segment，get時候不加鎖，使用volatile來保證可見性，當要統計全局時（比如size），首先會嘗試多次計算modcount來確定，這幾次嘗試中，是否有其他線程進行了修改操作，如果沒有，則直接返回size。如果有，則需要依次鎖住所有的Segment來計算。
jdk7中ConcurrentHashmap中，當長度過長碰撞會很頻繁，鏈表的增改刪查操作都會消耗很長的時間，影響性能,所以jdk8 中完全重寫了concurrentHashmap,代碼量從原來的1000多行變成了 6000多 行，實現上也和原來的分段式存儲有很大的區別。
主要設計上的變化有以下幾點:
- 不采用segment而采用node，鎖住node來實現減小鎖粒度。
- 設計了MOVED狀態 當resize的中過程中 線程2還在put數據，線程2會幫助resize。
- 使用3個CAS操作來確保node的一些操作的原子性，這種方式代替了鎖。
- sizeCtl的不同值來代表不同含義，起到了控制的作用。
至於為什么JDK8中使用synchronized而不是ReentrantLock，我猜是因為JDK8中對synchronized有了足夠的優化吧。
Reference：
1. http://www.jianshu.com/p/4806633fcc55
2. https://www.zhihu.com/question/22438589
3. http://blog.csdn.net/u010723709/article/details/48007881