Java集合框架ConcurrentHashMap的源码深度：实现极致并发的红黑树机制
大家好，今天我们来深入探讨Java集合框架中的ConcurrentHashMap，特别是它在实现极致并发时所采用的红黑树机制。ConcurrentHashMap是Java并发编程中一个非常重要的类，它提供了线程安全的哈希表实现，在高并发场景下拥有卓越的性能。我们将从源码层面分析其并发机制和红黑树的应用。
一、ConcurrentHashMap的基本结构
ConcurrentHashMap并没有采用全局锁来保证线程安全，而是采用了分段锁（Segment）机制，在JDK 8之后，废弃了Segment，而是采用了Node数组 + CAS + synchronized来保证并发安全。
-
Node:
Node是ConcurrentHashMap中最基本的存储单元，它封装了key-value键值对。它是一个不可变的类，一旦创建，其hash值和key值都不能被修改。static class Node<K,V> implements Map.Entry<K,V> { final int hash; final K key; volatile V val; volatile Node<K,V> next; Node(int hash, K key, V val, Node<K,V> next) { this.hash = hash; this.key = key; this.val = val; this.next = next; } // ... 省略 get/set 方法 ... } -
Table:
ConcurrentHashMap内部维护一个Node<K,V>[] table数组，它是一个哈希表，用于存储所有的键值对。table的大小是2的幂次方，这样可以通过位运算来快速定位元素。 -
sizeCtl: 这是一个非常重要的控制字段，用于控制
ConcurrentHashMap的初始化和扩容。它的不同值代表不同的状态：- 小于0：表示
ConcurrentHashMap正在进行初始化或扩容操作。- -1：表示正在初始化。
- -N：表示有N-1个线程正在进行扩容操作。
- 等于0：表示
ConcurrentHashMap还没有被初始化。 - 大于0：
- 初始化之后，代表下一次进行扩容时的阈值。
- 在初始化之前，代表初始容量。
- 小于0：表示
二、并发机制：CAS + synchronized
ConcurrentHashMap的并发机制是基于CAS（Compare and Swap）和synchronized来实现的。
-
CAS: CAS是一种原子操作，用于无锁地更新变量。它包含三个操作数：内存地址V，预期值A，新值B。如果内存地址V的值等于预期值A，那么将内存地址V的值更新为B，否则什么都不做。
ConcurrentHashMap使用CAS来更新table中的节点，避免了使用锁带来的性能开销。static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i, Node<K,V> c, Node<K,V> v) { return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v); }其中，
U是Unsafe类的实例，ASHIFT和ABASE用于计算数组元素的内存地址。 -
synchronized:synchronized关键字用于保证在更新链表或红黑树时只有一个线程可以访问。虽然ConcurrentHashMap尽量避免使用锁，但在某些情况下，例如在插入或删除节点时，仍然需要使用synchronized来保证数据的一致性。不过，这里的synchronized锁的是链表或红黑树的头节点，而不是整个table，这样可以大大减少锁的粒度，提高并发性能。
三、ConcurrentHashMap的初始化
ConcurrentHashMap的初始化过程是懒加载的，只有在第一次插入元素时才会进行初始化。初始化过程由initTable()方法完成。
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;
sc = n - (n >>> 2); // n - n/4 loadFactor = 0.75
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}- 检查
table是否为空: 如果table为空，则进入初始化流程。 - 检查
sizeCtl:sizeCtl用于控制初始化和扩容。如果sizeCtl小于0，说明有其他线程正在进行初始化或扩容，当前线程让出CPU时间片。 - CAS更新
sizeCtl: 如果sizeCtl大于等于0，说明当前线程可以进行初始化。使用CAS将sizeCtl设置为-1，表示当前线程正在进行初始化。 - 初始化
table: 如果table仍然为空，则创建一个新的Node数组，并将其赋值给table。table的大小由sizeCtl决定，如果sizeCtl大于0，则使用sizeCtl作为table的大小，否则使用默认大小16。 - 更新
sizeCtl: 初始化完成后，更新sizeCtl为n - (n >>> 2)，其中n为table的大小。这个值表示下一次进行扩容的阈值。
四、ConcurrentHashMap的put()方法
ConcurrentHashMap的put()方法用于插入键值对。它的实现非常复杂，需要考虑多种情况，例如table是否为空、节点是否存在、是否需要扩容等。
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();
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
if (tab == null || (n = tab.length) == 0)
tab = initTable();
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);
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, null);
break;
}
}
}
else if (f instanceof TreeBin) {
Node<K,V> p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value);
binCount = 2;
if (p != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}- 检查
key和value是否为空: 如果key或value为空，则抛出NullPointerException。 - 计算
hash值: 使用spread()方法计算key的hash值。 - 无限循环: 进入一个无限循环，直到插入成功或发生异常。
- 检查
table是否为空: 如果table为空，则调用initTable()方法进行初始化。 - 定位节点: 使用
(n - 1) & hash计算节点在table中的位置。 - 如果节点为空: 如果节点为空，则使用CAS将新的
Node插入到该位置。 - 如果节点正在进行扩容: 如果节点的
hash值为MOVED，说明该节点正在进行扩容，则调用helpTransfer()方法帮助扩容。 - 如果节点不为空: 如果节点不为空，则需要对该节点进行加锁，然后进行插入或更新操作。
- 如果节点是链表: 遍历链表，如果找到相同的
key，则更新value，否则将新的Node插入到链表的末尾。 - 如果节点是红黑树: 调用
TreeBin.putTreeVal()方法将新的Node插入到红黑树中。
- 如果节点是链表: 遍历链表，如果找到相同的
- 检查是否需要转换为红黑树: 如果链表的长度超过
TREEIFY_THRESHOLD（默认为8），则调用treeifyBin()方法将链表转换为红黑树。 - 更新元素个数: 调用
addCount()方法更新ConcurrentHashMap中元素的个数。
五、红黑树的应用：treeifyBin()和TreeBin
当链表长度超过一定阈值（TREEIFY_THRESHOLD，默认为8）时，ConcurrentHashMap会将链表转换为红黑树，以提高查找效率。红黑树的实现主要由treeifyBin()方法和TreeBin类完成。
-
treeifyBin():treeifyBin()方法用于将链表转换为红黑树。它首先检查table的长度是否小于MIN_TREEIFY_CAPACITY（默认为64），如果小于，则进行扩容，否则将链表转换为红黑树。private final void treeifyBin(Node<K,V>[] tab, int index) { Node<K,V> b; int n, sc; if (tab != null) { if ((n = tab.length) < MIN_TREEIFY_CAPACITY) { tryPresize(n << 1); } else if ((b = tabAt(tab, index)) != null && b.hash >= 0) { synchronized (b) { if (tabAt(tab, index) == b) { TreeNode<K,V> hd = null, tl = null; for (Node<K,V> e = b; e != null; e = e.next) { TreeNode<K,V> p = new TreeNode<K,V>(e.hash, e.key, e.val, null, null); if ((hd = p.prev = (tl == null) ? p : tl) == null) hd = p; else tl.next = p; tl = p; } setTabAt(tab, index, new TreeBin<K,V>(hd)); } } } } } -
TreeBin:TreeBin是红黑树的根节点。它继承自Node，但它并不存储键值对，而是存储红黑树的根节点。TreeBin类提供了一系列方法用于操作红黑树，例如putTreeVal()、removeTreeNode()等。static final class TreeBin<K,V> extends Node<K,V> { TreeNode<K,V> root; volatile TreeNode<K,V> first; volatile Thread waiter; volatile int lockState; // values for lockState static final int WRITER = 1; // set while holding write lock static final int WAITER = 2; // set when waiting TreeBin(TreeNode<K,V> b) { super(TREEBIN, null, null, null); this.first = b; TreeNode<K,V> root = null; for (TreeNode<K,V> x = b, next; x != null; x = next) { next = (TreeNode<K,V>)x.next; x.left = x.right = null; if (root == null) { x.parent = null; x.red = false; root = x; } else { K k = x.key; int h = x.hash; Class<?> kc = null; for (TreeNode<K,V> p = root;;) { int dir, ph; K pk = p.key; if ((ph = p.hash) > h) dir = -1; else if (ph < h) dir = 1; else if ((kc == null && (kc = comparableClassFor(k)) == null) || (dir = compareComparables(kc, k, pk)) == 0) dir = tieBreakOrder(k, pk); TreeNode<K,V> xp = p; if ((p = (dir <= 0) ? p.left : p.right) == null) { x.parent = xp; if (dir <= 0) xp.left = x; else xp.right = x; root = balanceInsertion(root, x); break; } } } } this.root = root; assert checkInvariants(root); } // ... 省略 putTreeVal(), removeTreeNode() 等方法 ... }
六、扩容机制：transfer()
当ConcurrentHashMap中的元素个数超过阈值时，就需要进行扩容。ConcurrentHashMap的扩容机制是渐进式的，它不会一次性将所有的元素都迁移到新的table中，而是每次只迁移一部分。扩容过程由transfer()方法完成。
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 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<K,V> fwd = new ForwardingNode<K,V>(nextTab);
boolean advance = true;
boolean finishing = false; // to ensure that we only do one sweep on exit
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; // recheck before commit
}
}
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 = null, hn = null;
for (Node<K,V> e = f; e != null; e = e.next) {
int rh = e.hash & n;
if (rh == 0) {
ln = new Node<K,V>(e.hash, e.key, e.val, ln);
}
else {
hn = new Node<K,V>(e.hash, e.key, e.val, hn);
}
}
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
}
}
}
}- 计算
stride:stride表示每个线程负责迁移的节点数量。它的大小由CPU核心数和table的大小决定。 - 创建
nextTab: 创建一个新的Node数组，其大小是table的两倍。 - 创建
ForwardingNode:ForwardingNode是一个特殊的节点，用于标记该节点正在进行迁移。 - 循环迁移节点: 每个线程负责迁移一部分节点。对于每个节点，需要重新计算其
hash值，然后将其插入到新的table中。 - 设置
ForwardingNode: 迁移完成后，将原来的节点设置为ForwardingNode，表示该节点已经迁移完成。 - 更新
sizeCtl: 所有线程都完成迁移后，更新sizeCtl为新的阈值。
七、ConcurrentHashMap的get()方法
ConcurrentHashMap的get()方法用于获取键值对。由于ConcurrentHashMap是线程安全的，因此get()方法不需要加锁。
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)
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;
}- 计算
hash值: 使用spread()方法计算key的hash值。 - 定位节点: 使用
(n - 1) & hash计算节点在table中的位置。 - 如果节点为空: 如果节点为空，则返回
null。 - 如果节点是链表: 遍历链表，如果找到相同的
key，则返回value。 - 如果节点是红黑树: 调用
TreeBin.find()方法在红黑树中查找节点，如果找到，则返回value。
八、总结
ConcurrentHashMap通过CAS、synchronized和红黑树等技术，实现了高效的并发访问。其分段锁的设计大大减少了锁的竞争，提高了并发性能。红黑树的应用则保证了在链表过长的情况下，仍然能够保持较高的查找效率。理解ConcurrentHashMap的源码，有助于我们更好地掌握Java并发编程的精髓。
九、关于并发性能的考量
ConcurrentHashMap 在并发性能上做了很多优化，例如：
- 无锁读取: 大部分读取操作不需要加锁，利用
volatile保证可见性。 - 细粒度锁: 使用
synchronized锁住链表或红黑树的头节点，而非整个哈希表。 - CAS操作: 使用 CAS 进行原子更新，减少锁的使用。
- 分段扩容: 扩容过程是渐进式的，不会阻塞所有线程。
这些优化使得 ConcurrentHashMap 在高并发场景下具有卓越的性能。
十、红黑树的优势与必要性
使用红黑树的主要原因是平衡查找效率。当哈希冲突严重，链表过长时，链表的查找效率会退化到 O(n)。而红黑树是一种自平衡的二叉查找树，可以保证在最坏情况下查找、插入、删除的时间复杂度为 O(log n)。
因此，当链表长度超过一定阈值时，ConcurrentHashMap 会将链表转换为红黑树，以提高查找效率。这种设计兼顾了空间和时间效率，使得 ConcurrentHashMap 在各种场景下都能保持良好的性能。
十一、扩容机制中的细节
ConcurrentHashMap 的扩容是一个复杂的过程，涉及到多个线程的协作。扩容过程中，每个线程负责迁移一部分节点，并使用 ForwardingNode 标记正在迁移的节点。这种渐进式扩容的方式可以避免一次性扩容带来的性能瓶颈。
此外，ConcurrentHashMap 还使用了 sizeCtl 字段来控制扩容，保证只有一个线程可以发起扩容操作，并记录当前的扩容状态。
十二、高效并发的基石：对源码的理解
对 ConcurrentHashMap 源码的深入理解，能够帮助我们更好地掌握 Java 并发编程的核心技术。通过学习 ConcurrentHashMap 的设计思想，我们可以将其应用到自己的项目中，提高程序的并发性能和稳定性。