20201224
程序员文章站
2022-05-12 11:34:19
...
JAVA8源码系列-ConcurrentHashMap
第一篇： JAVA8源码系列-ConcurrentHashMap文章目录
一、ConcurrentHashMap简单介绍
ConcurrentHashMap是HashMap的并发版本，在java8的实现中通过synchronized关键字对每个桶加锁来实现的，以下主要分析ConcurrentHashMap的put方法和get方法。
二、ConcurrentHashMap#put()方法
put方法主要调用putVal方法
public V put(K key, V value) {
return putVal(key, value, false);
}
接下来看putVal方法，该方法比较长，对关键代码加了注释
final V putVal(K key, V value, boolean onlyIfAbsent) {
//键值都不为null
if (key == null || value == null) throw new NullPointerException();
//计算hash值，通过高16位和低16取异或并和 0x7fffffff 做 与 运算。
int hash = spread(key.hashCode());
int binCount = 0;
//for死循环中进行自旋
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) {
//通过n-1 & hash代替取模运算，通过tabAt方法发现该桶为空，尝试通过CAS插入
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
//通过CAS添加该桶第一个元素成功，跳出自旋
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判断的目的是为了防止当前线程被阻塞到获取f锁期间，扩容的线程改变了当前桶的头节点
if (tabAt(tab, i) == f) {
if (fh >= 0) {
//头节点hash值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)))) {
//若遇到key相同的节点，则跳出自旋
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
Node<K,V> pred = e;
if ((e = e.next) == null) {
//如果到了链表末尾仍然没有找到相等的，则把当前值包装一个Node插入链表末尾，跳出自旋
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) {
if (binCount >= TREEIFY_THRESHOLD)
//达到树化阈值，开始将链表转为树
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
//在此方法里面会根据情况调用扩容相关代码
addCount(1L, binCount);
return null;
}
初始化桶的代码如下：
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
//多线程环境，通过自旋+CAS操作，保证桶初始化一定正确且成功
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 {
//重新判空，因为在下面的finally语句块中会有sizeCtl = sc，假设初始化sizeCtl为1，如果不判空则可能会重新生成桶
//具体流程为：假设线程t1执行到else if的地方时间片耗尽，线程t2执行CAS操作成功进入else if，执行完毕之后t2在finally中将sizeCtl重新赋值为1，线程t1在此时抢到时间片继续执行CAS操作，可以成功，如果此时不进行判空则会重复初始化导致线程安全问题。
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);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
三、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)
//头节点hash值小于0，要么是树，要么是正在扩容且该桶中已经迁移完，要么是RESERVED
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;
}
四、ConcurrentHashMap#size() 方法
size方法主要调用sumCount方法
public int size() {
long n = sumCount();
return ((n < 0L) ? 0 :
(n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
(int)n);
}
sumCount()中通过加和counterCells和baseCount的值，counterCells是一个CounterCell的数据，CounterCell中只有一个value属性。
在没有并发的情况下，新增额的时候给baseCount加1，当并发的时候，CAS 修改 baseCount 失败后，就创建一个CounterCell对象，对象的value 属性是 1。在计算 size 的时候，会将 baseCount 和 CounterCell 数组中的元素的 value 累加，得到总的大小，但这个数字仍旧可能是不准确的。
final long sumCount() {
CounterCell[] as = counterCells; CounterCell a;
long sum = baseCount;
if (as != null) {
for (int i = 0; i < as.length; ++i) {
if ((a = as[i]) != null)
sum += a.value;
}
}
return sum;
}
五、扩容机制
扩容的时候通过sizeCtl控制，低16为保存当前正在执行扩容的线程数量，扩容的时候，每个线程通过CAS操作修改transferIndex占有桶的一段进行扩容。
扩容之后一个节点的下标要么是扩容前的下标，要么是扩容前的下标+扩容前的长度，这是因为每次扩容都是按照2的倍数扩容的。利用这个性质我们在扩容的时候
关于扩容最后一个线程的还要全部扫描一遍的问题，这篇博客给出了回答，同时还有ConcurrentHashMap的作者Doug Lea的邮件回复https://blog.csdn.net/anlian523/article/details/107328200
http://cs.oswego.edu/pipermail/concurrency-interest/2020-July/017171.html
答案就是这是一个遗留bug，现在已经不需要扫描了。
总结
ConcurrentHashMap还是比较复杂的，其他方法之后有时间再分析。
推荐阅读