当前位置： 首页 > news >正文

ConcurrentHashMap 原理

news 2025/7/4 9:57:24

并发死链（jdk7）

jdk7头插法导致出现1->35->1的环形链表。

在这里插入图片描述

内部属性和内部类（jdk8）

// 默认为 0
// 当初始化时，为 -1
// 当扩容时，为 -(1 + 扩容线程数)
// 当初始化或扩容完成后，为 下一次的扩容的阈值大小，扩容阈值为3/4，扩容大小为2倍private transient volatile int sizeCtl;// 整个 ConcurrentHashMap 就是一个 Node []，hash的每一个节点
static class Node<K,V> implements Map.Entry<K,V> {}//hash 表，实际为为多个Node节点的集合
transient volatile Node<K,V>[] table;// 扩容时的新 hash 表，
private transient volatile Node<K,V>[] nextTable;// 扩容时如果某个 bin 迁移完毕，用 ForwardingNode 作为旧 table bin 的头结点，单一个get访问旧的节点数据，发现发现是fNode则转向到迁移后的数组nextTable
static final class ForwardingNode<K,V> extends Node<K,V> {}// 用在 compute 以及 computeIfAbsent 时，用来占位，计算完成后替换为普通 Node
static final class ReservationNode<K,V> extends Node<K,V> {}// 作为 treebin 的头节点，存储 root 和 first，红黑树的头节点，当table长度>64且链表长于8，链表转换为红黑树，当红黑树节点<6时转换为链表
static final class TreeBin<K,V> extends Node<K,V> {}// 作为 treebin 的节点，存储 parent, left, right，红黑树的子节点
static final class TreeNode<K,V> extends Node<K,V> {}## 重要方法

在这里插入图片描述

构造器分析（8）

可以看到实现了懒惰初始化，在构造方法中仅仅计算了 table 的大小，以后在第一次使用时才会真正创建。

传入initialCapacity但不等于将会创建的hash表的大小，先要对照是否小于并发度concurrencyLevel，在通过tableSizeFor整理为2的n次方，因为hash表长度不是2的你次方会出错。

public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) {if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)throw new IllegalArgumentException();if (initialCapacity < concurrencyLevel) initialCapacity = concurrencyLevel; long size = (long)(1.0 + (long)initialCapacity / loadFactor);int cap = (size >= (long)MAXIMUM_CAPACITY)?MAXIMUM_CAPACITY : tableSizeFor((int)size);this.sizeCtl = cap;
}

get 流程（8）

public V get(Object key) {Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;//spread方法能确保返回结果是正数，因为负整数的哈希码有别的意义int h = spread(key.hashCode());// (n - 1) & h 位运算相当于取模操作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为负数表示该bin在扩容中或是treebin红黑树，这时调用find方法来查找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;
}ForwardingNode(Node<K,V>[] tab) {super(MOVED, null, null, null);this.nextTable = tab;

Put 流程（8）

// 其中 spread 方法会综合高位低位，具有更好的 hash 性
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) {// f 是链表头节点// fh 是链表头节点的 hash// i 是链表在 table 中的下标Node<K,V> f; int n, i, fh;// 要创建 tableif (tab == null || (n = tab.length) == 0)// 初始化 table 使用了 cas，无需 synchronized 创建成功，进入下一轮循环tab = initTable();// 要创建链表头节点else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {// 添加链表头使用了 cas，无需 synchronizedif (casTabAt(tab, i, null,new Node<K,V>(hash, key, value, null)))break;}// 帮忙扩容else if ((fh = f.hash) == MOVED)// 锁住链表，帮忙迁移，帮忙之后，进入下一轮循环tab = helpTransfer(tab, f);else {// 桶下标冲突V oldVal = null;// 锁住链表头节点synchronized (f) {// 再次确认链表头节点没有被移动if (tabAt(tab, i) == f) {// 链表if (fh >= 0) {binCount = 1;// 遍历链表for (Node<K,V> e = f;; ++binCount) {K ek;// 找到相同的 key 对比valueif (e.hash == hash &&((ek = e.key) == key ||(ek != null && key.equals(ek)))) {oldVal = e.val;// 更新if (!onlyIfAbsent)e.val = value;break;}Node<K,V> pred = e;// 已经是最后的节点了，新增 Node，追加至链表尾if ((e = e.next) == null) {pred.next = new Node<K,V>(hash, key,value, null);oldVal = p.val;if (!onlyIfAbsent)p.val = value;}}// 释放链表头节点的锁}if (binCount != 0) {if (binCount >= TREEIFY_THRESHOLD)// 如果链表长度 >= 树化阈值(8)，进行链表转为红黑树treeifyBin(tab, i);if (oldVal != null)return oldVal;break;}}// 增加 size 计数，还用于类似于LongAgger的多累加单元计数，提高并发度addCount(1L, binCount);return null;}

initable 原理（8）

保证只有一个线程可以创建线程。

private final Node<K,V>[] initTable() {Node<K,V>[] tab; int sc;// 当哈希表未初始化（tab为null或长度为0）时，持续循环while ((tab = table) == null || tab.length == 0) {// 如果sizeCtl为负数，说明已有线程在进行初始化或扩容，当前线程让出CPU时间片if ((sc = sizeCtl) < 0)Thread.yield();// 尝试通过CAS操作将sizeCtl设置为-1，以此表示当前线程要初始化哈希表else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {try {// 再次检查哈希表是否未初始化if ((tab = table) == null || tab.length == 0) {// 如果sizeCtl大于0，使用其值作为初始容量；否则使用默认容量（16）int n = (sc > 0)? sc : DEFAULT_CAPACITY;// 创建指定大小的哈希表数组Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];table = tab = nt;// 计算扩容阈值，n - (n >>> 2) 即 n - n/4 ，表示容量的3/4sc = n - (n >>> 2);}} finally {// 初始化完成后，设置sizeCtl为计算出的扩容阈值sizeCtl = sc;}break;}}return tab;
}

在 ConcurrentHashMap 中，sizeCtl（简称 sc ）有以下 4 种状态位：

  • 0：表示哈希表还未初始化 。在这种状态下，它代表着默认的初始化大小（DEFAULT_CAPACITY，即 16 ），后续会根据需要进行初始化操作。
  • -1：表示哈希表正在初始化 。当一个线程通过 CAS 操作将sizeCtl 从 0 设置为 -1 时，意味着该线程获得了初始化哈希表的权限，正在进行初始化工作。
  • 负数且不等于-1：比如 -N ，表示有 N - 1 个线程正在进行扩容操作 。在扩容过程中，通过这种方式来标识有多个线程参与协作扩容。
  • 正数：表示哈希表的扩容阈值 。即当哈希表中元素数量达到这个阈值时，会触发扩容操作，以保证哈希表的性能。

addCount 原理（8）

// check 是之前 binCount 的个数
private final void addCount(long x, int check) {CounterCell[] as; long b, s;if (// 已经有了 counterCells，向 cell 累加(as = counterCells) != null ||// 还没有，向 baseCount 累加!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {CounterCell a; long v; int m;boolean uncontended = true;if (// 还没有 counterCellsas == null || (m = as.length - 1) < 0 ||// 还没有 cell(a = as[ThreadLocalRandom.getProbe() & m]) == null ||// cell cas 增加计数失败!(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {// 创建累加单元数组和cell，累加重试fullAddCount(x, uncontended);return;}if (check <= 1)return;// 获取元素个数s = sumCount();}if (check >= 0) {Node<K,V>[] tab, nt; int n, sc;// 当元素个数s大于等于扩容阈值sizeCtl，且哈希表存在且未达到最大容量时，进入循环while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&(n = tab.length) < MAXIMUM_CAPACITY) {int rs = resizeStamp(n);if (sc < 0) {// 检查扩容相关标志是否符合预期，不符合则跳出循环if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||transferIndex <= 0)break;// newtable已经创建了，协助进行扩容if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))transfer(tab, nt);}// 需要扩容，此时newtable还未创建else if (U.compareAndSwapInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) + 2))transfer(tab, null);s = sumCount();}}}
}

addCount 维护的计数主要有以下作用：

  • 反映集合元素数量

  • 触发扩容操作

  • 辅助数据结构转换判断

size计算流程（8）

size计算实际发生在put，remove改变集合元素的操作之中

  • 没有竞争发生，向baseCount累加计数

  • 有竞争发生，新建counterCells，向其中的一个cell累加计数

    • counterCells初始有两个cell

    • 如果计数竞争比较激烈，会创建新的cell来累加计数

注意：虽然size和LongAddr同样使用累加单元保证原子性，但size不能保证size的过程中总数是否被改变，具有弱一致性。

public int size() {// 调用sumCount方法获取元素数量的总和long n = sumCount();// 对获取的数量进行范围处理：如果小于0则返回0；如果超过Integer的最大值则返回Integer的最大值；否则返回转换为int后的数值return (n < 0L)? 0 :(n > (long)Integer.MAX_VALUE)? Integer.MAX_VALUE :(int)n;
}final long sumCount() {CounterCell[] as = counterCells; CounterCell a;// 初始化总和为baseCount计数long sum = baseCount;if (as != null) {// 遍历counterCells数组，将每个有效的CounterCell中的计数值累加到sum中for (int i = 0; i < as.length; ++i) {if ((a = as[i]) != null)sum += a.value;}}return sum;
}

transfer（8）

每次扩容都是计算扩容后大小，并不直接创建实例，只有使用时才创建。

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; // subdivide rangeif (nextTab == null) {  // initiating，即初始化扩容相关操作try {// 未检查异常块（这里主要是创建新的哈希表数组，可能会有内存相关异常）@SuppressWarnings("unchecked")// 懒加载创建哈希表数组Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];// 创建新的哈希表数组，大小为原数组n的2倍，用于扩容后的存储nextTab = nt;} catch (Throwable ex) {  // try to cope with OOME，尝试处理内存溢出错误（OOM）// 如果创建新数组时出现异常，将sizeCtl设置为Integer的最大值，阻止后续扩容操作sizeCtl = Integer.MAX_VALUE;return;}nextTable = nextTab;// 设置transferIndex为原数组大小n，用于标识扩容迁移的起始位置transferIndex = n;}int nextn = nextTab.length;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) { ... }if (i < 0 || i >= n || i + n >= nextn) { ... }else if ((f = tabAt(tab, i)) == null)advance = casTabAt(tab, i, null, fwd); // 处理完将列表头替换为ForwardingNodeelse if ((fh = f.hash) == MOVED)advance = true; // already processedelse {// 有元素开始处理synchronized (f) {if (tabAt(tab, i) == f) {Node<K,V> ln, hn;if (fh >= 0) { ... }else if (f instanceof TreeBin) { ... }}}}}
}

构造器分析（7）

它维护了一个 segment 数组，每个 segment 对应一把锁

  • 优点：如果多个线程访问不同的 segment，实际是没有冲突的，这与 jdk8 中是类似的

  • 缺点：Segments 数组默认大小为16，这个容量初始化指定后就不能改变了，并且不是懒惰初始化

public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) {if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)throw new IllegalArgumentException();if (concurrencyLevel > MAX_SEGMENTS)concurrencyLevel = MAX_SEGMENTS;// ssize 必须是 2^n，即 2, 4, 8, 16 ... 表示了 segments 数组的大小int sshift = 0;int ssize = 1;while (ssize < concurrencyLevel) {++sshift;ssize <<= 1;}// segmentShift 默认是 32 - 4 = 28this.segmentShift = 32 - sshift;// segmentMask 默认是 15 即 0000 0000 0000 1111this.segmentMask = ssize - 1;if (initialCapacity > MAXIMUM_CAPACITY)initialCapacity = MAXIMUM_CAPACITY;int c = initialCapacity / ssize;if (c * ssize < initialCapacity)++c;int cap = MIN_SEGMENT_TABLE_CAPACITY;while (cap < c)cap <<= 1;// 创建 segments and segments[0]Segment<K,V> s0 =new Segment<K,V>(loadFactor, (int)(cap * loadFactor),(HashEntry<K,V>[])new HashEntry[cap]);// 可以看到扩容后直接创建实例，非懒加载Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]this.segments = ss;
}

在这里插入图片描述

定位segment（7）

  1. 确定段数组大小（ssize）
    ssize 是大于或等于 concurrencyLevel 的最小的 2 的幂次方。例如：
    • 若 concurrencyLevel = 16，则 ssize = 16（2⁴）。
    • 若 concurrencyLevel = 17，则 ssize = 32（2⁵）。
  2. 计算段索引位移量（sshift）
    sshift 表示 ssize 对应的二进制位数减 1。例如：
    • 若 ssize = 16（2⁴），则 sshift = 4。
    • 若 ssize = 32（2⁵），则 sshift = 5。
  3. 计算 segmentShift
    segmentShift 的值为 32 - sshift。例如：
    • 若 sshift = 4，则 segmentShift = 32 - 4 = 28。
    • 若 sshift = 5，则 segmentShift = 32 - 5 = 27。

主要是有segmentShift和segmentMask计算决定将key的hash结果匹配到哪个segment。

在这里插入图片描述

Put（7）

segment数组和第一个segment节点不是懒加载得，其他节点是懒加载。

public V put(K key, V value) {Segment<K,V> s;if (value == null)throw new NullPointerException();int hash = hash(key);// 计算出 segment 下标int j = (hash >>> segmentShift) & segmentMask;// 获得 segment 对象，判断是否为 null，是则创建该 segmentif ((s = (Segment<K,V>)UNSAFE.getObject(segments, (j << SSHIFT) + SBASE)) == null) {// 这时不能确定是否真的为 null，因为其它线程也发现该 segment 为 null，// 因此在 ensureSegment 里用 cas 方式保证该 segment 安全性s = ensureSegment(j);}// 进入 segment 的put 流程return s.put(key, hash, value, false);
}

segment继承了可重入锁（ReentrantLock），它的put方法为：

注意：这里得扩容是对segment节点单独扩容。各个segment节点相互独立。

final V put(K key, int hash, V value, boolean onlyIfAbsent) {// 尝试加锁HashEntry<K,V> node = tryLock()? null :// 如果不成功，进入 scanAndLockForPut 流程// 如果是多核 cpu 最多 tryLock 64 次，进入 lock 流程// 在尝试期间，还可以顺便看该节点在链表中有没有，如果没有顺便创建出来scanAndLockForPut(key, hash, value);// 执行到这里 segment 已经被成功加锁，可以安全执行V oldValue;try {HashEntry<K,V>[] tab = table;int index = (tab.length - 1) & hash;HashEntry<K,V> first = entryAt(tab, index);for (HashEntry<K,V> e = first;;) {if (e != null) {// 更新K k;if ((k = e.key) == key ||(e.hash == hash && key.equals(k))) {oldValue = e.value;if (!onlyIfAbsent) {e.value = value;++modCount;}e = e.next;}else {// 新增// 1) 之前等待锁时，node 已经被创建，next 指向链表头if (node != null)node.setNext(first);else// 2) 创建新 nodenode = new HashEntry<K,V>(hash, key, value, first);int c = count + 1;// 3) 扩容if (c > threshold && tab.length < MAXIMUM_CAPACITY)rehash(node);else// 将 node 作为链表头setEntryAt(tab, index, node);++modCount;count = c;oldValue = null;break;}}} finally {unlock();}}

rehash（7）

扩容发生在put时，put时新增的节点还没有插入，会在扩容后再计算并插入新的位置。

示例对于索引为1位置下有链表1-1-1-1-1，当扩容后会根据新的hash数组检查链表中所有节点得哈希位置（索引位）是否改变，链表1-1-1-1-1计算后是1-3-1-1-1发现第二位位置发生改变，那第二位之后得节点直接移动（尾插法）到新的hash数组的索引为1位置，实现节点复用，而第二位和第一位都需要new节点并链接到新的索引位（不是真正的移动）。

private void rehash(HashEntry<K,V> node) {HashEntry<K,V>[] oldTable = table;int oldCapacity = oldTable.length;int newCapacity = oldCapacity << 1;threshold = (int)(newCapacity * loadFactor);HashEntry<K,V>[] newTable =(HashEntry<K,V>[]) new HashEntry[newCapacity];int sizeMask = newCapacity - 1;for (int i = 0; i < oldCapacity ; i++) {HashEntry<K,V> next = e.next;int idx = e.hash & sizeMask;if (next == null)  // Single node on listnewTable[idx] = e;else { // Reuse consecutive sequence at same slotHashEntry<K,V> lastRun = e;int lastIdx = idx;// 过一遍链表，尽可能把 rehash 后 idx 不变的节点重用for (HashEntry<K,V> last = next;last != null;last = last.next) {int k = last.hash & sizeMask;if (k != lastIdx) {lastIdx = k;lastRun = last;}}newTable[lastIdx] = lastRun;// 剩余节点需要新建for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {V v = p.value;int h = p.hash;int k = h & sizeMask;HashEntry<K,V> n = newTable[k];newTable[k] = new HashEntry<K,V>(h, p.key, v, n);}}}
}// 扩容完成，才加入新的节点
int nodeIndex = node.hash & sizeMask;//add thenewnode
node.setNext(newTable[nodeIndex]);
newTable[nodeIndex]=node;// 替换为新的HashEntrytable
table = newTable;
}

get 流程（7）

get时并未加锁，用了UNSAFE 方法保证了可见性，扩容过程中，get先发生就从I旧表取内容，get后发生就从新表取内容。

由于无论是Segment列表还是hashEntry[]，需要保证数组元素的可见性，加volatile修饰数组本身是不行的，对于数组元素必须要必须配合这个unsafe对象的这个getObjectVolatile保证。

public V get(Object key) {Segment<K,V> s; // manually integrate access methods to reduce overheadHashEntry<K,V>[] tab;int h = hash(key);// u 为 segment 对象在数组中的偏移量long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;// s 即为 segmentif ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&(tab = s.table) != null) {for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile(tab, (((long)((tab.length - 1) & h)) << TSHIFT) + TBASE);e != null; e = e.next) {K k;// 定位到桶下标，检查keyif (((k = e.key) == key || (e.hash == h && key.equals(k)))return e.value;}}return null;
}

size计算流程（7）

size弱一致性。

public int size() {// Try a few times to get accurate count. On failure due to// continuous async changes in table, resort to locking.final Segment<K,V>[] segments = this.segments;int size;boolean overflow; // true if size overflows 32 bitslong sum;        // sum of modCountslong last = 0L;  // previous sumint retries = -1; // first iteration isn't retrytry {for (;;) {if (retries++ == RETRIES_BEFORE_LOCK) {// 超过重试次数，需要创建所有 segment 并加锁，没超过重试次数，就cas修改for (int j = 0; j < segments.length; ++j)ensureSegment(j).lock(); // force creation}sum = 0L;// 记录计算size过程中的修改次数size = 0;// 记录sizeoverflow = false;// 遍历每个segment对象for (int j = 0; j < segments.length; ++j) {Segment<K,V> seg = segmentAt(segments, j);if (seg != null) {// 每有修改就++sum += seg.modCount;int c = seg.count;if (c < 0 || (size += c) < 0)overflow = true;}}// 比较size计算过程前后有没有修改，没有修改则size有效if (sum == last)break;last = sum;}} finally {if (retries > RETRIES_BEFORE_LOCK) {for (int j = 0; j < segments.length; ++j)// 超过重试次数，需要创建所有 segment 并加锁，现在unlocksegmentAt(segments, j).unlock();}}return overflow? Integer.MAX_VALUE : size;
}
http://www.lryc.cn/news/579489.html

相关文章：

  • Linux多线程（十二）之【生产者消费者模型】
  • 汽车ECU产线烧录和检测软件怎么做？
  • Flutter 3.29+使用isar构建失败
  • HarmonyOS ArkTS卡片堆叠滑动组件实战与原理详解（含源码）
  • Java网络编程：TCP/UDP套接字通信详解
  • I/O 进程 7.2
  • 在Ubuntu 24.04主机上创建Ubuntu 14.04编译环境的完整指南
  • （一）复习（模块注入/minimal api/EF和Dapper实现CQRS）
  • Ubuntu Gnome 安装和卸载 WhiteSur-gtk-theme 类 Mac 主题的正确方法
  • Frida：配置自动补全 in VSCode
  • TCP 三次握手与四次挥手详解
  • MyBatis 之基础概念与框架原理详解
  • RabbitMQ 通过HTTP API删除队列命令
  • 【如何判断Linux系统是Ubuntu还是CentOS】
  • Centrifugo 深度解析：构建高性能实时应用的开源引擎
  • 记忆翻牌记忆力小游戏流量主微信小程序开源
  • 网创vip课程视频教程、付费网络课程以及网赚培训，学习引流、建站、赚钱。8个T的全套课程
  • 【2.3 漫画SpringSecurity - 守护应用安全的钢铁卫士】
  • ATE FT ChangeKit学习总结-20250630
  • Easy-excel监听器中对批量上传的工单做错误收集
  • Redisson使用示例
  • 请求未达服务端？iOS端HTTPS链路异常的多工具抓包排查记录
  • 【Bug Recod】更新中...
  • Day50
  • 一文详解Character AI：实用指南+ ChatGPT、Gemini对比分析
  • contenteditable网页富文本编辑无法选中图片
  • Swift 的基础设计哲学是 “通过模块化组合实现安全与效率的平衡“，就像用标准化工业零件建造摩天大楼
  • 一台香港原生ip站群服务器多少钱？
  • 如何在Ubuntu上检查MySQL是否启动并放开3306端口
  • C++笔记-位图和布隆过滤器