首页 > 嗟来之食 > 探索jdk8之ConcurrentHashMap 的实现机制 – 淮左 –
2016
07-28

探索jdk8之ConcurrentHashMap 的实现机制 – 淮左 –

在介绍ConcurrentHashMap源码之前,很有必要复习下java并发编程中的一些基础知识,比如内存模型等。
存储模型
并发编程中的三个概念
1、原子性
2、可见性
3、重排序
对HashMap在jdk8有所了解
对CAS有所了解
对内置锁和显示锁等有所了解
jdk8对ConcurrentHashMap做了很大的调整,首先因为HashMap在jdk8已经做了数据结构上的优化,增加了红黑树,详情可以参考我之前的博客。所以,jdk7针对ConcurrentHashMap的改进,主要是增加了分段锁Segment对HashEntity的控制,完美的解决了HashMap的安全问题,在JMM中有个名称叫安全发布,已经不适用了。那么,在jdk8如果保持性能的情况下对其进行修改了?它到底做了那些事情呢?
因为ConcurrentHashMap涉及的内容太多,jdk8有六千多行代码,jdk7才一两千行吧。所以我在想怎么一步步的对其进行剖解,最后我还是觉得按照程序的思路来吧,首先我们跑个ConcurrentHashMap的程序,然后进行调试,来一步步展开。
public static void main(String[] args) {
Map<String, String> cm = new ConcurrentHashMap<String, String>();
for (int i = 0; i < 14; i++) {
cm.put("key_" + i, "huaizuo_" + i);
}
}
首先初始化一个ConcurrentHashMap,因为我们是用默认构造函数,我们来看下初始化的一些重要的字段,去掉英文注释。
/**
* races. Updated via CAS.
* 记录容器的容量大小,通过CAS更新
*/
private transient volatile long baseCount;

/**
* 这个sizeCtl是volatile的,那么他是线程可见的,一个思考:它是所有修改都在CAS中进行,但是sizeCtl为什么不设计成LongAdder(jdk8出现的)类型呢?
* 或者设计成AtomicLong(在高并发的情况下比LongAdder低效),这样就能减少自己操作CAS了。
*
* 来看下注释,当sizeCtl小于0说明有多个线程正则等待扩容结果,参考transfer函数
*
* sizeCtl等于0是默认值,大于0是扩容的阀值
*/
private transient volatile int sizeCtl;

/**
* 自旋锁 (锁定通过 CAS) 在调整大小和/或创建 CounterCells 时使用。 在CounterCell类更新value中会使用,功能类似显示锁和内置锁,性能更好
* 在Striped64类也有应用
*/
private transient volatile int cellsBusy;
还有最重要的节点类Node,注意val和next是volatile类型
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;
}
接下来我们要把元素put到ConcurrentHashMap中了,那么我们来看下putVal的源码吧
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;
//这边加了一个循环,就是不断的尝试,因为在table的初始化和casTabAt用到了compareAndSwapInt、compareAndSwapObject
//因为如果其他线程正在修改tab,那么尝试就会失败,所以这边要加一个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) {//1
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))//2
break; // no lock when adding to empty bin
}
else if ((fh = f.hash) == MOVED)// a
tab = helpTransfer(tab, f);
else {
V oldVal = null;
//这个地方设计非常的巧妙,内置锁synchronized锁住了f,因为f是指定特定的tab[i]的,
// 所以就锁住了整行链表,这个设计跟分段锁有异曲同工之妙,只是其他读取操作需要用cas来保证
synchronized (f) {
if (tabAt(tab, i) == f) {//3
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;
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;
}
我们看到代码注释中的1、2、3我特定标注的,因为这些操作都是按照CAS的,其中关键部分已经做了注释,要正确取到真实数据需要知道变量所在的内存偏移量。
@SuppressWarnings("unchecked")
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);
}

/*
*但是这边为什么i要等于((long)i << ASHIFT) + ABASE呢,计算偏移量
*ASHIFT是指tab[i]中第i个元素在相对于数组第一个元素的偏移量,而ABASE就算第一数组的内存素的偏移地址
*所以呢,((long)i << ASHIFT) + ABASE就算i最后的地址
* 那么compareAndSwapObject的作用就算tab[i]和c比较,如果相等就tab[i]=v否则tab[i]=c;
*/
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);
}
关于sun.misc.Unsafe
// Unsafe mechanics
private static final sun.misc.Unsafe U;
private static final long SIZECTL;
private static final long TRANSFERINDEX;
private static final long BASECOUNT;
private static final long CELLSBUSY;
private static final long CELLVALUE;
private static final long ABASE;
private static final int ASHIFT;

static {
try {
U = sun.misc.Unsafe.getUnsafe();
Class<?> k = ConcurrentHashMap.class;
//获取ConcurrentHashMap这个对象字段sizeCtl在内存中的偏移量
SIZECTL = U.objectFieldOffset
(k.getDeclaredField("sizeCtl"));
TRANSFERINDEX = U.objectFieldOffset
(k.getDeclaredField("transferIndex"));
BASECOUNT = U.objectFieldOffset
(k.getDeclaredField("baseCount"));
CELLSBUSY = U.objectFieldOffset
(k.getDeclaredField("cellsBusy"));
Class<?> ck = CounterCell.class;
CELLVALUE = U.objectFieldOffset
(ck.getDeclaredField("value"));
Class<?> ak = Node[].class;
//可以获取数组第一个元素的偏移地址
ABASE = U.arrayBaseOffset(ak);
//arrayIndexScale可以获取数组的转换因子,也就是数组中元素的增量地址
//将arrayBaseOffset与arrayIndexScale配合使用,可以定位数组中每个元素在内存中的位置。
int scale = U.arrayIndexScale(ak);
if ((scale & (scale – 1)) != 0)
throw new Error("data type scale not a power of two");
ASHIFT = 31 – Integer.numberOfLeadingZeros(scale);
} catch (Exception e) {
throw new Error(e);
}
}
还是继续看put源码,看到//a注释,当(fh = f.hash) == MOVED,说明f.hash值为-1(MOVED为-1的final),那么如果hash什么时候回等于-1呢?为什么会有-1这种情况呢?这要涉及到ForwardingNode类
static final class ForwardingNode<K,V> extends Node<K,V> {
final Node<K,V>[] nextTable;
ForwardingNode(Node<K,V>[] tab) {
//MOVED 位-1,说明ForwardNode的节点的hash值为-1
super(MOVED, null, null, null);
this.nextTable = tab;
}
这个类是继承Node类的,他在初始化的时候hash值传了MOVED,我们知道ConcurrentHashMap在的数据结构是Table[]和链表组成,所以如果Table节点是ForwardNode节点的话那么Hash的值就等于-1,那么什么时候Node会变成ForwardNode呢?就是在扩容的时候,旧的Table的节点会临时用ForwardNode代替。待会会介绍。
我们还是继续一步步看代码,看inputVal的注释a,这个方法helpTransfer,如果线程进入到这边说明已经有其他线程正在做扩容操作,这个是一个辅助方法
/**
* Helps transfer if a resize is in progress.
*/
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);
while (nextTab == nextTable && table == tab &&
(sc = sizeCtl) < 0) {
//下面几种情况和addCount的方法一样,请参考addCount的备注
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;
}
这边如果table的某一个节点对应的链表超过一定的长度之后,就要把链表转化为红黑树的操作我就不详细的在这边文章介绍了,对于转化的操作其实和HashMap是一样的,但是这里涉及到并发,它其实也是通过synchronized和CAS来控制并发的。好了,当我们的putVal执行到addCount的时候
/**
* Adds to count, and if table is too small and not already
* resizing, initiates transfer. If already resizing, helps
* perform transfer if work is available. Rechecks occupancy
* after a transfer to see if another resize is already needed
* because resizings are lagging additions.
*
* @param x the count to add
* @param check if <0, don't check resize, if <= 1 only check if uncontended
*/
private final void addCount(long x, int check) {
CounterCell[] as; long b, s;

//U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x) 每次竟来都baseCount都加1因为x=1
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))) {
//多线程CAS发生失败的时候执行
fullAddCount(x, uncontended);//2
return;
}
if (check <= 1)
return;
s = sumCount();
}
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);
if (sc < 0) {//如果小于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);
}
//这个时候sizeCtl已经等于(rs << RESIZE_STAMP_SHIFT) + 2等于一个大的负数,这边加上2很巧妙,因为transfer后面对sizeCtl–操作的时候,最多只能减两次就结束
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
s = sumCount();
}
}
}
看上面注释1,每次都会对baseCount 加1,如果并发竞争太大,那么可能导致U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x) 失败,那么为了提高高并发的时候baseCount可见性失败的问题,又避免一直重试,这样性能会有很大的影响,那么在jdk8的时候是有引入一个类Striped64,其中LongAdder和DoubleAdder就是对这个类的实现。这两个方法都是为解决高并发场景而生的,是AtomicLong的加强版,AtomicLong在高并发场景性能会比LongAdder差。但是LongAdder的空间复杂度会高点。
// See LongAdder version for explanation
private final void fullAddCount(long x, boolean wasUncontended) {
int h;
//获取当前线程的probe值作为hash值,如果0则强制初始化当前线程的Probe值,初始化的probe值不为0
if ((h = ThreadLocalRandom.getProbe()) == 0) {
ThreadLocalRandom.localInit(); // force initialization
h = ThreadLocalRandom.getProbe();
wasUncontended = true;//设置未竞争标记为true
}
boolean collide = false; // True if last slot nonempty
for (;;) {
CounterCell[] as; CounterCell a; int n; long v;
if ((as = counterCells) != null && (n = as.length) > 0) {
if ((a = as[(n – 1) & h]) == null) {
if (cellsBusy == 0) { // Try to attach new Cell如果当前没有CounterCell就创建一个
CounterCell r = new CounterCell(x); // Optimistic create
if (cellsBusy == 0 &&
U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {//这边加上cellsBusy锁
boolean created = false;
try { // Recheck under lock
CounterCell[] rs; int m, j;
if ((rs = counterCells) != null &&
(m = rs.length) > 0 &&
rs[j = (m – 1) & h] == null) {
rs[j] = r;
created = true;
}
} finally {
cellsBusy = 0;//释放cellsBusy锁,让其他线程可以进来
}
if (created)
break;
continue; // Slot is now non-empty
}
}
collide = false;
}
else if (!wasUncontended) // CAS already known to fail wasUncontended为false说明已经发生了竞争,重置为true重新执行上面代码
wasUncontended = true; // Continue after rehash
else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))//对cell的value值进行累计x(1)
break;
else if (counterCells != as || n >= NCPU)
collide = false; // At max size or stale 表明as已经过时,说明cells已经初始化完成,看下面,重置collide为false表明已经存在竞争
else if (!collide)
collide = true;
else if (cellsBusy == 0 &&
U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
try {
if (counterCells == as) {// Expand table unless stale 下面的代码主要是给counterCells扩容,