JDK1.8 concurrentHashMap设计原理与优化

前面写到了JDK1.8 hashMap优化，那么在jdk1.8中对concurrentHashMap也做了不少的优化，本文也会从源码来分析下这些优化，以及为什么要这么做。首先还是提出需要关注的点:

  • 为什么使用CAS+synchronized替换了segemnt+ReentrantLock来保证线程安全

  • concurrentHashMap为什么不允许key和value为null

  • 如何高效实现在多线程情况下扩容

  • 黑科技@sun.misc.Contended

存储结构

jdk1.8concurrentHashMap的存储结构基本和hashMap是一样的了，就是单纯的数组链表；不再是1.7里面的segments数组，每一个segment相当于是一个map，而且每一个segment都实现ReentrantLock。

显然对比起现在的实现，在内存上是有浪费的。

找到真正的桶需要两次hash

如下图所示：

synchronized锁优化

提起synchronized，可能大家都有个印象是这个锁是重量级的，而且性能在线程竞争比较激烈情况下比reentrantLock差。例如下图是JDK1.5对比,确实是这样。

但是这已经是过往了，synchronized作为java语法层面的同步，java虚拟机是可以在线程和对象元数据中记录synchronized中锁的信息的，开发团队是投入了大量的精力进行优化的。

synchronized与reentrantLock对比

前面提到了可以在对象的元数据里面存储synchronized信息，在对象头里面有个“Mark word”，正是这个Mark word使对象可以表示：未锁定；轻量级锁；重量级锁；偏向锁几种状态。而且还会有自适应性自旋的优化。

回到concurrentHashMap的put方法中,可以发现，当确定了元素要放入的桶之后，如果桶里面没有节点，要么不用加锁，直接CAS尝试将该元素作为头结点；如果里面有了其他的元素，这时才会对头结点f进行加锁，锁住这个链表。也就是说每个桶都是单独的锁，这样不同线程并发争抢到同一个桶的概率已经很小，而且由于自适应自旋机制，就算发生了争抢，也不会将线程挂起，自旋等待即可。实际上这已经是乐观锁的设计思路了。

  • reentrantLock本质上是一个悲观锁，就算在非公平模式下，会尝试一次CAS获取锁，如果获取不到，那么会被添加到AQS等待队列再尝试一次，还是获取不到的就会被阻塞住，顺序等待被唤醒。

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();
elseif ((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
}
elseif ((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;
}
}
}
elseif (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;
}

高效扩容

扩容是个很消耗性能的点，之前在hashMap中也提到过，最好能够预估需要存入的数据数量，初始化就设定号，尽量避免频繁的扩容，特别是在多线程环境中，更加需要小心谨慎，加上各种“骚操作”才能提升效率，能从根本上避免就最好。

private final void addCount(long x, int check) {
CounterCell[] as; long b, s;
//利用CAS方法更新 baseCount 的值
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {// 1
CounterCell a; long v; int m;
boolean uncontended = true;
if (as == null || (m = as.length - 1) < 0 ||
(a = as[ThreadLocalRandom.getProbe() & m]) == null ||
!(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
// 多线程修改baseCount时，竞争失败的线程会执行fullAddCount(x, uncontended),把x的值插入到counterCell类中
fullAddCount(x, uncontended); // 2
return;
}
if (check <= 1)//这里添加之后，不是树就不用操作了
return;
s = sumCount();
}
//如果check值大于等于0 则需要检验是否需要进行扩容操作
if (check >= 0) {
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);
// 如果小于0 说明已经有线程在进行扩容了
if (sc < 0) {
// 以下的情况说明已经有在扩容或者多线程进行了扩容，其他线程直接 break 不要进入扩容
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);
}
// 当前线程是唯一的或是第一个发起扩容的线程  此时nextTable=null
elseif (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
s = sumCount();
}
}
}
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
Node<K,V>[] nextTab;int sc;
//条件：原结点不为空,这个结点是协助结点（ForwardingNode）并且不处于最后
if (tab != null && (f instanceof ForwardingNode) &&
(nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
int rs = resizeStamp(tab.length);
//sizeCtl是负数，标明还在扩容；tab和现在占用的table不同，说明扩容还在进行当中
while (nextTab == nextTable && table == tab && (sc = sizeCtl) < 0) {
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || transferIndex <= 0)
break;
//对现在的表长度做操作，如果没有改动，说明其他进程没有在扩容
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}

@sun.misc.Contended

在源码里面CounterCell是被Contended注解的，表示消除伪共享，因为这个值表示桶里面的元素个数，是经常改变的，也就是在缓存中是经常要被置为无效的，如果有其他的变量与其在同一个缓存行，那么会相互影响，降低并发量。

    @sun.misc.Contended static final class CounterCell {
volatile long value;
CounterCell(long x) { value = x; }
}

key value不允许为null

hashMap中是允许key或者value为null的，那么为什么concurrentHashMap不允许呢？其实作者Doug Lea大神亲自解释过：也就是说当get(key)为null，在concurrentHashMap中难以分辨出是不存在这个key对应的value还是本身key对应的value就是null。

以上是 JDK1.8 concurrentHashMap设计原理与优化 的全部内容， 来源链接： utcz.com/a/27568.html

回到顶部