ConcurrentHashMap 是一個(gè)線程安全的map。其他特點(diǎn)和hashMap一致。
相比如1.7的分段式鎖的設(shè)計(jì)，1.8采用了cas和synchronized的技術(shù)。
下面來分析他各個(gè)階段的線程安全的設(shè)計(jì)。
1、讀操作
讀操作，因?yàn)樵贛ap中的設(shè)計(jì)，value值都設(shè)計(jì)成volatile，根據(jù)happen-before原則，被volatile修飾的對(duì)象，寫操作一定先發(fā)生于讀操作，所以讀數(shù)據(jù)時(shí)，讀取到的一定是最新的數(shù)據(jù)，所以這個(gè)操作無需加鎖。
2、寫操作
直接看代碼
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;
//第一次put 初始化tab
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//如果當(dāng)前槽是空的,不需要加鎖。用cas替換當(dāng)前槽的頭節(jié)點(diǎn)
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break;
}
//如果當(dāng)前槽已經(jīng)被擴(kuò)容（MOVED 是擴(kuò)容時(shí)已經(jīng)移除的標(biāo)志）,一起進(jìn)入擴(kuò)容，如何擴(kuò)容后續(xù)會(huì)分析。
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
//正式put 操作，先將槽的頭節(jié)點(diǎn)鎖住，剩下的操作和HashMap一致
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;
}
}
}
}
if (binCount != 0) {
//當(dāng)前節(jié)點(diǎn)超過8，進(jìn)行數(shù)化
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
//將map 容量加1 并且判斷是否需要擴(kuò)容
addCount(1L, binCount);
return null;
}
總結(jié)一下以上代碼：
- 判斷tab是否初始化，沒有初始化則進(jìn)行初始化。
- 判斷要當(dāng)前槽是否是空的，空的槽無需加鎖（tab是volatile的,能夠保證讀取到的是最新頭結(jié)點(diǎn)，當(dāng)時(shí)無法保證節(jié)點(diǎn)后的引用是最新的），直接用CAS替換當(dāng)前槽頭節(jié)點(diǎn)。
- 判斷當(dāng)前槽是否已經(jīng)被擴(kuò)容（已經(jīng)擴(kuò)容的槽頭節(jié)點(diǎn)會(huì)被替換成ForwardingNode 這個(gè)node 的hash 就是MOVED），如果發(fā)現(xiàn)頭節(jié)點(diǎn)已經(jīng)被替換了，說明正在擴(kuò)容。則當(dāng)前線程加入擴(kuò)容(helpTransfer)。
- 進(jìn)行put操作，將當(dāng)前槽的頭節(jié)點(diǎn)鎖住。剩下的操作和HashMap一致。
- 將當(dāng)前容量加1，并且計(jì)算是否需要擴(kuò)容操作，如果是，則進(jìn)行擴(kuò)容。
3、擴(kuò)容操作
擴(kuò)容操作是通過拷貝底層tab,實(shí)現(xiàn)了CopyOnWrite特性。ConcurrHashMap在并發(fā)情況下擴(kuò)容不會(huì)出現(xiàn)讀取錯(cuò)誤的數(shù)據(jù)。擴(kuò)容的過程時(shí)候依次拷貝每一個(gè)槽，直達(dá)將全部槽都拷貝完成。
//底層數(shù)組擴(kuò)容
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
//nextTab 是擴(kuò)容后的tab，這里初始化一下
if (nextTab == null) {
try {
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;
//將要擴(kuò)容的下標(biāo) 設(shè)置為數(shù)組長度
transferIndex = n;
}
int nextn = nextTab.length;
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
//不知道要干什么
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 = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1);
return;
}
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
}
}
//當(dāng)前槽是空的，將當(dāng)前槽頭節(jié)點(diǎn)替換為fwd
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
//當(dāng)前槽被其他線程處理過了，直接跳過
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {//開始拷貝數(shù)組
synchronized (f) {
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
if (fh >= 0) {
//說明①.
//頭節(jié)點(diǎn)的hash runBit(runbit 是 hash 與上 n(數(shù)組長度)。
//舉例說明一下，如果n = 16 ,當(dāng)前槽下標(biāo)是4.
//根據(jù)槽下標(biāo)的計(jì)算公式（n-1）& hash.
//hash為4、20、36、52、68。那么hash & n 依次為0、16、0、16、0)
//在擴(kuò)容后的數(shù)組中的下標(biāo)依次為4、20、4、20、4
//說明hash & n 為0 的節(jié)點(diǎn)，在擴(kuò)容后的數(shù)組下標(biāo)不變，而不為0的下標(biāo)為index + n (擴(kuò)容前的下標(biāo)加上數(shù)組長度。)
//ln(lowNode 低節(jié)點(diǎn)意思) hn(hignNode 高節(jié)點(diǎn)意思)
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;
//與之前節(jié)點(diǎn)的高低位不同
//簡單說明一下 一個(gè)槽中 如果所有的高低位是一致的，那么只需要移動(dòng)頭節(jié)點(diǎn)就夠了，節(jié)點(diǎn)的引用并沒有發(fā)生變化。
//例如 8個(gè)節(jié)點(diǎn)runbit 分別是0 1 0 1 1 1 1 1
//那么只需要將第二位的節(jié)點(diǎn)指向第四位，第一位的指向第三位，就可以完成拷貝工作。所以這里只需要記錄一下 最后一次高低位變化的數(shù)據(jù)。
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
//看最后一位是高位還是低位
if (runBit == 0) {
ln = lastRun;
hn = null;
}
else {
hn = lastRun;
ln = null;
}
//遍歷節(jié)點(diǎn) 重新生成 鏈表
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);
}
//利用cas 替換高低位節(jié)點(diǎn)
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
//將原來數(shù)組的頭節(jié)點(diǎn) 替換為fwd
setTabAt(tab, i, fwd);
advance = true;
}//樹的操作基本相同，唯一有區(qū)別的是擴(kuò)容后需要反樹化
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;
}
}
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é)一下以上代碼：
- 如果要拷貝的數(shù)字沒有初始化，則進(jìn)行初始化。將拷貝的數(shù)字下標(biāo)置未數(shù)據(jù)長度，倒序拷貝。
- 當(dāng)前槽沒有數(shù)據(jù)。直接CAS將槽的頭節(jié)點(diǎn)替換為ForwardingNode
- 當(dāng)前槽已經(jīng)被別的線程修改過了 直接跳過。
- 拷貝數(shù)組，詳細(xì)講代碼上注釋。這里不再贅述。
- 拷貝完成 替換原來數(shù)組 頭節(jié)點(diǎn)為ForwardingNode(這里可以說明上面 頭節(jié)點(diǎn)為ForwardingNode ，說明當(dāng)前槽已經(jīng)拷貝完成，并且正在擴(kuò)容)
4、刪除
刪除操作沒有什么好說的，原理和put操作是一致的，鎖住頭部，控制并發(fā)。
