C/C++教程
ConcurentHashMap设计与源码分析
本文主要是介绍ConcurentHashMap设计与源码分析，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！
简介
ConcurentHashMap是java.util.concurrent包下的一个线程安全的类，继承自Map类，用于存储具有键（key）、值（value）映射关系的双列集合。其数据结构与HashMap类似，都是使用数组+链表+树（红黑树）的结构实现。
优点
- 线程安全，在高并发情况下与HashTable相比效率更高（HashMap Vs. ConcurrentHashMap Vs. HashTable）
- 在使用Iterator迭代时不会抛出ConcurrentModificationException异常（fail-fast机制）
数据结构
ConcurentHashMap底层使用数组加链表的形式存储，K-V通过内部类Node包装，当链表长度大于8时，转换为树节点（TreeNode），超过64时改用红黑树（一种自平衡二叉查找树）
ConcurentHashMap数据结构的实现主要通过Node、TreeNode、TreeBin等内部类实现，其UML图如下：
Node
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;
}
}
Node类实现了Entry接口，用于存储节点的hash（哈希值）、key（键）、value（值）以及next（下一个节点的地址）四个属性。
TreeNode
TreeNode继承了Node类，用于存储ConcurentHashMap中的树结构，其构造方法如下：
TreeNode(int hash, K key, V val, Node<K,V> next,
TreeNode<K,V> parent) {
super(hash, key, val, next);
this.parent = parent;
}
在构造方法中，TreeNode调用了Node的构造方法，并指定了该节点的父节点。
TreeBin
TreeBin用于包装TreeNode类，当链表过长时，TreeBin会把TreeNode转换为红黑树。事实上，在ConcurentHashMap的“数组”中（也就是树的根节点）所存储的并不是TreeNode而是TreeBin。TreeBin不存储key/value，TreeBin还维护了一个读写锁，使得读必须等待写操作完成才能进行。
ForwardingNode
ForwardingNode用于标记正在迁移中的Node。在其构造方法会生成一个key、value 和 next 都为 null，且 hash 为 MOVED 的 Node。
static final class ForwardingNode<K,V> extends Node<K,V> {
final Node<K, V>[] nextTable;
ForwardingNode(Node<K, V>[] tab) {
super(MOVED, null, null, null);
this.nextTable = tab;
}
}
结构示意图
线程安全
1、CAS机制
对节点的修改操作都通过CAS来完成，CAS机制实现了无锁化的修改值的操作，可以大大降低锁代理的性能消耗。
2、volatile关键字
在ConcurentHashMap中，部分变量使用了volatile关键字修饰，保证了变量的可见性和指令的有序性。例如对节点操作进行控制的sizeCtl变量，在Node类中的val、next变量。
3、synchronized
ConcurentHashMap需要使用synchronized对数组中的的非空节点进行加锁操作（空节点可通过CAS直接进行操作，不需要加锁），如put方法及transfer方法。
4、Unsafe类和三个tabAt方法
Java是无法对操作系统底层进行操作的，所以CAS等操作的具体实现都需要Unsafe类以对底层进行操作。而对于节点的取值、设值、修改等操作，ConcurentHashMap基于Unsafe类封装了三个tabAt方法。
/**
* ((long)i << ASHIFT) + ABASE用于计算出元素的真实地址
* ASHIFT为每个节点（Node）的偏移量（位数）
* ABASE为头节点的地址（arrayBaseOffset）
*/
// 获得在i位置上的Node节点
static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}
// 利用CAS算法设置i位置上的Node节点
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);
}
// 设置节点位置的值
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}
方法详解
构造函数
ConcurentHashMap有多个重载的构造方法，可传入三个参数
- (int) initialCapacity 指ConcurrentHashMap的初始容量
- (float) loadFactor 加载因子
- (int) concurrencyLevel 并发度
在Java7中，ConcurentHashMap使用Segment分片的形式实现，Segment之间允许线程进行并发操作，而concurrencyLevel则是用来设置Segment[]数组长度的，concurrencyLevel的最小2次幂便为实际并发度。
而在Java8中，ConcurentHashMap摒弃了Segment，改用CAS加上TreeBin等辅助类实现，并发度concurrencyLevel也就没有实际意义了。
public ConcurrentHashMap(int initialCapacity) {
if (initialCapacity < 0)
throw new IllegalArgumentException();
// MAXIMUM_CAPACITY = 1 << 30（2^30 = 1073741824）
// 如果大小为MAXIMUM_CAPACITY最大总量的一半，那么直接将容量设为MAXIMUM_CAPACITY，否则计算最小幂次方
int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
MAXIMUM_CAPACITY :
// 1.5 * initialCapacity + 1
tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
this.sizeCtl = cap;
}
构造函数进行了sizeCtl的赋值，sizeCtl作为控制标识符，不同的数值代表不同的意义
- -1代表正在初始化
- -N表示有N-1个线程正在进行扩容操作
- hash表还没有被初始化时，该数值表示初始化或下一次进行扩容的大小。
- 初始化之后，它的值始终是当前ConcurrentHashMap容量的0.75倍
initTable初始化
初始化一个容量为sizeCtl的Node数组
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
// sizeCtl小于0，表示有其他线程正在进行初始化操作，把线程挂起
if ((sc = sizeCtl) < 0)
// 自旋
Thread.yield(); // lost initialization race; just spin
// 利用CAS方法把sizectl的值置为-1，表示本线程正在进行初始化，防止其他线程进入
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if ((tab = table) == null || tab.length == 0) {
// 默认大小16
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];// 初始化Node
table = tab = nt;
// 设置一个扩容的阈值 相当于0.75*n
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
互斥同步进入阻塞状态需要很大的开销，initTable方法使用了自旋锁，通过Thread.yield()使线程让步，然后忙循环直到sizeCtl满足条件
tableSizeFor函数详解
Returns a power of two table size for the given desired capacity.
返回大于输入参数且最近的2的整数次幂的数
/**
* 使最高位的1后面的位全变为1，最后再让结果n+1，即得到了2的整数次幂的值
*/
private static final int tableSizeFor(int c) {
int n = c - 1;
n |= n >>> 1;
n |= n >>> 2;
n |= n >>> 4;
n |= n >>> 8;
n |= n >>> 16;
return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
}
put
public V put(K key, V value) {
return putVal(key, value, false);
}
final V putVal(K key, V value, boolean onlyIfAbsent) {
// 判空，key和value不能为空
if (key == null || value == null) throw new NullPointerException();
// spread将较高的哈希值扩展为较低的哈希值，并将最高位强制为0
int hash = spread(key.hashCode());
// binCount用于记录相应链表的长度
int binCount = 0;
// 死循环
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
// tab为空，初始化table
if (tab == null || (n = tab.length) == 0)
tab = initTable();
// 根据hash值计算出在table里面的位置，若该位置的值为空，直接放入元素
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
}
// 存在节点，说明发生了hash碰撞，需要对表进行扩容
// 如果该位置的节点存在值且为MOVED(-1)，说明正在扩容
else if ((fh = f.hash) == MOVED)
// helpTransfer方法用于增加线程以协助扩容
tab = helpTransfer(tab, f);
else {
V oldVal = null;
// 节点上锁
synchronized (f) {
if (tabAt(tab, i) == f) {
// fh > 0 说明这个节点是一个链表的节点 不是树的节点
if (fh >= 0) {
binCount = 1;
// 遍历节点
for (Node<K,V> e = f;; ++binCount) {
K ek;
// 存在该key，替换其值
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;
// 不存在该key，插入新Node
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;
binCount = 2;
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
// TREEIFY_THRESHOLD = 8
// 链表长度大于8，转换为树节点
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
// 将当前ConcurrentHashMap的元素数量+1
addCount(1L, binCount);
return null;
}
putVal()方法首先获取到key的hash值，该key所对应位置的值为空，直接放入，若该键对应的位置存在节点，则判断是否该节点为链表节点还是树节点，再使用其相应的方法将Node放入
putVal()方法大概的流程图如下：
get方法
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
// 获取hash值
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;
}
// TreeBin或ForwardingNode，调用find方法
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;
}
treeifyBin
Replaces all linked nodes in bin at given index unless table is too small, in which case resizes instead.
将所有索引处的链表节点替换为二叉树，如果表太小则改为调整大小
private final void treeifyBin(Node<K,V>[] tab, int index) {
Node<K,V> b; int n, sc;
if (tab != null) {
// 链表tab的长度小于64（MIN_TREEIFY_CAPACITY=64）时，进行扩容
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
// 调用tryPresize进行扩容（所传参数n即链表长度 * 2）
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 ((p.prev = tl) == null)
hd = p;
else
tl.next = p;
tl = p;
}
setTabAt(tab, index, new TreeBin<K,V>(hd));
}
}
}
}
}
tryPresize
Tries to presize table to accommodate the given number of elements.
尝试调整表的大小以适应给定的元素数量。
private final void tryPresize(int size) {
int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
tableSizeFor(size + (size >>> 1) + 1);
int sc;
// 正在扩容
while ((sc = sizeCtl) >= 0) {
Node<K,V>[] tab = table; int n;
// tab未初始化，进行初始化，与initTable类似
if (tab == null || (n = tab.length) == 0) {
n = (sc > c) ? sc : c;
if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if (table == tab) {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = nt;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
}
}
// 如果扩容大小没有达到阈值，或者超过最大容量
else if (c <= sc || n >= MAXIMUM_CAPACITY)
break;
// 调用transfer()扩容
else if (tab == table) {
int rs = resizeStamp(n);
if (sc < 0) {
Node<K,V>[] nt;
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);
}
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
}
}
}
transfer
Moves and/or copies the nodes in each bin to new table.
将树中的节点移动和复制到新表中
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
// NCPU为可用的CPU线程数
// stride可以理解为步长，最小值为16
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
// nextTab为null，进行初始化
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 = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
// transferIndex用于控制迁移的位置
transferIndex = n;
}
int nextn = nextTab.length;
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
// advance为true说明这个节点已经处理过
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
// i 是位置索引，bound 是边界，从后往前移
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
// 此处的while循环用作遍历hash原表中的节点
while (advance) {
int nextIndex, nextBound;
// 处理从bound到i的节点
if (--i >= bound || finishing)
advance = false;
// transferIndex 小于等于 0，说明原hash表的所有位置都有相应的线程去处理了
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
// i指向transferIndex，bound指向(transferIndex-stride)
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;
// 迁移完成，nextTab赋值给table
if (finishing) {
nextTable = null;
table = nextTab;
// 重新计算sizeCtl，得出的值是新数组长度的 0.75 倍
sizeCtl = (n << 1) - (n >>> 1);
return;
}
// 任务完成，使用CAS将sizeCtl减1
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
}
}
// 位置 i 为空，放入空节点fwd
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
// 节点为ForwardingNode，表示已迁移
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
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);
}
// 在i位置放入一个链表
setTabAt(nextTab, i, ln);
// 在i+n位置放入另一个链表
setTabAt(nextTab, i + n, hn);
// 在table的i位置上插入forwardNode节点， 表示已经处理过该节点
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;
}
}
// 小于UNTREEIFY_THRESHOLD(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;
}
}
}
}
}
}
还有些事
1、为什么要取最小二次幂
因为HashMap通过对hash值的i = (n - 1) & hash运算实现均匀分布，若n不为2的次幂数，就不能保证均匀分布。
参考文章
1、ConcurrentHashMap总结
2、Java7/8 中的 HashMap 和 ConcurrentHashMap 全解析
3、jdk1.8的HashMap和ConcurrentHashMap
这篇关于ConcurentHashMap设计与源码分析的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！
您可能喜欢
-
如何获取 OpenAI API Key 用于ChatGPT AI大模型开发？11-26
-
MATLAB 中 A(7)=[];什么意思？-icode9专业技术文章分享11-26
-
UniApp 中如何实现使用输入法时保持页面列表不动的效果？-icode9专业技术文章分享11-26
-
在 UniApp 中怎么实现输入法弹出时禁止页面向上滚动？-icode9专业技术文章分享11-26
-
WebSocket是什么，怎么使用？-icode9专业技术文章分享11-26
-
页面有多个ref 要动态传入怎么实现？-icode9专业技术文章分享11-26
-
在 UniApp 中实现一个底部输入框的常见方法有哪些？-icode9专业技术文章分享11-26
-
RocketMQ入门指南：搭建与使用全流程详解11-26
-
RocketMQ入门教程：轻松搭建与使用指南11-26
-
手写RocketMQ：从入门到实践的简单教程11-26
-
【机器学习(二)】分类和回归任务-决策树(Decision Tree,DT)算法-Sentosa_DSML社区版11-25
-
增量更新怎么做？-icode9专业技术文章分享11-23
-
压缩包加密方案有哪些？-icode9专业技术文章分享11-23
-
用shell怎么写一个开机时自动同步远程仓库的代码？-icode9专业技术文章分享11-23
-
webman可以同步自己的仓库吗？-icode9专业技术文章分享11-23
栏目导航
