您好,登录后才能下订单哦!
小编给大家分享一下ConcurrentHashMap怎么用,希望大家阅读完这篇文章之后都有所收获,下面让我们一起去探讨吧!
首先看一下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怎么用”有了一定的了解,如果想了解更多相关知识,欢迎关注亿速云行业资讯频道,感谢各位的阅读!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。