ConcurrentHashMap 是如何支持并发扩容骚操作，你了解吗？

ConcurrentHashMap 是 Java 中一个高效的并发容器，为了支持高并发环境下的扩容操作，它采用了一些复杂而巧妙的设计。以下是对 ConcurrentHashMap 并发扩容的详细解析。 ### 先说结论 ConcurrentHashMap 的扩容主要包含以下步骤： 1. **计算每个线程要处理的桶区间**：根据 CPU 核心数和 Map 数组长度计算，每个线程平均处理一定数量的桶，默认每个线程处理 16 个桶。 2. **初始化新数组**：将数组容量扩展为原来的两倍。 3. **多线程并发扩容**：通过死循环分配和处理桶的区间，确保每个线程并发地处理指定数量的桶，直至所有桶的迁移完成。 ### 源码分析 以下是 ConcurrentHashMap 中的扩容方法 `transfer` 的源码及注释： ```java 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; // 初始化新数组 if (nextTab == null) { try { Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1]; nextTab = nt; } catch (Throwable ex) { sizeCtl = Integer.MAX_VALUE; return; } nextTable = nextTab; transferIndex = n; } int nextn = nextTab.length; ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab); boolean advance = true; boolean finishing = false; // 死循环，分配和处理桶的区间 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; } } else if ((f = tabAt(tab, i)) == null) advance = casTabAt(tab, i, null, fwd); else if ((fh = f.hash) == MOVED) advance = true; else { // 处理桶中的数据 synchronized (f) { if (tabAt(tab, i) == f) { Node<K,V> ln, hn; if (fh >= 0) { // 处理链表节点 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; 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); 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; } } } } } } ``` ### 具体步骤解析 #### 1. 计算每个线程要处理的桶区间 根据 CPU 核心数和数组长度，计算每个线程要处理的桶数量。默认每个线程处理 16 个桶。 ```java int n = tab.length, stride; if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) stride = MIN_TRANSFER_STRIDE; ``` #### 2. 初始化新数组 如果 `nextTab` 未初始化，则扩容为原来的两倍。更新成员变量 `nextTable` 和 `transferIndex`。 ```java if (nextTab == null) { try { Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1]; nextTab = nt; } catch (Throwable ex) { sizeCtl = Integer.MAX_VALUE; return; } nextTable = nextTab; transferIndex = n; } ``` #### 3. 多线程并发扩容 通过死循环分配和处理桶的区间，确保每个线程并发地处理指定数量的桶。采用 `ForwardingNode` 占位符标识已经迁移的桶，避免重复处理。使用 `synchronized` 关键字锁定桶，防止并发修改。如果桶是链表，将链表节点重新分配到新数组的两个位置（高位和低位）。如果桶是红黑树，重新分配红黑树节点，必要时转换为链表。 ```java boolean advance = true; boolean finishing = false; 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; } } else if ((f = tabAt(tab, i)) == null) advance = casTabAt(tab, i, null, fwd); else if ((fh = f.hash) == MOVED) advance = true; else { synchronized (f) { if (tabAt(tab, i) == f) { Node<K,V> ln, hn; if (fh >= 0) { 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; 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); 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; } } } } } } ``` 通过以上代码和解析，可以看到 ConcurrentHashMap 的扩容操作采用了多线程并发的方式，通过合理分配任务，确保每个线程处理的桶数量均衡，并通过 `ForwardingNode` 和 `synchronized` 锁实现安全的并发扩容。这种设计确保了在高并发环境下，ConcurrentHashMap 能够高效地进行扩容，同时继续对外提供稳定的服务。

### ConcurrentHashMap 是如何支持并发扩容 ConcurrentHashMap 是 Java 中一个高效的并发容器，为了支持高并发环境下的扩容操作，它采用了一些复杂而巧妙的设计。以下是对 ConcurrentHashMap 并发扩容的详细解析。 ### 触发扩容的条件 1. **添加新元素后，元素个数达到扩容阈值**： 当添加新元素后，元素个数达到了容量的扩容阈值（通常为容量的 75%），会触发扩容操作。 2. **调用 `putAll` 方法时**： 当使用 `putAll` 方法将其他 Map 的所有元素添加到 ConcurrentHashMap 时，如果当前容量不足以容纳所有元素，会触发扩容。 3. **某个槽内的链表长度达到 8，但数组长度小于 64**： 当某个槽（bucket）内的链表长度达到 8 时，如果数组长度小于 64，则会触发扩容操作。 ### 统计元素个数 对于扩容条件 1，ConcurrentHashMap 需要统计元素的个数。为了解决高并发环境下的统计问题，它采用了类似 LongAdder 的分散热点数据的解决方案。 **具体实现**： - **CounterCell**：ConcurrentHashMap 内部定义了一个 CounterCell 类，用于计数。每个 CounterCell 被 `@sun.misc.Contended` 注解修饰，以防止伪共享问题。 - **BASECOUNT**：除了 CounterCell 之外，还有一个 BASECOUNT 变量用于在并发量较小时直接进行计数。 **统计方式**： - 每当线程添加一个元素到容器中，线程会映射到一个 CounterCell 的元素上，并采用 CAS 算法进行加 1 操作。 - 如果当前 CounterCell 已有线程在操作，或并发量较小时，计数会直接累加到 BASECOUNT 上。 **获取元素个数**： - 需要获取容器元素个数时，将所有 CounterCell 的值加上 BASECOUNT 的值即可。 ### 扩容机制 触发扩容后，ConcurrentHashMap 会进行如下操作： 1. **分配新数组**： - 新的数组大小通常是旧数组的两倍。 2. **任务分配**： - 每个线程负责不小于 16 个槽（bucket）的元素扩容，线程会从右向左划分 16 个槽进行迁移。 3. **迁移操作**： - 线程锁定当前槽中的链表头元素进行迁移。 - 如果在迁移过程中有 `get` 请求，会继续访问旧的链表。 - 对于插入、修改、删除、合并、compute 等操作，遇到 `ForwardingNode` 时，当前线程会加入扩容任务。 4. **迁移过程中的协作**： - 当一个线程发现一个槽需要迁移时，它会锁住这个槽，并开始迁移该槽中的元素。 - 在迁移过程中，如果有其他线程进行更新操作，这些线程会帮助一起扩容。 - 扩容结束后，更新操作继续进行。 ### 扩容过程中如何提供服务 在扩容过程中，ConcurrentHashMap 通过以下机制确保对外提供服务： 1. **读操作**： - 在迁移过程中，读操作可以继续访问旧数组中的数据。 2. **写操作**： - 写操作在遇到 `ForwardingNode` 时会参与扩容，帮助迁移数据。 ### 总结 ConcurrentHashMap 的扩容机制可以总结为以下三点： 1. **触发扩容的条件**： - 添加新元素后达到扩容阈值。 - 调用 `putAll` 方法时容量不足。 - 槽内链表长度达到 8 且数组长度小于 64。 2. **元素个数的统计**： - 使用类似 LongAdder 的分散计数策略，通过 CounterCell 和 BASECOUNT 进行计数。 3. **扩容过程**： - 分配新数组，每个线程负责一部分槽的迁移。 - 迁移过程中，读操作不受影响，写操作会参与扩容。 这种设计确保了在高并发环境下，ConcurrentHashMap 能够高效地进行扩容，同时继续对外提供稳定的服务。