死磕 java集合之ConcurrentHashMap源碼分析(三)
死磕 java集合之ConcurrentHashMap源碼分析(一)
死磕 java集合之ConcurrentHashMap源碼分析(二)
刪除元素
刪除元素跟添加元素一樣,都是先找到元素所在的桶,然後采用分段鎖的思想鎖住整個桶,再進行操作。
public V remove(Object key) { // 調用替換節點方法 return replaceNode(key, null, null); } final V replaceNode(Object key, V value, Object cv) { // 計算hash int hash = spread(key.hashCode()); // 自旋 for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; if (tab == null || (n = tab.length) == 0 || (f = tabAt(tab, i = (n - 1) & hash)) == null) // 如果目標key所在的桶不存在,跳出循環返回null break; else if ((fh = f.hash) == MOVED) // 如果正在擴容中,協助擴容 tab = helpTransfer(tab, f); else { V oldVal = null; // 標記是否處理過 boolean validated = false; synchronized (f) { // 再次驗證當前桶第一個元素是否被修改過 if (tabAt(tab, i) == f) { if (fh >= 0) { // fh>=0表示是鏈表節點 validated = true; // 遍歷鏈表尋找目標節點 for (Node<K,V> e = f, pred = null;;) { K ek; if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { // 找到了目標節點 V ev = e.val; // 檢查目標節點舊value是否等於cv if (cv == null || cv == ev || (ev != null && cv.equals(ev))) { oldVal = ev; if (value != null) // 如果value不為空則替換舊值 e.val = value; else if (pred != null) // 如果前置節點不為空 // 刪除當前節點 pred.next = e.next; else // 如果前置節點為空 // 說明是桶中第一個元素,刪除之 setTabAt(tab, i, e.next); } break; } pred = e; // 遍歷到鏈表尾部還沒找到元素,跳出循環 if ((e = e.next) == null) break; } } else if (f instanceof TreeBin) { // 如果是樹節點 validated = true; TreeBin<K,V> t = (TreeBin<K,V>)f; TreeNode<K,V> r, p; // 遍歷樹找到了目標節點 if ((r = t.root) != null && (p = r.findTreeNode(hash, key, null)) != null) { V pv = p.val; // 檢查目標節點舊value是否等於cv if (cv == null || cv == pv || (pv != null && cv.equals(pv))) { oldVal = pv; if (value != null) // 如果value不為空則替換舊值 p.val = value; else if (t.removeTreeNode(p)) // 如果value為空則刪除元素 // 如果刪除後樹的元素個數較少則退化成鏈表 // t.removeTreeNode(p)這個方法返回true表示刪除節點後樹的元素個數較少 setTabAt(tab, i, untreeify(t.first)); } } } } } // 如果處理過,不管有沒有找到元素都返回 if (validated) { // 如果找到了元素,返回其舊值 if (oldVal != null) { // 如果要替換的值為空,元素個數減1 if (value == null) addCount(-1L, -1); return oldVal; } break; } } } // 沒找到元素返回空 return null; }
(1)計算hash;
(2)如果所在的桶不存在,表示沒有找到目標元素,返回;
(3)如果正在擴容,則協助擴容完成後再進行刪除操作;
(4)如果是以鏈表形式存儲的,則遍歷整個鏈表查找元素,找到之後再刪除;
(5)如果是以樹形式存儲的,則遍歷樹查找元素,找到之後再刪除;
(6)如果是以樹形式存儲的,刪除元素之後樹較小,則退化成鏈表;
(7)如果確實刪除了元素,則整個map元素個數減1,並返回舊值;
(8)如果沒有刪除元素,則返回null;
獲取元素
獲取元素,根據目標key所在桶的第一個元素的不同采用不同的方式獲取元素,關鍵點在於find()方法的重寫。
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()); // 如果元素所在的桶存在且裏面有元素 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) // hash小於0,說明是樹或者正在擴容 // 使用find尋找元素,find的尋找方式依據Node的不同子類有不同的實現方式 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; }
(1)hash到元素所在的桶;
(2)如果桶中第一個元素就是該找的元素,直接返回;
(3)如果是樹或者正在遷移元素,則調用各自Node子類的find()方法尋找元素;
(4)如果是鏈表,遍歷整個鏈表尋找元素;
(5)獲取元素沒有加鎖;
獲取元素個數
元素個數的存儲也是采用分段的思想,獲取元素個數時需要把所有段加起來。
public int size() { // 調用sumCount()計算元素個數 long n = sumCount(); return ((n < 0L) ? 0 : (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE : (int)n); } final long sumCount() { // 計算CounterCell所有段及baseCount的數量之和 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; } } return sum; }
(1)元素的個數依據不同的線程存在在不同的段裏;(見addCounter()分析)
(2)計算CounterCell所有段及baseCount的數量之和;
(3)獲取元素個數沒有加鎖;
總結
(1)ConcurrentHashMap是HashMap的線程安全版本;
(2)ConcurrentHashMap采用(數組 + 鏈表 + 紅黑樹)的結構存儲元素;
(3)ConcurrentHashMap相比於同樣線程安全的HashTable,效率要高很多;
(4)ConcurrentHashMap采用的鎖有 synchronized,CAS,自旋鎖,分段鎖,volatile等;
(5)ConcurrentHashMap中沒有threshold和loadFactor這兩個字段,而是采用sizeCtl來控制;
(6)sizeCtl = -1,表示正在進行初始化;
(7)sizeCtl = 0,默認值,表示後續在真正初始化的時候使用默認容量;
(8)sizeCtl > 0,在初始化之前存儲的是傳入的容量,在初始化或擴容後存儲的是下一次的擴容門檻;
(9)sizeCtl = (resizeStamp << 16) + (1 + nThreads),表示正在進行擴容,高位存儲擴容郵戳,低位存儲擴容線程數加1;
(10)更新操作時如果正在進行擴容,當前線程協助擴容;
(11)更新操作會采用synchronized鎖住當前桶的第一個元素,這是分段鎖的思想;
(12)整個擴容過程都是通過CAS控制sizeCtl這個字段來進行的,這很關鍵;
(13)遷移完元素的桶會放置一個ForwardingNode節點,以標識該桶遷移完畢;
(14)元素個數的存儲也是采用的分段思想,類似於LongAdder的實現;
(15)元素個數的更新會把不同的線程hash到不同的段上,減少資源爭用;
(16)元素個數的更新如果還是出現多個線程同時更新一個段,則會擴容段(CounterCell);
(17)獲取元素個數是把所有的段(包括baseCount和CounterCell)相加起來得到的;
(18)查詢操作是不會加鎖的,所以ConcurrentHashMap不是強一致性的;
(19)ConcurrentHashMap中不能存儲key或value為null的元素;
彩蛋——值得學習的技術
ConcurrentHashMap中有哪些值得學習的技術呢?
我認為有以下幾點:
(1)CAS + 自旋,樂觀鎖的思想,減少線程上下文切換的時間;
(2)分段鎖的思想,減少同一把鎖爭用帶來的低效問題;
(3)CounterCell,分段存儲元素個數,減少多線程同時更新一個字段帶來的低效;
(4)@sun.misc.Contended(CounterCell上的註解),避免偽共享;(p.s.偽共享我們後面也會講的^^)
(5)多線程協同進行擴容;
(6)你又學到了哪些呢?
彩蛋——不能解決的問題
ConcurrentHashMap不能解決什麽問題呢?
請看下面的例子:
private static final Map<Integer, Integer> map = new ConcurrentHashMap<>();
public void unsafeUpdate(Integer key, Integer value) {
Integer oldValue = map.get(key);
if (oldValue == null) {
map.put(key, value);
}
}
這裏如果有多個線程同時調用unsafeUpdate()這個方法,ConcurrentHashMap還能保證線程安全嗎?
答案是不能。因為get()之後if之前可能有其它線程已經put()了這個元素,這時候再put()就把那個線程put()的元素覆蓋了。
那怎麽修改呢?
答案也很簡單,使用putIfAbsent()方法,它會保證元素不存在時才插入元素,如下:
public void safeUpdate(Integer key, Integer value) {
map.putIfAbsent(key, value);
}
那麽,如果上面oldValue不是跟null比較,而是跟一個特定的值比如1進行比較怎麽辦?也就是下面這樣:
public void unsafeUpdate(Integer key, Integer value) {
Integer oldValue = map.get(key);
if (oldValue == 1) {
map.put(key, value);
}
}
這樣的話就沒辦法使用putIfAbsent()方法了。
其實,ConcurrentHashMap還提供了另一個方法叫replace(K key, V oldValue, V newValue)可以解決這個問題。
replace(K key, V oldValue, V newValue)這個方法可不能亂用,如果傳入的newValue是null,則會刪除元素。
public void safeUpdate(Integer key, Integer value) {
map.replace(key, 1, value);
}
那麽,如果if之後不是簡單的put()操作,而是還有其它業務操作,之後才是put(),比如下面這樣,這該怎麽辦呢?
public void unsafeUpdate(Integer key, Integer value) {
Integer oldValue = map.get(key);
if (oldValue == 1) {
System.out.println(System.currentTimeMillis());
/**
* 其它業務操作
*/
System.out.println(System.currentTimeMillis());
map.put(key, value);
}
}
這時候就沒辦法使用ConcurrentHashMap提供的方法了,只能業務自己來保證線程安全了,比如下面這樣:
public void safeUpdate(Integer key, Integer value) {
synchronized (map) {
Integer oldValue = map.get(key);
if (oldValue == null) {
System.out.println(System.currentTimeMillis());
/**
* 其它業務操作
*/
System.out.println(System.currentTimeMillis());
map.put(key, value);
}
}
}
這樣雖然不太友好,但是最起碼能保證業務邏輯是正確的。
當然,這裏使用ConcurrentHashMap的意義也就不大了,可以換成普通的HashMap了。
上面只是舉一個簡單的例子,我們不能聽說ConcurrentHashMap是線程安全的,就認為它無論什麽情況下都是線程安全的,還是那句話盡信書不如無書。
這也正是我們讀源碼的目的之一,了解其本質,才能在我們的實際工作中少挖坑,不論是挖給別人還是挖給自己^^。
死磕 java集合之ConcurrentHashMap源碼分析(三)