ConcurrentHashMap源码解析,多线程扩容 (3)

可以看到,这种 put 将加锁的粒度又变得更小,仅仅锁到了那个,要插入的数组的位置的那个链表(或者树根节点),显然根本不会影响到其他的不在数组的这个位置的插入操作

要知道确定的那个桶里的元素本来由于 hashmap 的升级,都是很均匀的,而且 synchronized 本身的实现又有优化,所以比 7 的做法好很多。

2.4 jdk 8 的 get() 方法

ConcurrentHashMap源码解析,多线程扩容

同样计算出索引位置;

如果在这个索引位置的根节点获取到了直接返回;

否则如果说 eh < 0,代表是一个红黑树的状态,那么就会调用 find 方法去查找;

否则就遍历链表,找到值并返回。

可以看到get 方法因为节点的对象本身就都是 volatile 的,具有可见性,因此 get 方法直接找到返回,否则返回 null,没有任何加锁操作

2.5 jdk 8 的 addCount 方法

添加元素的相关操作之后,最后会调用 addCount 方法,也就是判断是否需要扩容,在这里面控制不同的策略。

ConcurrentHashMap源码解析,多线程扩容

整体思路,在圈出来的判断 check>= 0 之前的操作:是对 hashMap 的 size 进行更新,为了防止多个线程竞争更改 baseCount 的值,会将多个线程分散到一个叫 CounterCell 的数组里面,对 cell 中的value值进行更改,最后再同步给 baseCount。

然后开始判断 check :

如果新容量大于当前的阈值(大小*0.75),才扩容;

如果 sc < 0 ,说明当前是有别的线程进行扩容操作的,因此要先判断一些极端情况,然后用 CAS 操作去修改 sizCtl ,增加一个协助扩容的线程,调用 transfer 方法,这个是扩容的核心方法

否则 sc 肯定是 >=0 了,代表数组还没创建,同样用 CAS 操作创建,再去扩容。

2.6 jdk8 的扩容方法 transfer()

transfer 方法主要就是完成将扩容任务分配给多个线程去处理,根据了CPU核心数和集合 length 计算每个核一轮处理桶的个数。

然后每个线程处理的最小单位只能是一个数组的位置,这个时候扩容之后,和HashMap 一样,其实只有原位置或者 原位置+数组长度 的位置,因为仍然有可能多个线程操作之间发生哈希冲突,就用到 synchronized。

源码很长,这里的详细注释参考的是一个博客:

https://www.cnblogs.com/gunduzi/p/13651664.html

/** * Moves and/or copies the nodes in each bin to new table. See * above for explanation. * * transferIndex 表示转移时的下标,初始为扩容前的 length。 * * 我们假设长度是 32 */ private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) { int n = tab.length, stride; // 将 length / 8 然后除以 CPU核心数。如果得到的结果小于 16,那么就使用 16。 // 这里的目的是让每个 CPU 处理的桶一样多,避免出现转移任务不均匀的现象,如果桶较少的话,默认一个 CPU(一个线程)处理 16 个桶 if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) stride = MIN_TRANSFER_STRIDE; // subdivide range 细分范围 stridea:TODO // 新的 table 尚未初始化 if (nextTab == null) { // initiating try { // 扩容 2 倍 Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1]; // 更新 nextTab = nt; } catch (Throwable ex) { // try to cope with OOME // 扩容失败, sizeCtl 使用 int 最大值。 sizeCtl = Integer.MAX_VALUE; return;// 结束 } // 更新成员变量 nextTable = nextTab; // 更新转移下标,就是 老的 tab 的 length transferIndex = n; } // 新 tab 的 length int nextn = nextTab.length; // 创建一个 fwd 节点,用于占位。当别的线程发现这个槽位中是 fwd 类型的节点,则跳过这个节点。 ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab); // 首次推进为 true,如果等于 true,说明需要再次推进一个下标(i--),反之,如果是 false,那么就不能推进下标,需要将当前的下标处理完毕才能继续推进 boolean advance = true; // 完成状态,如果是 true,就结束此方法。 boolean finishing = false; // to ensure sweep before committing nextTab // 死循环,i 表示下标,bound 表示当前线程可以处理的当前桶区间最小下标,死循环的作用是保证拷贝全部完成。 for (int i = 0, bound = 0;;) { Node<K,V> f; int fh; // 如果当前线程可以向后推进;这个循环就是控制 i 递减。同时,每个线程都会进入这里取得自己需要转移的桶的区间 //这个循环只是用来控制每个线程每轮最多copy的桶的个数,如果只有一个线程在扩容,也是可以完成的,只是分成多轮 while (advance) { int nextIndex, nextBound; // 对 i 减一,判断是否大于等于 bound (正常情况下,如果大于 bound 不成立,说明该线程上次领取的任务已经完成了。那么,需要在下面继续领取任务) // 如果对 i 减一大于等于 bound(还需要继续做任务),或者完成了,修改推进状态为 false,不能推进了。任务成功后修改推进状态为 true。 // 通常,第一次进入循环,i-- 这个判断会无法通过,从而走下面的 nextIndex 赋值操作(获取最新的转移下标)。其余情况都是:如果可以推进, //将 i 减一,然后修改成不可推进。如果 i 对应的桶处理成功了,改成可以推进。 if (--i >= bound || finishing) advance = false;// 这里设置 false,是为了防止在没有成功处理一个桶的情况下却进行了推进 // 这里的目的是:1. 当一个线程进入时,会选取最新的转移下标。2. 当一个线程处理完自己的区间时,如果还有剩余区间的没有别的线程处理。再次获取区间。 else if ((nextIndex = transferIndex) <= 0) { // 如果小于等于0,说明没有区间了 ,i 改成 -1,推进状态变成 false,不再推进,表示,扩容结束了,当前线程可以退出了 // 这个 -1 会在下面的 if 块里判断,从而进入完成状态判断 i = -1; advance = false;// 这里设置 false,是为了防止在没有成功处理一个桶的情况下却进行了推进 }// CAS 修改 transferIndex,即 length - 区间值,留下剩余的区间值供后面的线程使用 else if (U.compareAndSwapInt (this, TRANSFERINDEX, nextIndex, nextBound = (nextIndex > stride ? nextIndex - stride : 0))) { bound = nextBound;// 这个值就是当前线程可以处理的最小当前区间最小下标 i = nextIndex - 1; // 初次对i 赋值,这个就是当前线程可以处理的当前区间的最大下标 advance = false; // 这里设置 false,是为了防止在没有成功处理一个桶的情况下却进行了推进,这样对导致漏掉某个桶。下面的 if (tabAt(tab, i) == f) 判断会出现这样的情况。 } } // 如果 i 小于0 (不在 tab 下标内,按照上面的判断,领取最后一段区间的线程扩容结束) // 如果 i >= tab.length(不知道为什么这么判断) // 如果 i + tab.length >= nextTable.length (不知道为什么这么判断) if (i < 0 || i >= n || i + n >= nextn) { int sc; if (finishing) { // 如果完成了扩容 nextTable = null;// 删除成员变量 table = nextTab;// 更新 table sizeCtl = (n << 1) - (n >>> 1); // 更新阈值 return;// 结束方法。 }// 如果没完成 //说明1 if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {// 尝试将 sc -1. 表示这个线程结束帮助扩容了,将 sc 的低 16 位减一。 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)// 如果 sc - 2 不等于标识符左移 16 位。如果他们相等了,说明没有线程在帮助他们扩容了。也就是说,扩容结束了。 return;// 不相等,说明没结束,当前线程结束方法。 finishing = advance = true;// 如果相等,扩容结束了,更新 finising 变量 i = n; // 再次循环检查一下整张表 } } else if ((f = tabAt(tab, i)) == null) // 获取老 tab i 下标位置的变量,如果是 null,就使用 fwd 占位。 advance = casTabAt(tab, i, null, fwd);// 如果成功写入 fwd 占位,再次推进一个下标 else if ((fh = f.hash) == MOVED)// 如果不是 null 且 hash 值是 MOVED。 advance = true; // already processed // 说明别的线程已经处理过了,再次推进一个下标 else {// 到这里,说明这个位置有实际值了,且不是占位符。对这个节点上锁。为什么上锁,防止 putVal 的时候向链表插入数据 synchronized (f) { // 判断 i 下标处的桶节点是否和 f 相同 if (tabAt(tab, i) == f) { Node<K,V> ln, hn;// low, height 高位桶,低位桶 // 如果 f 的 hash 值大于 0 。TreeBin 的 hash 是 -2 if (fh >= 0) { // 对老长度进行与运算(第一个操作数的的第n位于第二个操作数的第n位如果都是1,那么结果的第n为也为1,否则为0) // 由于 Map 的长度都是 2 的次方(000001000 这类的数字),那么取于 length 只有 2 种结果,一种是 0,一种是1 // 如果是结果是0 ,Doug Lea 将其放在低位,反之放在高位,目的是将链表重新 hash,放到对应的位置上,让新的取于算法能够击中他。 int runBit = fh & n; Node<K,V> lastRun = f; // 尾节点,且和头节点的 hash 值取于不相等 // 遍历这个桶 //说明2 for (Node<K,V> p = f.next; p != null; p = p.next) { // 取于桶中每个节点的 hash 值 int b = p.hash & n; // 如果节点的 hash 值和首节点的 hash 值取于结果不同 if (b != runBit) { runBit = b; // 更新 runBit,用于下面判断 lastRun 该赋值给 ln 还是 hn。 lastRun = p; // 这个 lastRun 保证后面的节点与自己的取于值相同,避免后面没有必要的循环 } } if (runBit == 0) {// 如果最后更新的 runBit 是 0 ,设置低位节点 ln = lastRun; hn = null; } else { hn = lastRun; // 如果最后更新的 runBit 是 1, 设置高位节点 ln = null; }// 再次循环,生成两个链表,lastRun 作为停止条件,这样就是避免无谓的循环(lastRun 后面都是相同的取于结果) for (Node<K,V> p = f; p != lastRun; p = p.next) { int ph = p.hash; K pk = p.key; V pv = p.val; // 如果与运算结果是 0,那么就还在低位 if ((ph & n) == 0) // 如果是0 ,那么创建低位节点 ln = new Node<K,V>(ph, pk, pv, ln); else // 1 则创建高位 hn = new Node<K,V>(ph, pk, pv, hn); } // 其实这里类似 hashMap // 设置低位链表放在新链表的 i setTabAt(nextTab, i, ln); // 设置高位链表,在原有长度上加 n setTabAt(nextTab, i + n, hn); // 将旧的链表设置成占位符 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); // 和链表相同的判断,与运算 == 0 的放在低位 if ((h & n) == 0) { if ((p.prev = loTail) == null) lo = p; else loTail.next = p; loTail = p; ++lc; } // 不是 0 的放在高位 else { if ((p.prev = hiTail) == null) hi = p; else hiTail.next = p; hiTail = p; ++hc; } } // 如果树的节点数小于等于 6,那么转成链表,反之,创建一个新的树 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; } } } } } }

流程如下:

根据操作系统的 CPU 核数和集合 length 计算每个核一轮处理桶的个数,最小是16

内容版权声明:除非注明,否则皆为本站原创文章。

转载注明出处:https://www.heiqu.com/wpswjs.html