JDK 1.8 ConcurrentHashMap 原始碼剖析
阿新 • • 發佈：2019-02-07
轉載兩篇不錯的文章：
第一篇：
前言
HashMap是我們平時開發過程中用的比較多的集合，但它是非執行緒安全的，在涉及到多執行緒併發的情況，進行put操作有可能會引起死迴圈，導致CPU利用率接近100%。
final HashMap<String, String> map = new HashMap<String, String>(2); for (int i = 0; i < 10000; i++) { new Thread(new Runnable() { @Override public void run() { map.put(UUID.randomUUID().toString(), ""); } }).start(); }
解決方案有Hashtable和Collections.synchronizedMap(hashMap)，不過這兩個方案基本上是對讀寫進行加鎖操作，一個執行緒在讀寫元素，其餘執行緒必須等待，效能可想而知。
所以，Doug Lea給我們帶來了併發安全的ConcurrentHashMap，它的實現是依賴於 Java 記憶體模型，所以我們在瞭解 ConcurrentHashMap 的之前必須瞭解一些底層的知識：
本文原始碼是JDK8的版本，與之前的版本有較大差異。
JDK1.6分析
ConcurrentHashMap採用 分段鎖的機制，實現併發的更新操作，底層採用陣列+連結串列+紅黑樹的儲存結構。
其包含兩個核心靜態內部類 Segment和HashEntry。
- Segment繼承ReentrantLock用來充當鎖的角色，每個 Segment 物件守護每個雜湊對映表的若干個桶。
- HashEntry 用來封裝對映表的鍵 / 值對；
- 每個桶是由若干個 HashEntry 物件連結起來的連結串列。
一個 ConcurrentHashMap 例項中包含由若干個 Segment 物件組成的陣列，下面我們通過一個圖來演示一下 ConcurrentHashMap 的結構：
ConcurrentHashMap儲存結構.png
JDK1.8分析
1.8的實現已經拋棄了Segment分段鎖機制，利用CAS+Synchronized來保證併發更新的安全，底層依然採用陣列+連結串列+紅黑樹的儲存結構。
Paste_Image.png
重要概念
在開始之前，有些重要的概念需要介紹一下：
- table：預設為null，初始化發生在第一次插入操作，預設大小為16的陣列，用來儲存Node節點資料，擴容時大小總是2的冪次方。
- nextTable：預設為null，擴容時新生成的陣列，其大小為原陣列的兩倍。
- sizeCtl ：預設為0，用來控制table的初始化和擴容操作，具體應用在後續會體現出來。
- -1 代表table正在初始化
- -N 表示有N-1個執行緒正在進行擴容操作
- 其餘情況：
1、如果table未初始化，表示table需要初始化的大小。
2、如果table初始化完成，表示table的容量，預設是table大小的0.75倍，居然用這個公式算0.75（n - (n >>> 2)）。
- Node：儲存key，value及key的hash值的資料結構。
其中value和next都用volatile修飾，保證併發的可見性。class Node<K,V> implements Map.Entry<K,V> { final int hash; final K key; volatile V val; volatile Node<K,V> next; ... 省略部分程式碼 } - ForwardingNode：一個特殊的Node節點，hash值為-1，其中儲存nextTable的引用。
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; } }
- 只有table發生擴容的時候，ForwardingNode才會發揮作用，作為一個佔位符放在table中表示當前節點為null或則已經被移動。
例項初始化
例項化ConcurrentHashMap時帶引數時，會根據引數調整table的大小，假設引數為100，最終會調整成256，確保table的大小總是2的冪次方，演算法如下：
ConcurrentHashMap<String, String> hashMap = new ConcurrentHashMap<>(100);
private static final int tableSizeFor(int c) {
int n = c - 1;
n |= n >>> 1;
n |= n >>> 2;
n |= n >>> 4;
n |= n >>> 8;
n |= n >>> 16;
return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
}注意，ConcurrentHashMap在建構函式中只會初始化sizeCtl值，並不會直接初始化table，而是延緩到第一次put操作。
table初始化
前面已經提到過，table初始化操作會延緩到第一次put行為。但是put是可以併發執行的，Doug Lea是如何實現table只初始化一次的？讓我們來看看原始碼的實現。
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
//如果一個執行緒發現sizeCtl<0，意味著另外的執行緒執行CAS操作成功，當前執行緒只需要讓出cpu時間片
if ((sc = sizeCtl) < 0)
Thread.yield(); // lost initialization race; just spin
else if (U.compareAndSwapInt(this, SIZECTL, sc, -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);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}sizeCtl預設為0，如果ConcurrentHashMap例項化時有傳引數，sizeCtl會是一個2的冪次方的值。所以執行第一次put操作的執行緒會執行Unsafe.compareAndSwapInt方法修改sizeCtl為-1，有且只有一個執行緒能夠修改成功，其它執行緒通過Thread.yield()讓出CPU時間片等待table初始化完成。
put操作
假設table已經初始化完成，put操作採用CAS+synchronized實現併發插入或更新操作，具體實現如下。
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
if (tab == null || (n = tab.length) == 0)
tab = initTable();
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);
...省略部分程式碼
}
addCount(1L, binCount);
return null;
}- hash演算法
static final int spread(int h) {return (h ^ (h >>> 16)) & HASH_BITS;} - table中定位索引位置，n是table的大小
int index = (n - 1) & hash - 獲取table中對應索引的元素f。
Doug Lea採用Unsafe.getObjectVolatile來獲取，也許有人質疑，直接table[index]不可以麼，為什麼要這麼複雜？
在java記憶體模型中，我們已經知道每個執行緒都有一個工作記憶體，裡面儲存著table的副本，雖然table是volatile修飾的，但不能保證執行緒每次都拿到table中的最新元素，Unsafe.getObjectVolatile可以直接獲取指定記憶體的資料，保證了每次拿到資料都是最新的。 - 如果f為null，說明table中這個位置第一次插入元素，利用Unsafe.compareAndSwapObject方法插入Node節點。
- 如果CAS成功，說明Node節點已經插入，隨後addCount(1L, binCount)方法會檢查當前容量是否需要進行擴容。
- 如果CAS失敗，說明有其它執行緒提前插入了節點，自旋重新嘗試在這個位置插入節點。
- 如果f的hash值為-1，說明當前f是ForwardingNode節點，意味有其它執行緒正在擴容，則一起進行擴容操作。
- 其餘情況把新的Node節點按連結串列或紅黑樹的方式插入到合適的位置，這個過程採用同步內建鎖實現併發，程式碼如下:
synchronized (f) { if (tabAt(tab, i) == f) { if (fh >= 0) { binCount = 1; for (Node<K,V> e = f;; ++binCount) { K ek; 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; } } } }
在節點f上進行同步，節點插入之前，再次利用tabAt(tab, i) == f判斷，防止被其它執行緒修改。- 如果f.hash >= 0，說明f是連結串列結構的頭結點，遍歷連結串列，如果找到對應的node節點，則修改value，否則在連結串列尾部加入節點。
- 如果f是TreeBin型別節點，說明f是紅黑樹根節點，則在樹結構上遍歷元素，更新或增加節點。
- 如果連結串列中節點數binCount >= TREEIFY_THRESHOLD(預設是8)，則把連結串列轉化為紅黑樹結構。
table擴容
當table容量不足的時候，即table的元素數量達到容量閾值sizeCtl，需要對table進行擴容。
整個擴容分為兩部分：
- 構建一個nextTable，大小為table的兩倍。
- 把table的資料複製到nextTable中。
這兩個過程在單執行緒下實現很簡單，但是ConcurrentHashMap是支援併發插入的，擴容操作自然也會有併發的出現，這種情況下，第二步可以支援節點的併發複製，這樣效能自然提升不少，但實現的複雜度也上升了一個臺階。
先看第一步，構建nextTable，毫無疑問，這個過程只能只有單個執行緒進行nextTable的初始化，具體實現如下：
private final void addCount(long x, int check) {
... 省略部分程式碼
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);
}
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
s = sumCount();
}
}
}通過Unsafe.compareAndSwapInt修改sizeCtl值，保證只有一個執行緒能夠初始化nextTable，擴容後的陣列長度為原來的兩倍，但是容量是原來的1.5。
節點從table移動到nextTable，大體思想是遍歷、複製的過程。
- 首先根據運算得到需要遍歷的次數i，然後利用tabAt方法獲得i位置的元素f，初始化一個forwardNode例項fwd。
- 如果f == null，則在table中的i位置放入fwd，這個過程是採用Unsafe.compareAndSwapObjectf方法實現的，很巧妙的實現了節點的併發移動。
- 如果f是連結串列的頭節點，就構造一個反序連結串列，把他們分別放在nextTable的i和i+n的位置上，移動完成，採用Unsafe.putObjectVolatile方法給table原位置賦值fwd。
- 如果f是TreeBin節點，也做一個反序處理，並判斷是否需要untreeify，把處理的結果分別放在nextTable的i和i+n的位置上，移動完成，同樣採用Unsafe.putObjectVolatile方法給table原位置賦值fwd。
遍歷過所有的節點以後就完成了複製工作，把table指向nextTable，並更新sizeCtl為新陣列大小的0.75倍 ，擴容完成。
紅黑樹構造
注意：如果連結串列結構中元素超過TREEIFY_THRESHOLD閾值，預設為8個，則把連結串列轉化為紅黑樹，提高遍歷查詢效率。
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}接下來我們看看如何構造樹結構，程式碼如下：
private final void treeifyBin(Node<K,V>[] tab, int index) {
Node<K,V> b; int n, sc;
if (tab != null) {
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
tryPresize(n << 1);
else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
synchronized (b) {
if (tabAt(tab, index) == b) {
TreeNode<K,V> hd = null, tl = null;
for (Node<K,V> e = b; e != null; e = e.next) {
TreeNode<K,V> p =
new TreeNode<K,V>(e.hash, e.key, e.val,
null, null);
if ((p.prev = tl) == null)
hd = p;
else
tl.next = p;
tl = p;
}
setTabAt(tab, index, new TreeBin<K,V>(hd));
}
}
}
}
}可以看出，生成樹節點的程式碼塊是同步的，進入同步程式碼塊之後，再次驗證table中index位置元素是否被修改過。
1、根據table中index位置Node連結串列，重新生成一個hd為頭結點的TreeNode連結串列。
2、根據hd頭結點，生成TreeBin樹結構，並把樹結構的root節點寫到table的index位置的記憶體中，具體實現如下：
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);
}主要根據Node節點的hash值大小構建二叉樹。這個紅黑樹的構造過程實在有點複雜，感興趣的同學可以看看原始碼。
get操作
get操作和put操作相比，顯得簡單了許多。
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
int h = spread(key.hashCode());
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
while ((e = e.next) != null) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}總結
ConcurrentHashMap 是一個併發雜湊對映表的實現，它允許完全併發的讀取，並且支援給定數量的併發更新。相比於 HashTable 和同步包裝器包裝的 HashMap，使用一個全域性的鎖來同步不同執行緒間的併發訪問，同一時間點，只能有一個執行緒持有鎖，也就是說在同一時間點，只能有一個執行緒能訪問容器，這雖然保證多執行緒間的安全併發訪問，但同時也導致對容器的訪問變成序列化的了。
1.6中採用ReentrantLock 分段鎖的方式，使多個執行緒在不同的segment上進行寫操作不會發現阻塞行為;1.8中直接採用了內建鎖synchronized，難道是因為1.8的虛擬機器對內建鎖已經優化的足夠快了？
本文首寫於有道雲筆記，並在小組分享會分享，先整理髮布，希望和大家交流探討。雲筆記地址
概述： 1、設計首要目的：維護併發可讀性（get、迭代相關）；次要目的：使空間消耗比HashMap相同或更好，且支援多執行緒高效率的初始插入（empty table）。 2、HashTable執行緒安全，但採用synchronized，多執行緒下效率低下。執行緒1put時，執行緒2無法put或get。 實現原理： 鎖分離： 在HashMap的基礎上，將資料分段儲存，ConcurrentHashMap由多個Segment組成，每個Segment都有把鎖。Segment下包含很多Node，也就是我們的鍵值對了。 如果還停留在鎖分離、Segment，那已經out了。 Segment雖保留，但已經簡化屬性，僅僅是為了相容舊版本。- CAS演算法；unsafe.compareAndSwapInt(this, valueOffset, expect, update); CAS(Compare And Swap)，意思是如果valueOffset位置包含的值與expect值相同，則更新valueOffset位置的值為update，並返回true，否則不更新，返回false。
- 與Java8的HashMap有相通之處，底層依然由“陣列”+連結串列+紅黑樹；
- 底層結構存放的是TreeBin物件，而不是TreeNode物件；
- CAS作為知名無鎖演算法，那ConcurrentHashMap就沒用鎖了麼？當然不是，hash值相同的連結串列的頭結點還是會synchronized上鎖。
private static final int MAXIMUM_CAPACITY = 1 << 30; // 2的30次方=1073741824
private static final intDEFAULT_CAPACITY = 16;
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8; // MAX_VALUE=2^31-1=2147483647
private static finalint DEFAULT_CONCURRENCY_LEVEL = 16;
private static final float LOAD_FACTOR = 0.75f;
static final int TREEIFY_THRESHOLD = 8; // 連結串列轉樹閥值，大於8時
static final int UNTREEIFY_THRESHOLD = 6; //樹轉連結串列閥值，小於等於6（tranfer時，lc、hc=0兩個計數器分別++記錄原bin、新binTreeNode數量，<=UNTREEIFY_THRESHOLD 則untreeify(lo)）。【僅在擴容tranfer時才可能樹轉連結串列】
static final int MIN_TREEIFY_CAPACITY = 64;
private static final int MIN_TRANSFER_STRIDE = 16;
private static int RESIZE_STAMP_BITS = 16;
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1; // 2^15-1，help resize的最大執行緒數
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS; // 32-16=16，sizeCtl中記錄size大小的偏移量
static final int MOVED = -1; // hash for forwarding nodes（forwarding nodes的hash值）、標示位
static final int TREEBIN = -2; // hash for roots of trees（樹根節點的hash值）
static final int RESERVED = -3; // hash for transient reservations（ReservationNode的hash值）
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
static final int NCPU = Runtime.getRuntime().availableProcessors(); // 可用處理器數量
/**
* 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.
*/
private transient volatile int sizeCtl;
sizeCtl是控制識別符號，不同的值表示不同的意義。
負數代表正在進行初始化或擴容操作
-1代表正在初始化
-N 表示有N-1個執行緒正在進行擴容操作
正數或0代表hash表還沒有被初始化，這個數值表示初始化或下一次進行擴容的大小，類似於擴容閾值。它的值始終是當前ConcurrentHashMap容量的0.75倍，這與loadfactor是對應的。實際容量>=sizeCtl，則擴容。部分建構函式：
- public ConcurrentHashMap(int initialCapacity,
- float loadFactor, int concurrencyLevel) {
- if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
- thrownew IllegalArgumentException();
- if (initialCapacity < concurrencyLevel) // Use at least as many bins
- initialCapacity = concurrencyLevel; // as estimated threads
- long size = (long)(1.0 + (long)initialCapacity / loadFactor);
- int cap = (size >= (long)MAXIMUM_CAPACITY) ?
- MAXIMUM_CAPACITY : tableSizeFor((int)size);
- this.sizeCtl = cap;
- }
concurrencyLevel： concurrencyLevel，能夠同時更新ConccurentHashMap且不產生鎖競爭的最大執行緒數，在Java8之前實際上就是ConcurrentHashMap中的分段鎖個數，即Segment[]的陣列長度。正確地估計很重要，當低估，資料結構將根據額外的競爭，從而導致執行緒試圖寫入當前鎖定的段時阻塞；相反，如果高估了併發級別，你遇到過大的膨脹，由於段的不必要的數量; 這種膨脹可能會導致效能下降，由於高數快取未命中。 在Java8裡，僅僅是為了相容舊版本而保留。唯一的作用就是保證構造map時初始容量不小於concurrencyLevel。 原始碼122行： Also, for compatibility with previous versions of this class, constructors may optionally specify an expected {@code concurrencyLevel} as an additional hint for internal sizing. 原始碼482行： Mainly: We leave untouched but unused constructor arguments refering to concurrencyLevel .…… …… 1、重要屬性： 1.1 Node：
- staticclass Node<K,V> implements Map.Entry<K,V> {
- finalint hash;
- final K key;
- volatile V val; // Java8增加volatile，保證可見性
- volatile Node<K,V> next;
- Node(inthash, K key, V val, Node<K,V> next) {
- this.hash = hash;
- this.key = key;
- this.val = val;
- this.next = next;
- }
- publicfinal K getKey() { return key; }
- publicfinal V getValue() { return val; }
- // HashMap呼叫Objects.hashCode()，最終也是呼叫Object.hashCode()；效果一樣
- publicfinalint hashCode() { returnkey.hashCode() ^ val.hashCode(); }
- publicfinal String toString(){ returnkey + "=" + val; }
- publicfinal V setValue(V value) { // 不允許修改value值，HashMap允許
- thrownew UnsupportedOperationException();
- }
- // HashMap使用if (o == this)，且巢狀if；concurrent使用&&
- publicfinalboolean equals(Object o) {
- Object k, v, u; Map.Entry<?,?> e;
- return ((oinstanceof 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(inth, Object k) { // 增加find方法輔助get方法
- Node<K,V> e = this;
- if (k != null) {
- do {
- K ek;
- if (e.hash == h &&
- ((ek = e.key) == k || (ek != null && k.equals(ek))))
- returne;
- } while ((e = e.next) != null);
- }
- returnnull;
- }
- }
- // Nodes for use in TreeBins，連結串列>8，才可能轉為TreeNode.
- // HashMap的TreeNode繼承至LinkedHashMap.Entry；而這裡繼承至自己實現的Node，將帶有next指標，便於treebin訪問。
- staticfinalclass TreeNode<K,V> extends Node<K,V> {
- TreeNode<K,V> parent; // red-black tree links
- TreeNode<K,V> left;
- TreeNode<K,V> right;
- TreeNode<K,V> prev; // needed to unlink next upon deletion
- boolean red;
- TreeNode(inthash, K key, V val, Node<K,V> next,
- TreeNode<K,V> parent) {
- super(hash, key, val, next);
- this.parent = parent;
- }
- Node<K,V> find(inth, Object k) {
- return findTreeNode(h, k, null);
- }
- /**
- * Returns the TreeNode (or null if not found) for the given key
- * starting at given root.
- */// 查詢hash為h，key為k的節點
- final TreeNode<K,V> findTreeNode(int h, Object k, Class<?> kc) {
- if (k != null) { // 比HMap增加判空
- TreeNode<K,V> p = this;
- do {
- intph, dir; K pk; TreeNode<K,V> q;
- TreeNode<K,V> pl = p.left, pr = p.right;
- if ((ph = p.hash) > h)
- p = pl;
- elseif (ph < h)
- p = pr;
- elseif ((pk = p.key) == k || (pk != null && k.equals(pk)))
- returnp;
- elseif (pl == null)
- p = pr;
- elseif (pr == null)
- p = pl;
- elseif ((kc != null ||
- (kc = comparableClassFor(k)) != null) &&
- (dir = compareComparables(kc, k, pk)) != 0)
- p = (dir < 0) ? pl : pr;
- elseif ((q = pr.findTreeNode(h, k, kc)) != null)
- returnq;
- else
- p = pl;
- } while (p != null);
- }
- returnnull;
- }
- }
- // 和HashMap相比，這裡的TreeNode相當簡潔；ConcurrentHashMap連結串列轉樹時，並不會直接轉，正如註釋（Nodes for use in TreeBins）所說，只是把這些節點包裝成TreeNode放到TreeBin中，再由TreeBin來轉化紅黑樹。
1.3 TreeBin
- // TreeBin用於封裝維護TreeNode，包含putTreeVal、lookRoot、UNlookRoot、remove、balanceInsetion、balanceDeletion等方法，這裡只分析其建構函式。
- // 當連結串列轉樹時，用於封裝TreeNode，也就是說，ConcurrentHashMap的紅黑樹存放的時TreeBin，而不是treeNode。
- TreeBin(TreeNode<K,V> b) {
- super(TREEBIN, null, null, null);//hash值為常量TREEBIN=-2,表示roots of trees
- 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;
- inth = x.hash;
- Class<?> kc = null;
- for (TreeNode<K,V> p = r;;) {
- intdir, ph;
- K pk = p.key;
- if ((ph = p.hash) > h)
- dir = -1;
- elseif (ph < h)
- dir = 1;
- elseif ((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);
- }
1.4 treeifyBin
- /**
- * Replaces all linked nodes in bin at given index unless table is
- * too small, in which case resizes instead.連結串列轉樹
- */
- privatefinalvoid treeifyBin(Node<K,V>[] tab, int index) {
- Node<K,V> b; intn, sc;
- if (tab != null) {
- if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
- tryPresize(n << 1); // 容量<64，則table兩倍擴容，不轉樹了
- elseif ((b = tabAt(tab, index)) != null && b.hash >= 0) {
- synchronized (b) { // 讀寫鎖
- if (tabAt(tab, index) == b) {
- TreeNode<K,V> hd = null, tl = null;
- for (Node<K,V> e = b; e != null; e = e.next) {
- TreeNode<K,V> p =
- new TreeNode<K,V>(e.hash, e.key, e.val,
- null, null);
- if ((p.prev = tl) == null)
- hd = p;
- else
- tl.next = p;
- tl = p;
- }
- setTabAt(tab, index, new TreeBin<K,V>(hd));
- }
- }
- }
- }
- }
1.5 ForwardingNode
- // A node inserted at head of bins during transfer operations.連線兩個table
- // 並不是我們傳統的包含key-value的節點，只是一個標誌節點，並且指向nextTable，提供find方法而已。生命週期：僅存活於擴容操作且bin不為null時，一定會出現在每個bin的首位。
- staticfinalclass ForwardingNode<K,V> extends Node<K,V> {
- final Node<K,V>[] nextTable;
- ForwardingNode(Node<K,V>[] tab) {
- super(MOVED, null, null, null); // 此節點hash=-1，key、value、next均為null
- this.nextTable = tab;
- }
- Node<K,V> find(int h, Object k) {
- // 查nextTable節點，outer避免深度遞迴
- outer: for (Node<K,V>[] tab = nextTable;;) {
- Node<K,V> e; intn;
- if (k == null || tab == null || (n = tab.length) == 0 ||
- (e = tabAt(tab, (n - 1) & h)) == null)
- returnnull;
- for (;;) { // CAS演算法多和死迴圈搭配！直到查到或null
- int eh; K ek;
- if ((eh = e.hash) == h &&
- ((ek = e.key) == k || (ek != null && k.equals(ek))))
- returne;
- 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)
- returnnull;
- }
- }
- }
- }
1.6 3個原子操作（呼叫頻率很高）
- @SuppressWarnings("unchecked") // ASHIFT等均為private static final
- staticfinal <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) { // 獲取索引i處Node
- return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
- }
- // 利用CAS演算法設定i位置上的Node節點（將c和table[i]比較，相同則插入v）。
- staticfinal <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);
- }
- // 設定節點位置的值，僅在上鎖區被呼叫
- staticfinal <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
- U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
- }
1.7 Unsafe
- //在原始碼的6277行到最後，有著ConcurrentHashMap中極為重要的幾個屬性（SIZECTL），unsafe靜態塊控制其修改行為。Java8中，大量運用CAS進行變數、屬性的無鎖修改，大大提高效能。
- // Unsafe mechanics
- privatestaticfinal sun.misc.Unsafe U;
- privatestaticfinallong SIZECTL;
- privatestaticfinallong TRANSFERINDEX;
- privatestaticfinallong BASECOUNT;
- privatestaticfinallong CELLSBUSY;
- privatestaticfinallong CELLVALUE;
- privatestaticfinallong ABASE;
- privatestaticfinalint 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);
- intscale = U.arrayIndexScale(ak);
- if ((scale & (scale - 1)) != 0)
- thrownew Error("data type scale not a power of two");
- ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
- } catch (Exception e) {
- thrownew Error(e);
- }
- }
1.8 擴容相關 tryPresize在putAll以及treeifyBin中呼叫