小編給大家分享一下ConcurrentHashMap怎么用,希望大家閱讀完這篇文章之后都有所收獲,下面讓我們一起去探討吧!
創新互聯網絡公司擁有10年的成都網站開發建設經驗,近千家客戶的共同信賴。提供成都網站設計、網站建設、外貿網站建設、網站開發、網站定制、買友情鏈接、建網站、網站搭建、響應式網站建設、網頁設計師打造企業風格,提供周到的售前咨詢和貼心的售后服務
首先看一下putVal方法,
if (tab == null || (n = tab.length) == 0) tab = initTable();
如果還沒有table的話,就要先初始化table
private final Node<K,V>[] initTable() { Node<K,V>[] tab; int sc; while ((tab = table) == null || tab.length == 0) { 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; // size 控制在 n的0.75 sc = n - (n >>> 2); } } finally { sizeCtl = sc; } break; } } return tab; }
這一段代碼相對簡單,這里的sizeCtl是整個過程中的一個非常重要的屬性,在擴容,初始化等過程過程中會多次遇到。在這里也是充當了一個排他鎖的作用,當它為-1的時候,其它線程等待。
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);
那么如果要插入的hash值為moved狀態即-1的時候,那么就要執行helpTransfer方法了,對,就是先讓幫助擴容。這里就要扯出來比較多的東西了,我們一點點來進行分析。
首先看看什么時候一個node的hash值變成了-1,一路看下去,只有
static final class ForwardingNode<K,V> extends Node<K,V>
這個類使用到了,它里面有一個屬性
final Node<K,V>[] nextTable;
從這里也大概就能看出來,這個時候ConcurrentHashmap處于擴容狀態,通過ForwardingNode就可以找到擴容后的table。
接著來看helpTransfer
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) { Node<K,V>[] nextTab; int sc; // 這里還要再次檢查 當前節點是不是 ForwardingNode 因為如果不是的話,沒有辦法找到nextTable,也就沒有辦法幫助擴容了 if (tab != null && (f instanceof ForwardingNode) && (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) { int rs = resizeStamp(tab.length); while (nextTab == nextTable && table == tab && (sc = sizeCtl) < 0) { if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || transferIndex <= 0) break; // 進來一個線程,則對sizeCtl+1,用以標記參與擴容的線程數 if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) { // 進行擴容操作 transfer(tab, nextTab); break; } } return nextTab; } return table; }
下來看看transfer擴容操作是如何執行的,這里感覺是ConcurrentHashmap的一個精華點,嘆為觀止。
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) { int n = tab.length, stride; // 首先進行分段,既每個線程每次處理的node數量,最小16 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]; nextTab = nt; } catch (Throwable ex) { // try to cope with OOME sizeCtl = Integer.MAX_VALUE; return; } nextTable = nextTab; transferIndex = n; } int nextn = nextTab.length; // 在這里創建了ForwardingNode 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; // 擴容時用的指針已經小于0,則結束 else if ((nextIndex = transferIndex) <= 0) { i = -1; advance = false; } // 擴容的指針,從大向小移動,從大向小移動,每次減小stride else if (U.compareAndSwapInt (this, TRANSFERINDEX, nextIndex, nextBound = (nextIndex > stride ? nextIndex - stride : 0))) { bound = nextBound; i = nextIndex - 1; advance = false; } } // i 小于 0 ,已經結束了 if (i < 0 || i >= n || i + n >= nextn) { int sc; // 如果已經結束了,那么把table設置為nextTable if (finishing) { nextTable = null; table = nextTab; sizeCtl = (n << 1) - (n >>> 1); return; } // 說明當前的線程已經工作結束,sizeCtl - 1 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 } } //如果節點為空,設置該節點為fwd else if ((f = tabAt(tab, i)) == null) advance = casTabAt(tab, i, null, fwd); else if ((fh = f.hash) == MOVED) advance = true; // already processed else { synchronized (f) { if (tabAt(tab, i) == f) { Node<K,V> ln, hn; if (fh >= 0) { // 這里為什么是 fh & n 做 & 運算 因為 15 的二進制是 1111 16是10000 31是 11111 // 所以,擴容前和擴容后只有第一位 & 運算后會變,其它位都不變,所以與 table.length & 就可以了 int runBit = fh & n; Node<K,V> lastRun = f; // 先遍歷一遍,確定 ni -> n rehash相等的一段,這樣下一次重新分配槽的時候這一段就不再遍歷 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); } setTabAt(nextTab, i, ln); setTabAt(nextTab, i + n, hn); // 把已完成的節點標記為fwd setTabAt(tab, i, fwd); advance = true; } 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; } } } } } }
再來看一下對于元素總數的統計實現。
private final void addCount(long x, int check)
首先我們遇到了CounterCell這個類,結構很簡單,只有一個long value,它是存儲數量的最小單元。
先看第一次的判斷條件,如果conterCells已經不為空,說明之前已經出現了并發增加baseCount,否則counterCell不會被初始化。
if ((as = counterCells) != null || !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x))
或者在改變baseCount的時候出現了沖突,執行下面代碼。
CounterCell a; long v; int m; boolean uncontended = true; // 如果 counterCell未初始化,或者長度為0 亦或者沒有這個對應的槽 再或者更新對應槽的時候出現沖突 // 這個時候說明要么 counterCell未初始化,要么說明又出現了對于同一個槽沖突,所以需要 fullAddCount來解決沖突 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();
再來看看fullAddCount做了什么
private final void fullAddCount(long x, boolean wasUncontended) { int h; // 如果ThreadLocalRandom還沒有被初始化過,說明還沒有發生過碰撞 if ((h = ThreadLocalRandom.getProbe()) == 0) { ThreadLocalRandom.localInit(); // force initialization h = ThreadLocalRandom.getProbe(); wasUncontended = true; } boolean collide = false; // True if last slot nonempty for (;;) { CounterCell[] as; CounterCell a; int n; long v; // 如果數組已經被初始化 if ((as = counterCells) != null && (n = as.length) > 0) { // 隨機選取的槽還未被初始化 if ((a = as[(n - 1) & h]) == null) { // 獲取鎖 if (cellsBusy == 0) { // Try to attach new Cell CounterCell r = new CounterCell(x); // Optimistic create // U.compareAndSwapInt(this, CELLSBUSY, 0, 1)通過cas操作來獲取鎖 if (cellsBusy == 0 && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { boolean created = false; try { // Recheck under lock CounterCell[] rs; int m, j; if ((rs = counterCells) != null && (m = rs.length) > 0 && rs[j = (m - 1) & h] == null) { rs[j] = r; created = true; } } finally { cellsBusy = 0; } if (created) break; continue; // Slot is now non-empty } } collide = false; } // 有競爭的 else if (!wasUncontended) // CAS already known to fail wasUncontended = true; // Continue after rehash else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x)) break; // 如果槽的數量已經超過了cpu個數,就不會碰撞了 else if (counterCells != as || n >= NCPU) collide = false; // At max size or stale else if (!collide) collide = true; // 獲取鎖,并對CounterCell進行擴容操作 else if (cellsBusy == 0 && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { try { if (counterCells == as) {// Expand table unless stale CounterCell[] rs = new CounterCell[n << 1]; for (int i = 0; i < n; ++i) rs[i] = as[i]; counterCells = rs; } } finally { cellsBusy = 0; } collide = false; continue; // Retry with expanded table } h = ThreadLocalRandom.advanceProbe(h); } // counter cell 沒有初始化的情況 else if (cellsBusy == 0 && counterCells == as && U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { boolean init = false; try { // Initialize table if (counterCells == as) { // 進行初始化 CounterCell[] rs = new CounterCell[2]; rs[h & 1] = new CounterCell(x); counterCells = rs; init = true; } } finally { cellsBusy = 0; } if (init) break; } else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x)) break; // Fall back on using base } }
我們前面講了擴容的機制,那么擴容的發起者肯定就是在addCount中了
while (s >= (long)(sc = sizeCtl) && (tab = table) != null && (n = tab.length) < MAXIMUM_CAPACITY)
這里有判斷 s 為 sumCount 即 baseCount加上各個節點的和為總數。如果s大于sizeCtl或者table不為空而且沒有到達最大值,則進行擴容操作。
看完了這篇文章,相信你對“ConcurrentHashMap怎么用”有了一定的了解,如果想了解更多相關知識,歡迎關注創新互聯行業資訊頻道,感謝各位的閱讀!
文章標題:ConcurrentHashMap怎么用
當前網址:http://m.newbst.com/article8/ijphip.html
成都網站建設公司_創新互聯,為您提供網頁設計公司、域名注冊、全網營銷推廣、企業網站制作、網站設計公司、網站建設
聲明:本網站發布的內容(圖片、視頻和文字)以用戶投稿、用戶轉載內容為主,如果涉及侵權請盡快告知,我們將會在第一時間刪除。文章觀點不代表本網站立場,如需處理請聯系客服。電話:028-86922220;郵箱:631063699@qq.com。內容未經允許不得轉載,或轉載時需注明來源: 創新互聯