Java1.8ConcurrnentHashMap源码分析
参考文献:http://cmsblogs.com/?p=2283
https://www.jianshu.com/p/e694f1e868ec
https://blog.****.net/panweiwei1994/article/details/78897275
JDK1.7
jdk1.7中采用Segment
+ HashEntry
的方式进行实现,结构如下:
ConcurrentHashMap
初始化时,计算出Segment
数组的大小ssize
和每个Segment
中HashEntry
数组的大小cap
,并初始化Segment
数组的第一个元素;其中ssize
大小为2的幂次方,默认为16,cap
大小也是2的幂次方,最小值为2,ConcurrentHashMap在Java1.7中通过“锁分段”来实现线程安全。ConcurrentHashMap将哈希表分成许多片段(segments),每一个片段(table)都类似于HashMap,它有一个HashEntry数组,数组的每项又是HashEntry组成的链表。每个片段都是Segment类型的,Segment继承了ReentrantLock,所以Segment本质上是一个可重入的互斥锁。这样每个片段都有了一个锁,这就是“锁分段”。线程如想访问某一key-value键值对,需要先获取键值对所在的segment的锁,获取锁后,其他线程就不能访问此segment了,但可以访问其他的segment。
put实现
当执行put
方法插入数据时,根据key的hash值,在Segment
数组中找到相应的位置,如果相应位置的Segment
还未初始化,则通过CAS进行赋值,接着执行Segment
对象的put
方法通过加锁机制插入数据,实现如下:
场景:线程A和线程B同时执行相同Segment
对象的put
方法
1、线程A执行tryLock()
方法成功获取锁,则把HashEntry
对象插入到相应的位置;
2、线程B获取锁失败,则执行scanAndLockForPut()
方法,在scanAndLockForPut
方法中,会通过重复执行tryLock()
方法尝试获取锁,在多处理器环境下,重复次数为64,单处理器重复次数为1,当执行tryLock()
方法的次数超过上限时,则执行lock()
方法挂起线程B;
3、当线程A执行完插入操作时,会通过unlock()
方法释放锁,接着唤醒线程B继续执行;
put操作的具体流程
1、将当前 Segment 中的 table 通过 key 的 hashcode 定位到 HashEntry。
2、遍历该 HashEntry,如果不为空则判断传入的 key 和当前遍历的 key 是否相等,相等则覆盖旧的 value。
3、不为空则需要新建一个 HashEntry 并加入到 Segment 中,同时会先判断是否需要扩容。
4、最后会解除在 1 中所获取当前 Segment 的锁。
get操作的具体流程
只需要将 Key 通过 Hash 之后定位到具体的 Segment ,再通过一次 Hash 定位到具体的元素上。
由于 HashEntry 中的 value 属性是用 volatile 关键词修饰的,保证了内存可见性,所以每次获取时都是最新值。
ConcurrentHashMap 的 get 方法是非常高效的,因为整个过程都不需要加锁。
因为concurrnenthashmap可以多线程来操作,所以他的size大小时实时变化的,一般的思路是统计每个Segment
对象中的元素个数,然后进行累加,但是这种方式计算出来的结果并不一样的准确的,因为在计算后面几个Segment
的元素个数时,已经计算过的Segment
同时可能有数据的插入或则删除,在1.7的实现中,采用了如下方式:
try {
for (;;) {
if (retries++ == RETRIES_BEFORE_LOCK) {
for (int j = 0; j < segments.length; ++j)
ensureSegment(j).lock(); // force creation
}
sum = 0L;
size = 0;
overflow = false;
for (int j = 0; j < segments.length; ++j) {
Segment<K,V> seg = segmentAt(segments, j);
if (seg != null) {
sum += seg.modCount;
int c = seg.count;
if (c < 0 || (size += c) < 0)
overflow = true;
}
}
if (sum == last)
break;
last = sum;
}
} finally {
if (retries > RETRIES_BEFORE_LOCK) {
for (int j = 0; j < segments.length; ++j)
segmentAt(segments, j).unlock();
}
}
先采用不加锁的方式,连续计算元素的个数,最多计算3次:
1、如果前后两次计算结果相同,则说明计算出来的元素个数是准确的;
2、如果前后两次计算结果都不同,则给每个Segment
进行加锁,再计算一次元素的个数;
JDK1.8
1.8中放弃了Segment
臃肿的设计,取而代之的是采用Node
+ CAS
+ Synchronized
来保证并发安全进行实现,结构如下:
ConcurrentHashMap定义了如下几个常量:
// 最大容量:2^30=1073741824
private static final int MAXIMUM_CAPACITY = 1 << 30;
// 默认初始值,必须是2的幕数
private static final int DEFAULT_CAPACITY = 16;
//
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
//
private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
//
private static final float LOAD_FACTOR = 0.75f;
// 链表转红黑树阀值,> 8 链表转换为红黑树
static final int TREEIFY_THRESHOLD = 8;
//树转链表阀值,小于等于6(tranfer时,lc、hc=0两个计数器分别++记录原bin、新binTreeNode数量,<=UNTREEIFY_THRESHOLD 则untreeify(lo))
static final int UNTREEIFY_THRESHOLD = 6;
//
static final int MIN_TREEIFY_CAPACITY = 64;
//
private static final int MIN_TRANSFER_STRIDE = 16;
//
private static int RESIZE_STAMP_BITS = 16;
// 2^15-1,help resize的最大线程数
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
// 32-16=16,sizeCtl中记录size大小的偏移量
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
// forwarding nodes的hash值
static final int MOVED = -1;
// 树根节点的hash值
static final int TREEBIN = -2;
// ReservationNode的hash值
static final int RESERVED = -3;
// 可用处理器数量
static final int NCPU = Runtime.getRuntime().availableProcessors();
- table:用来存放Node节点数据的,默认为null,默认大小为16的数组,每次扩容时大小总是2的幂次方;
- nextTable:扩容时新生成的数据,数组为table的两倍;
- Node:节点,保存key-value的数据结构;
- ForwardingNode:一个特殊的Node节点,hash值为-1,其中存储nextTable的引用。只有table发生扩容的时候,ForwardingNode才会发挥作用,作为一个占位符放在table中表示当前节点为null或则已经被移动
-
sizeCtl:控制标识符,用来控制table初始化和扩容操作的,在不同地方有不同的用途,其值也不同,所代表的含义也不同
- 负数代表正在进行初始化或扩容操作
- -1代表正在初始化
- -N 表示有N-1个线程正在进行扩容操作
- 正数或0代表hash表还没有被初始化,这个数值表示初始化或下一次进行扩容的大小
下面看一下concurrnenthashmap的get()方法
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
//计算key的哈希值
int h = spread(key.hashCode());
//如果表不为空,表长度大于0,key所在的桶的头结点不为null
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
if ((eh = e.hash) == h) {//如果查到的桶的头结点的key哈希值与参数key的哈希值相同
if ((ek = e.key) == key || (ek != null && key.equals(ek)))//如果查到的桶的头结点的key参数key相等,返回桶的头结点的value
return e.val;
}
else if (eh < 0)//如果查到的桶的头结点的key哈希值小于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;
}
}
//如果都没找到,返回null
return null;
}
可以将步骤总结如下:
- 通过key计算哈希值
- 通过哈希值找到桶
- 通过哈希值和桶来查找节点
3.1. 以此判断桶的头结点是不是要找的节点
3.2. 如果不是,判断桶的头节点的哈希值是否小于0,如果是则说明要找的节点在树上
3.3. 如果以上两个条件都不满足,则说明要找的节点在链表上,遍历链表,查找节点 - 如果通过以上步骤找到了节点,返回节点的value。没找到,就返回null。
从源码中可以看出,上面的步骤并没有加锁。
concurrnenthashmap的put()方法
public V put(K key, V value) {
return putVal(key, value, false);
}
/** Implementation for put and putIfAbsent */
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;
//如果table没有初始化,初始化table
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//根据哈希值计算在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
}
//如果要插入的位置是一个forwordingNode节点,表示正在扩容,那么当前线程帮助扩容
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
//进行到这一步,说明要插入的位置有值,需要加锁
synchronized (f) {
//确定f是tab中的头节点
if (tabAt(tab, i) == f) {
//如果头结点的哈希值大于等于0,说明要插入的节点在链表中
if (fh >= 0) {
binCount = 1;
//遍历链表中的所有节点
for (Node<K,V> e = f;; ++binCount) {
K ek;
//如果某一节点的key哈希值和key与参数相等,替换节点的value
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对应的节点,根据参数新建节点,插入链表尾部
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;
}
}
}
}
//如果binCount不为0,说明插入或者替换操作完成了
if (binCount != 0) {
//判断节点数量是否大于8,如果大于就需要把链表转化成红黑树
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;//如果在链表中找到了指定key的节点,返回被替换的value
break;
}
}
}
//能执行到这一步,说明节点不是被替换的,是被插入的,所以要将map的元素数量加1
addCount(1L, binCount);
return null;
}
可以将步骤总结如下:
- 计算key哈希值
- 根据哈希值计算在table中的位置
- 根据哈希值执行插入或替换操作
3.1 如果这个位置没有值,直接将键值对放进去,不需要加锁。
3.2 如果要插入的位置是一个forwordingNode节点,表示正在扩容,那么当前线程帮助扩容
3.3 加锁。以下操作都需要加锁。
3.4 如果要插入的节点在链表中,遍历链表中的所有节点,如果某一节点的key哈希值和key与参数相等,替换节点的value,记录被替换的值;如果遍历到了最后一个节点,还没找到key对应的节点,根据参数新建节点,插入链表尾部。
3.5 如果要插入的节点在树中,则按照树的方式插入或替换节点。如果是替换操作,记录被替换的值 - 判断节点数量是否大于8,如果大于就需要把链表转化成红黑树
- 如果操作3中执行的是替换操作,返回被替换的value。程序结束。
- 能执行到这一步,说明节点不是被替换的,是被插入的,所以要调用addCount方法,该方法主要做两件事,一件更新baseCount的值,第二件检测是否进行扩容
当ConcurrentHashMap中table元素个数达到了容量阈值(sizeCtl)时,则需要进行扩容操作。
transfer()方法为ConcurrentHashMap扩容操作的核心方法。由于ConcurrentHashMap支持多线程扩容,而且也没有进行加锁,所以实现会变得有点儿复杂。整个扩容操作分为两步:
- 构建一个nextTable,其大小为原来大小的两倍,这个步骤是在单线程环境下完成的
- 将原来table里面的内容复制到nextTable中,这个步骤是允许多线程操作的,所以性能得到提升,减少了扩容的时间消耗
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
// 每核处理的量小于16,则强制赋值16
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]; //构建一个nextTable对象,其容量为原来容量的两倍
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
transferIndex = n;
}
int nextn = nextTab.length;
// 连接点指针,用于标志位(fwd的hash值为-1,fwd.nextTable=nextTab)
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
// 当advance == true时,表明该节点已经处理过了
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
// 控制 --i ,遍历原hash表中的节点
while (advance) {
int nextIndex, nextBound;
if (--i >= bound || finishing)
advance = false;
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
// 用CAS计算得到的transferIndex
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; // table 指向nextTable
sizeCtl = (n << 1) - (n >>> 1); // sizeCtl阈值为原来的1.5倍
return; // 跳出死循环,
}
// CAS 更扩容阈值,在这里面sizectl值减一,说明新加入一个线程参与到扩容操作
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
}
}
// 遍历的节点为null,则放入到ForwardingNode 指针节点
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
// f.hash == -1 表示遍历到了ForwardingNode节点,意味着该节点已经处理过了
// 这里是控制并发扩容的核心
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {
// 节点加锁
synchronized (f) {
// 节点复制工作
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
// fh >= 0 ,表示为链表节点
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);
}
// 在nextTable i 位置处插上链表
setTabAt(nextTab, i, ln);
// 在nextTable i + n 位置处插上链表
setTabAt(nextTab, i + n, hn);
// 在table i 位置处插上ForwardingNode 表示该节点已经处理过了
setTabAt(tab, i, fwd);
// advance = true 可以执行--i动作,遍历节点
advance = true;
}
// 如果是TreeBin,则按照红黑树进行处理,处理逻辑与上面一致
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;
}
}
// 扩容后树节点个数若<=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;
}
}
}
}
}
}
上面的源码有点儿长,稍微复杂了一些,在这里我们抛弃它多线程环境,我们从单线程角度来看:
- 为每个内核分任务,并保证其不小于16
- 检查nextTable是否为null,如果是,则初始化nextTable,使其容量为table的两倍
- 死循环遍历节点,知道finished:节点从table复制到nextTable中,支持并发,请思路如下:
- 如果节点 f 为null,则插入ForwardingNode(采用Unsafe.compareAndSwapObjectf方法实现),这个是触发并发扩容的关键
- 如果f为链表的头节点(fh >= 0),则先构造一个反序链表,然后把他们分别放在nextTable的i和i + n位置,并将ForwardingNode 插入原节点位置,代表已经处理过了
- 如果f为TreeBin节点,同样也是构造一个反序 ,同时需要判断是否需要进行unTreeify()操作,并把处理的结果分别插入到nextTable的i 和i+nw位置,并插入ForwardingNode 节点
- 所有节点复制完成后,则将table指向nextTable,同时更新sizeCtl = nextTable的0.75倍,完成扩容过程
在多线程环境下,ConcurrentHashMap用两点来保证正确性:ForwardingNode和synchronized。当一个线程遍历到的节点如果是ForwardingNode,则继续往后遍历,如果不是,则将该节点加锁,防止其他线程进入,完成后设置ForwardingNode节点,以便要其他线程可以看到该节点已经处理过了,如此交叉进行,高效而又安全。