ConcurrentHashMap
ConcurrentHashMap 源码分析
线程安全哈希表。JDK 8 起放弃分段锁 Segment,改成「空桶 CAS + 非空桶 synchronized 锁头节点」,锁粒度从一段桶缩小到一个桶;读操作全程无锁,计数用 LongAdder 思路分散热点,扩容允许多线程分片协作。
代码块收起展开
// 基于 JDK 17 (本地 ms-17.0.18), java.util.concurrent.ConcurrentHashMap
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
implements ConcurrentMap<K,V>, Serializable {
// 负 hash 专用于控制节点。正常节点 hash 被 spread() 抹掉符号位保证非负，
// 所以读写路径只看一个 int 就能分辨"数据节点"还是"特殊节点"，不用 instanceof
static final int MOVED = -1; // ForwardingNode：该桶已迁走，去 nextTable 找
static final int TREEBIN = -2; // 红黑树根（TreeBin 代理节点）
static final int RESERVED = -3; // computeIfAbsent 占位节点
static final int HASH_BITS = 0x7fffffff; // 抹符号位的掩码
static final int TREEIFY_THRESHOLD = 8; // 链长达 8 转红黑树（还要求表容量 >= 64，否则先扩容）
// 桶数组。volatile 保证扩容换表后其它线程立刻可见；懒初始化，首次插入才建表
transient volatile Node<K,V>[] table;
// 扩容时的新表，仅迁移期间非 null
private transient volatile Node<K,V>[] nextTable;
// 低竞争时的计数器，CAS 更新；冲突后计数分散到 counterCells
private transient volatile long baseCount;
// 一个字段身兼数职（省内存 + 单点 CAS 即可协调初始化/扩容）：
// -1 正在初始化
// 其它负数 正在扩容，低 16 位 = 参与扩容的线程数 + 1
// 0 或正数 未建表时是初始容量，建表后是下次扩容阈值（0.75n）
private transient volatile int sizeCtl;
// 扩容时下一个待认领的桶区间上界，线程 CAS 它来"领任务"
private transient volatile int transferIndex;
// 计数分散单元数组，思路同 LongAdder 的 cells
private transient volatile CounterCell[] counterCells;
// 高 16 位异或进低 16 位：表小的时候索引只用低位，把高位的差异掺进来减少碰撞；
// & HASH_BITS 抹符号位，给负 hash 让路（MOVED/TREEBIN/RESERVED）
static final int spread(int h) {
return (h ^ (h >>> 16)) & HASH_BITS;
}
// 读桶必须走 volatile 语义（acquire），否则可能读到别的线程 CAS 写入前的旧值
@SuppressWarnings("unchecked")
static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
return (Node<K,V>)U.getReferenceAcquire(tab, ((long)i << ASHIFT) + ABASE);
}
// 空桶插入的基础：CAS 保证并发插同一个空桶只有一个赢
static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
Node<K,V> c, Node<K,V> v) {
return U.compareAndSetReference(tab, ((long)i << ASHIFT) + ABASE, c, v);
}
// 只在持有桶锁时调用，所以 release 语义就够，不需要完整 CAS
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
U.putReferenceRelease(tab, ((long)i << ASHIFT) + ABASE, v);
}
// get 全程无锁：volatile 读 table + tabAt 读桶 + Node.val 也是 volatile
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
int h = spread(key.hashCode());
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：多态分发给 find()——
return (p = e.find(h, key)) != null ? p.val : null; // ForwardingNode 转发到新表，TreeBin 走树查
while ((e = e.next) != null) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}
public V put(K key, V value) {
return putVal(key, value, false);
}
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException(); // 拒绝 null：并发下无法区分"不存在"和"值为 null"
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) { // 无限重试：CAS 失败、帮完扩容都回到这里重来
Node<K,V> f; int n, i, fh; K fk; V fv;
if (tab == null || (n = tab.length) == 0)
tab = initTable(); // 懒初始化，CAS 抢 sizeCtl
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value)))
break; // 空桶直接 CAS 放入，全程无锁——最快路径
}
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f); // 撞上迁移标记：先帮忙搬完这轮，再回来插新表
else if (onlyIfAbsent // putIfAbsent 命中头节点时不加锁直接返回旧值
&& fh == hash
&& ((fk = f.key) == key || (fk != null && key.equals(fk)))
&& (fv = f.val) != null)
return fv;
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;
if (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;
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key, value); // 尾插（JDK7 头插在扩容时会成环，8 起彻底改掉）
break;
}
}
}
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
else if (f instanceof ReservationNode) // computeIfAbsent 的函数里又改同一个桶 -> 直接抛
throw new IllegalStateException("Recursive update");
}
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i); // 里面还会判容量：< 64 时选择扩容而非树化
if (oldVal != null)
return oldVal; // 覆盖旧值不改元素个数，不走 addCount
break;
}
}
}
addCount(1L, binCount); // 计数 +1，顺带检查是否要扩容
return null;
}
}计数与扩容触发。size 不用 AtomicLong 而抄 LongAdder:所有写线程去争同一个计数器会把并发优势全部抵消,所以宁可把”精确”换成”分散”。
代码块收起展开
// 基于 JDK 17 (本地 ms-17.0.18), java.util.concurrent.ConcurrentHashMap（以下均为其成员）
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(); // 抢初始化输了：让出 CPU 自旋等，不阻塞（初始化很快，不值得挂起）
else if (U.compareAndSetInt(this, SIZECTL, sc, -1)) { // CAS 置 -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;
sc = n - (n >>> 2); // 阈值 = n - n/4 = 0.75n，位运算避免浮点
}
} finally {
sizeCtl = sc; // finally 里恢复，建表抛异常也不会把 -1 卡死
}
break;
}
}
return tab;
}
private final void addCount(long x, int check) {
CounterCell[] cs; long b, s;
if ((cs = counterCells) != null ||
!U.compareAndSetLong(this, BASECOUNT, b = baseCount, s = b + x)) { // 无竞争时只 CAS baseCount，一步到位
CounterCell c; long v; int m;
boolean uncontended = true;
if (cs == null || (m = cs.length - 1) < 0 ||
(c = cs[ThreadLocalRandom.getProbe() & m]) == null || // 用线程探针值散列到自己的 cell
!(uncontended =
U.compareAndSetLong(c, CELLVALUE, v = c.value, v + x))) {
fullAddCount(x, uncontended); // cell 也撞了：进完整流程（建/扩 cells 数组，同 Striped64）
return;
}
if (check <= 1)
return;
s = sumCount();
}
if (check >= 0) { // remove 传负 check，缩容不存在，不用检查
Node<K,V>[] tab, nt; int n, sc;
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
int rs = resizeStamp(n) << RESIZE_STAMP_SHIFT; // 扩容戳：高 16 位标识"从容量 n 扩"这一代
if (sc < 0) { // 已有人在扩：尝试加入
if (sc == rs + MAX_RESIZERS || sc == rs + 1 ||
(nt = nextTable) == null || transferIndex <= 0) // 人满/已收尾/没剩余任务就不掺和
break;
if (U.compareAndSetInt(this, SIZECTL, sc, sc + 1)) // 线程数 +1
transfer(tab, nt);
}
else if (U.compareAndSetInt(this, SIZECTL, sc, rs + 2)) // 第一个发起者：戳 + 2（2 表示 1 个线程）
transfer(tab, null);
s = sumCount();
}
}
}
// @Contended 填充缓存行，防止相邻 cell 伪共享互相打脏缓存
@jdk.internal.vm.annotation.Contended static final class CounterCell {
volatile long value;
CounterCell(long x) { value = x; }
}
// 无锁遍历求和：读的过程中别的线程还在加，结果天然是弱一致的近似值
final long sumCount() {
CounterCell[] cs = counterCells;
long sum = baseCount;
if (cs != null) {
for (CounterCell c : cs)
if (c != null)
sum += c.value;
}
return sum;
}
public int size() {
long n = sumCount();
return ((n < 0L) ? 0 :
(n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
(int)n);
}多线程协作扩容。核心思想:把迁移任务按 stride 切成区间,谁来谁认领一段;迁完的桶插 ForwardingNode 路标,读到它就去新表。
代码块收起展开
// 基于 JDK 17 (本地 ms-17.0.18), java.util.concurrent.ConcurrentHashMap（以下均为其成员）
// 写线程撞到 ForwardingNode 时调用：与其干等，不如加入扩容加速它结束
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
Node<K,V>[] nextTab; int sc;
if (tab != null && (f instanceof ForwardingNode) &&
(nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
int rs = resizeStamp(tab.length) << RESIZE_STAMP_SHIFT;
while (nextTab == nextTable && table == tab &&
(sc = sizeCtl) < 0) { // 三个条件任一变化都说明这轮扩容已结束
if (sc == rs + MAX_RESIZERS || sc == rs + 1 ||
transferIndex <= 0)
break;
if (U.compareAndSetInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}
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; // 每个线程一次至少领 16 个桶，太碎反而全在争 transferIndex
if (nextTab == null) { // 发起者负责建 2 倍新表
try {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
nextTab = nt;
} catch (Throwable ex) { // OOM 兜底：阈值拉满，放弃扩容
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
transferIndex = n; // 任务从表尾往表头发放
}
int nextn = nextTab.length;
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab); // hash = MOVED 的路标，所有迁完的桶共用这一个
boolean advance = true;
boolean finishing = false;
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
while (advance) { // 认领任务：CAS 把 transferIndex 往前推 stride，得到自己的 [bound, i] 区间
int nextIndex, nextBound;
if (--i >= bound || finishing)
advance = false;
else if ((nextIndex = transferIndex) <= 0) {
i = -1; // 任务发完了
advance = false;
}
else if (U.compareAndSetInt
(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); // 2n - n/2 = 0.75 * 2n
return;
}
if (U.compareAndSetInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { // 线程数 -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); // 空桶也要插路标，否则并发 put 会插进旧表丢数据
else if ((fh = f.hash) == MOVED)
advance = true; // 已被别的线程迁过
else {
synchronized (f) { // 迁移和 putVal 用同一把桶锁，天然互斥
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
if (fh >= 0) {
// 容量翻倍后 hash & n 只有 0/1 两种结果：0 留原下标 i（ln 链），1 去 i+n（hn 链）
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; // 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); // lastRun 之前的节点复制新建（头插到 ln/hn），
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;
}
// ... TreeBin 拆分与 ReservationNode 检查省略：树同样按 (hash & n) 拆成 lo/hi 两串，
// 拆完一侧节点数 <= UNTREEIFY_THRESHOLD(6) 就退化回链表（untreeify）
}
}
}
}
}原理串讲
一次 put(key, value) 的完整链路:先 spread() 把 hashCode 高 16 位异或进低位并抹掉符号位,然后进 putVal() 的无限循环。
表还没建就走 initTable()——用 CAS 把 sizeCtl 置成 -1 当”初始化锁”,输掉的线程 Thread.yield() 自旋等待。
定位到桶后分三条路:桶为空就 casTabAt() 无锁插入,成功直接结束,这是最快路径;桶头是 MOVED 说明正在扩容,走 helpTransfer() 帮忙搬完再重试;否则 synchronized (f) 锁住头节点,锁内先用 tabAt(tab, i) == f 复查(拿锁的间隙头节点可能已被删除或迁移,复查失败就外层循环重来),然后按链表尾插或 TreeBin.putTreeVal() 插入。
为什么锁头节点用 synchronized 而不用 ReentrantLock?每个桶头都可能成为锁,ReentrantLock 要为每把锁背一个 AQS 对象,百万级桶的内存开销不可接受;而 synchronized 直接复用节点自身的对象头 markword,零额外内存,且桶锁的临界区极短(遍历一条通常不超过 8 个节点的链),JVM 的自适应自旋和锁升级恰好擅长这种场景。
插入成功后走 addCount(1L, binCount)。计数先 CAS baseCount,失败说明有竞争,转而用线程探针值散列到 counterCells 的某个格子上 CAS,再失败进 fullAddCount() 建格或扩格——这就是 LongAdder。
为什么不用一个 AtomicLong?因为每次 put/remove 都要计数,所有线程 CAS 同一个变量会把刚刚用桶锁拆散的竞争重新聚拢到一个点上,热点又回来了。
代价是 size() 只能 sumCount() 遍历求近似和,弱一致——并发容器把”完全精确的 size”这个本来就没意义的语义(返回瞬间就可能变了)卖掉,换回了写入的线性扩展。
计数后若 s >= sizeCtl 就触发扩容:发起者 CAS 把 sizeCtl 置成 (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2(高 16 位是这一代扩容的戳,防止上一轮的残余线程混进下一轮;低 16 位是线程数+1),然后进 transfer()。
每个参与线程 CAS transferIndex 认领一段至少 16 个桶的区间,从表尾往表头搬。
搬一个桶时同样 synchronized 锁头节点,利用容量是 2 的幂的性质,按 hash & n 是 0 还是 1 把链拆成留在原下标的 ln 链和去 i + n 的 hn 链——先找 lastRun(链尾去向一致的整段直接复用),其余节点复制新建,旧链原封不动,所以读线程在迁移进行中仍能安全遍历旧表。
桶迁完用 setTabAt(tab, i, fwd) 插入 ForwardingNode:此后读到它的 get() 经由 ForwardingNode.find() 转发到新表,写线程则被 helpTransfer() 拉来当劳动力。
最后一个退出的线程(sc - 2 == resizeStamp(n) << RESIZE_STAMP_SHIFT)复查一遍全表后提交:table = nextTab,sizeCtl 设为新阈值。
get() 从头到尾没有锁:table 是 volatile,tabAt() 用 acquire 语义读桶,Node.val 和 next 也是 volatile,所以写线程的修改对读线程即时可见。
负 hash 的桶头多态分发给 e.find(),ForwardingNode 的实现跳到新表接着找,TreeBin 的实现走树——读路径不需要知道扩容正在发生。
设计取舍
- 禁 null 键值:
get返回 null 必须无歧义地等于”不存在”,否则并发下containsKey+get两步之间 map 可能已被改,null 的含义没法确认。 size()弱一致,get也可能读到调用瞬间之后被删除的值:单个操作原子,跨操作不保证快照语义,迭代器同理(弱一致,不抛 ConcurrentModificationException)。- 复合操作不原子:
if (!map.containsKey(k)) map.put(k, v)依然有竞态,要用putIfAbsent/compute/merge。 - 扩容不是 stop-the-world:读走 ForwardingNode 转发,写来帮忙搬桶,代价是 put 的最坏延迟被搬桶拖长(延迟换吞吐)。
- 树化阈值 8 且要求容量 >= 64:hash 均匀时链长到 8 的概率约千万分之一(泊松分布),真到 8 大概率是 hashCode 太烂,树化只是兜底而非常态。