更多文章請關(guān)注公號（ID：CodeReading）
本文是建立在上篇HashMap源碼分析的基礎(chǔ)上。其中的一些重復(fù)的方法和知識點(diǎn)不會再贅述。有疑惑的同學(xué)可以移步到上一篇文章。依舊以jdk1.8源碼為基礎(chǔ)來講解ConcurrentHashMap。它的大體結(jié)構(gòu)與HashMap相同，table容量同樣要求是2的冪次。
HashMap高效快捷，但不安全，特別是2020年，安全很重要。目前市面上有3種提供安全的map的方式，分別是
- hashtable：相對古老的線程安全機(jī)制。任一時(shí)間只有一個(gè)線程能寫操作?，F(xiàn)在基本棄用，被效率更高的ConcurrentHashMap替代。
- synchronizedMap：Collections中的內(nèi)部類，可以把普通的map轉(zhuǎn)成線程安全的map。原理就是在操作對象上加synchronized。
- ConcurrentHashMap：線程安全的HashMap。
如何做到線程安全
多線程并發(fā)帶來的問題目前總體有2種解決機(jī)制。
- 悲觀機(jī)制：認(rèn)為最壞的結(jié)果肯定發(fā)生，從開始就設(shè)置一定規(guī)則最大限度減少發(fā)生概率，有點(diǎn)以防萬一、未雨綢繆的意思。比如安全套，肯定不會每次都中，可萬一呢？并發(fā)的悲觀實(shí)現(xiàn)一般采用鎖，java中的synchronized關(guān)鍵字也被廣泛應(yīng)用在多線程并發(fā)的場景中。但是鎖會降低程序性能。
- 樂觀機(jī)制：認(rèn)為結(jié)果總是好的，先干了再說，不行再想辦法：重試，補(bǔ)救，版本號控制等。意外懷孕的癡男怨女們可能就太樂觀了，事后只能采取補(bǔ)救措施。CAS就是樂觀機(jī)制。
ConcurrentHashMap中主要采用的CAS+自旋，改成功就改，改不成功繼續(xù)試（自旋）。也有synchronized配合使用。
CAS & Unsafe
CAS的全稱是Compare And Swap，即比較交換，它也是JUC的基礎(chǔ)。我們只是簡單介紹它的原理，細(xì)節(jié)問題需要同學(xué)們另行研究。
/*
* v-表示要更新的變量,e-表示預(yù)期值,n-新值。
* 方法的目的是給變量v修改值。
* 如果變量v的值和預(yù)期值e相等就把v的值改成n，返回true，如果不相等就返回false，有其他線程修改該值。
* 這里可能出現(xiàn)ABA問題（從A變成B又變回A，造成變量沒變得假象），但java做了很多優(yōu)化?？梢院雎圆挥?jì)。
*/
boolean CAS(v,e,n)
cas流程很好理解，但在多cpu多線程的情況下會不會不安全，放心安全。java的cas其實(shí)是通過Unsafe類方法調(diào)用cpu的一條cas原子指令。操作系統(tǒng)本身是對內(nèi)存操作做了線程安全的，篇幅太少也說不清楚，這里大家可以自行研究一下JMM，JMM不是本文重點(diǎn)。這里只要知道結(jié)論就是CAS可以保證原子性。不過它只提供單個(gè)變量的原子性，多個(gè)變量的原子性還需要借助synchronized。
Unsafe是java里的另類，java有個(gè)特點(diǎn)是安全，并不允許程序員直接操作內(nèi)存，而Unsafe類可以，才有條件執(zhí)行CAS方法。但是不建議大家使用Unsafe.class，因?yàn)椴话踩瑂un公司有計(jì)劃取消Unsafe。
源碼解析
sizeCtl & constructor
ConcurrentHashMap和HashMap在各自的構(gòu)造函數(shù)中都沒有做數(shù)組初始化，初始化放在了第一次添加元素中。值得注意的是ConcurrentHashMap中有一個(gè)屬性sizeCtl特別重要，理清楚它的變化，也就理解了整個(gè)Map源碼的流程。下面是它的說明
/**
* 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.
* <p>
* 控制標(biāo)識符，用來控制table的初始化和擴(kuò)容的操作，不同的值有不同的含義
* <p>
* 1. 當(dāng)為負(fù)數(shù)時(shí)：-1代表正在初始化，-N代表有N-1個(gè)線程正在進(jìn)行擴(kuò)容
* <p>
* 2.當(dāng)為0時(shí)：代表當(dāng)時(shí)的table還沒有被初始化
* <p>
* 3.當(dāng)為正數(shù)時(shí)：未初始化表示的是初始化數(shù)組的初始容量，如果已經(jīng)初始化，
* 記錄的是擴(kuò)容的閾值（達(dá)到閾值進(jìn)行擴(kuò)容）
*/
private transient volatile int sizeCtl;
再看一下ConcurrentHashMap帶初始化容量的代碼
/**
* Creates a new, empty map with an initial table size
* accommodating the specified number of elements without the need
* to dynamically resize.
*
* @param initialCapacity The implementation performs internal
* sizing to accommodate this many elements.
* @throws IllegalArgumentException if the initial capacity of
* elements is negative
*
* 此時(shí)sizeCtl記錄的就是數(shù)組的初始化容量
*
* 比如initialCapacity=5
* 調(diào)用tableSizeFor（5+5/2+1）==tableSizeFor(8)
*/
public ConcurrentHashMap(int initialCapacity) {
if (initialCapacity < 0)
throw new IllegalArgumentException();
int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
MAXIMUM_CAPACITY :
tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
this.sizeCtl = cap;
}
/**
* Returns a power of two table size for the given desired capacity.
* See Hackers Delight, sec 3.2
* 返回一個(gè)大于等于c的2的冪次方數(shù)
*
* 當(dāng)c=8時(shí)
* n = c-1=7
* 接下來驗(yàn)算最終結(jié)果
* 0000 0000 0000 0000 0000 0000 0000 0111
* >>> 1
* = 0000 0000 0000 0000 0000 0000 0000 0011
* | 0000 0000 0000 0000 0000 0000 0000 0111
* = 0000 0000 0000 0000 0000 0000 0000 0111
* >>> 2
* = 0000 0000 0000 0000 0000 0000 0000 0001
* | 0000 0000 0000 0000 0000 0000 0000 0111
* = 0000 0000 0000 0000 0000 0000 0000 0111
* >>> 4
* = 0000 0000 0000 0000 0000 0000 0000 0000
* | 0000 0000 0000 0000 0000 0000 0000 0111
* = 0000 0000 0000 0000 0000 0000 0000 0111
* 下面再 >>> 8 和 >>> 16后的二進(jìn)制都是0
* 所以最終結(jié)果就是111，也就是7最后返回結(jié)果再+1，等于8
*
* 總結(jié) 右移一共1+2+4+8+16=31位，和與之對應(yīng) | 運(yùn)算
* 最終把n的二進(jìn)制中所有1移到低位。新的數(shù)高位都是0，低位都是1。這樣格式的數(shù)在HashMap中提到過，就是2的冪次-1。
* 最后結(jié)果是這個(gè)數(shù)+1，那就是2的冪次。
*/
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;
}
當(dāng)我們new ConcurrentHashMap(c) 時(shí)，初始化容量并不是c，而是一個(gè)大于等于c的2的冪次方數(shù)。我們利用發(fā)射來驗(yàn)證下
public static void main(String[] args) {
ConcurrentHashMap concurrentHashMap = new ConcurrentHashMap(5);
Class clazz = concurrentHashMap.getClass();
try {
Field field = clazz.getDeclaredField("sizeCtl");
//打開私有訪問
field.setAccessible(true);
//獲取屬性
String name = field.getName();
//獲取屬性值
Object value = field.get(concurrentHashMap);
System.out.println("ConcurrentHashMap的初始容量為：= "+value);
} catch (Exception e) {
e.printStackTrace();
}
}
--打印結(jié)果是: Map的初始容量=8
put & putVal
/**
* Maps the specified key to the specified value in this table.
* Neither the key nor the value can be null.
*
* <p>The value can be retrieved by calling the {@code get} method
* with a key that is equal to the original key.
*
* @param key key with which the specified value is to be associated
* @param value value to be associated with the specified key
* @return the previous value associated with {@code key}, or
* {@code null} if there was no mapping for {@code key}
* @throws NullPointerException if the specified key or value is null
*/
@Override
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) {
//如果有空值或者空鍵，直接拋異常
if (key == null || value == null) {
throw new NullPointerException();
}
//兩次hash，減少hash沖突，可以均勻分布
int hash = spread(key.hashCode());
int binCount = 0;
//迭代當(dāng)前table
for (Node<K, V>[] tab = table; ; ) {
Node<K, V> f;
int n, i, fh;
//1. 如果table未初始化，先初始化
if (tab == null || (n = tab.length) == 0) {
tab = initTable();
}
//如果i位置沒有數(shù)據(jù)，cas插入
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//cas和外側(cè)else if條件形成雙保險(xiǎn)，保證數(shù)據(jù)安全
if (casTabAt(tab, i, null,
new Node<K, V>(hash, key, value, null))) {
break; // no lock when adding to empty bin
}
}
//2. hash值是MOVED表示數(shù)組正在擴(kuò)容，則協(xié)助擴(kuò)容，先擴(kuò)容在新加元素
else if ((fh = f.hash) == MOVED) {
tab = helpTransfer(tab, f);
} else {
//hash計(jì)算的bucket不為空，且當(dāng)前沒有處于擴(kuò)容操作，進(jìn)行元素添加
V oldVal = null;
//對當(dāng)前bucket進(jìn)行加鎖，保證線程安全，執(zhí)行元素添加操作
synchronized (f) {
//判斷是否為f，防止它變成tree
if (tabAt(tab, i) == f) {
//hash值>=0 表示該節(jié)點(diǎn)是鏈表結(jié)構(gòu)
if (fh >= 0) {
binCount = 1;
//e記錄的是頭節(jié)點(diǎn)
for (Node<K, V> e = f; ; ++binCount) {
K ek;
//相同的key進(jìn)行put就會覆蓋原先的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;
//紅黑樹結(jié)構(gòu)旋轉(zhuǎn)插入
if ((p = ((TreeBin<K, V>) f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent) {
p.val = value;
}
}
}
}
}
if (binCount != 0) {
//鏈表長度大于8時(shí)轉(zhuǎn)換紅黑樹
if (binCount >= TREEIFY_THRESHOLD) {
treeifyBin(tab, i);
}
if (oldVal != null) {
return oldVal;
}
break;
}
}
}
//統(tǒng)計(jì)size，并且檢查是否需要擴(kuò)容
addCount(1L, binCount);
return null;
}
putVal()總體是自旋+CAS的方式，流程和HashMap一樣。
- 自旋：
- 如果table==null，調(diào)用initTable()初始化
- 如果沒有hash碰撞就CAS添加
- 如果正在擴(kuò)容就協(xié)助擴(kuò)容
- 如果存在hash碰撞，如果是單向列表就插到bucket尾部，如果是紅黑樹就插入數(shù)結(jié)構(gòu)
- 如果鏈表bucket長度大于8，轉(zhuǎn)紅黑樹
- 如果添加成功就調(diào)用addCount()方法統(tǒng)計(jì)size，檢查是否需要擴(kuò)容
從源碼中可以看到put新元素時(shí)，如果發(fā)生hash沖突，先鎖定發(fā)生沖突的bucket，不影響其他bucket操作，達(dá)到并發(fā)安全且高效的目的。下面是`putVal
initTable，初始化table
/**
* Initializes table, using the size recorded in sizeCtl.
* 初始化table，從新記錄sizeCtl值，此時(shí)值為數(shù)組下次擴(kuò)容的閾值
*/
private final Node<K, V>[] initTable() {
Node<K, V>[] tab;
int sc;
//再次判斷空的table才能進(jìn)入初始化操作
while ((tab = table) == null || tab.length == 0) {
// sizeCtl<0，也就是下面elseif把sizeCtl設(shè)置成-1. 表示其他線程已經(jīng)在初始化了或者擴(kuò)容了，掛起當(dāng)前線程，自旋等待
if ((sc = sizeCtl) < 0) {
Thread.yield();
//CAS設(shè)置SIZECTL為-1，如果設(shè)置成功繼續(xù)執(zhí)行下面操作，如果失敗，說明此時(shí)有其他線程正在執(zhí)行操作，繼續(xù)自旋
} else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
//double check，保證線程安全，可能有線程已經(jīng)同步完了
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
Node<K, V>[] nt = (Node<K, V>[]) new Node<?, ?>[n];
table = tab = nt;
//記錄下次擴(kuò)容的大小，相當(dāng)于n-n/4=0.75n
sc = n - (n >>> 2);
}
} finally {
//此時(shí)sizeCtl的值為下次擴(kuò)容的閾值
sizeCtl = sc;
}
break;
}
}
return tab;
}
helpTransfer 協(xié)助擴(kuò)容
/**
* Helps transfer if a resize is in progress.
* <p>
* 如果數(shù)組正在擴(kuò)容，協(xié)助之，多個(gè)工作線程一起擴(kuò)容
* 從舊的table的元素復(fù)制到新的table中
*
*/
final Node<K, V>[] helpTransfer(Node<K, V>[] tab, Node<K, V> f) {
Node<K, V>[] nextTab;
int sc;
//如果f是ForwardingNode，說明f正在擴(kuò)容，hash值已經(jīng)被標(biāo)為MOVED。
//ForwardingNode.nextTable就是新table不為空
if (tab != null && (f instanceof ForwardingNode) &&
(nextTab = ((ForwardingNode<K, V>) f).nextTable) != null) {
//根據(jù) length 得到一個(gè)前16位的標(biāo)識符，數(shù)組容量大小。
int rs = resizeStamp(tab.length);
//多重條件判斷未擴(kuò)容完成，還在進(jìn)行中,新老數(shù)組都沒有變，且sizeCtl<0
while (nextTab == nextTable && table == tab &&
(sc = sizeCtl) < 0) {
// 1. sizeCtl 無符號右移16位獲得高16位如果不等 rs 標(biāo)識符變了
// 2. (sc == rs + 1),表示擴(kuò)容結(jié)束
// 3. （sc == rs + MAX_RESIZERS）達(dá)到了最大幫助線程個(gè)數(shù) 65535個(gè)
// 4. transferIndex<= 0 也表示擴(kuò)容已經(jīng)結(jié)束
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || transferIndex <= 0)
break;
//增加一個(gè)線程幫助擴(kuò)容
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}
TreeNode結(jié)構(gòu)
/**
* Nodes for use in TreeBins
*/
static final class 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(int hash, K key, V val, Node<K, V> next,
TreeNode<K, V> parent) {
super(hash, key, val, next);
this.parent = parent;
}
}
TreeNode繼承了Node，又多了prev等，這里很少人注意，其實(shí)它在維護(hù)紅黑樹的同時(shí)也維護(hù)了雙向列表。雖然紅黑樹查詢方便，但遷移真的好難，借助雙向列表做遷移會容易很多。
transfer 單向列表擴(kuò)容
/**
* Moves and/or copies the nodes in each bin to new table. See
* above for explanation.
* 多線程擴(kuò)容操作
*/
private final void transfer(Node<K, V>[] tab, Node<K, V>[] nextTab) {
int n = tab.length, stride;
//數(shù)組遷移分塊執(zhí)行，每核處理的bucket量小于16個(gè)，則強(qiáng)制賦值16，
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) {
stride = MIN_TRANSFER_STRIDE; // subdivide range
}
//如果是擴(kuò)容線程，此時(shí)新數(shù)組為null
if (nextTab == null) { // initiating
try {
//構(gòu)建新數(shù)組，其容量為原來容量的2倍
@SuppressWarnings("unchecked")
Node<K, V>[] nt = (Node<K, V>[]) new Node<?, ?>[n << 1];
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
//記錄線程開始遷移的bucket，從后往前遷移
transferIndex = n;
}
int nextn = nextTab.length;
//已經(jīng)遷移的桶位，會用fwd占位（這個(gè)節(jié)點(diǎn)的hash值為MOVED），這個(gè)在put方法中見到過
ForwardingNode<K, V> fwd = new ForwardingNode<K, V>(nextTab);
// 當(dāng)advance == true時(shí)，表明該節(jié)點(diǎn)已經(jīng)處理過了
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
for (int i = 0, bound = 0; ; ) {
Node<K, V> f;
int fh;
//計(jì)算每一個(gè)線程負(fù)責(zé)哪部分，遷移以后賦fwd節(jié)點(diǎn)
//i記錄當(dāng)前正在遷移桶位的索引值
//bound記錄下一次任務(wù)遷移的開始桶位
//--i>=bound 表示當(dāng)前線程分配的遷移任務(wù)還沒有完成
while (advance) {
int nextIndex, nextBound;
if (--i >= bound || finishing) {
advance = false;
//沒有元素需要遷移
} else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
} else if (U.compareAndSwapInt // 用CAS計(jì)算得到下一次任務(wù)遷移的開始桶位，值值給transferIndex
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
bound = nextBound;
i = nextIndex - 1;
advance = false;
}
}
//沒有更多的需要遷移的bucket
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
// 擴(kuò)容結(jié)束后，table指向新數(shù)組，重新計(jì)算擴(kuò)容閾值，賦值給sizeCtl
if (finishing) {
nextTable = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1);
return;
}
// 擴(kuò)容任務(wù)線程數(shù)減1
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
//判斷當(dāng)前所有擴(kuò)容任務(wù)是否執(zhí)行完成,相等表明完成
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) {
return;
}
finishing = advance = true;
i = n; // recheck before commit
}
} else if ((f = tabAt(tab, i)) == null) { //當(dāng)前節(jié)點(diǎn)為null，在該位置添加一個(gè)ForwardingNode
advance = casTabAt(tab, i, null, fwd);
} else if ((fh = f.hash) == MOVED) {//如果是ForwardingNode，說明已經(jīng)擴(kuò)容過
advance = true; // already processed
} else {
synchronized (f) {
if (tabAt(tab, i) == f) {
Node<K, V> ln, hn;
// fh >= 0 ,表示為鏈表節(jié)點(diǎn)
if (fh >= 0) {
// 構(gòu)造兩個(gè)鏈表，一個(gè)是原鏈表，另一個(gè)是原鏈表的反序排列
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);
}
}
// 先擴(kuò)容再插入相應(yīng)值
//新table的i位置添加元素
setTabAt(nextTab, i, ln);
//新table的i+1位置添加元素
setTabAt(nextTab, i + n, hn);
// 舊table i 位置處插上ForwardingNode，表示該節(jié)點(diǎn)已經(jīng)處理過
setTabAt(tab, i, fwd);
advance = true;
// 紅黑樹處理邏輯，實(shí)質(zhì)上是維護(hù)雙向鏈表
} 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;
}
}
// 擴(kuò)容后紅黑樹節(jié)點(diǎn)個(gè)數(shù)若<=6，將樹轉(zhuǎn)單向鏈表
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;
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
}
}
}
}
}
總結(jié)
ConcurrentHashMap設(shè)計(jì)之精妙驚為天人，不愧為大師之作。有3個(gè)點(diǎn)可能之前沒有注意
- 在
new ConcurrentHashMap(c)時(shí)，初始容量并不是傳入的值。而是一個(gè)大于等于該值的2的冪次方值 - 世人都知道鏈表大于6的時(shí)候會轉(zhuǎn)紅黑樹，卻很少有人提及在紅黑樹節(jié)點(diǎn)個(gè)數(shù)小于等于6時(shí)會轉(zhuǎn)成鏈表
- ConcurrentHashMap和HashMap的數(shù)據(jù)結(jié)構(gòu)嚴(yán)格的說應(yīng)該是數(shù)組+單向列表+（紅黑樹+雙向鏈表）
本人水平有限，文中難免會有謬誤，還請各位指出。
參考
jdk1.8 & jdk1.7#ConcurrentHashMap源碼
黑馬程序員公開課
