上一篇文章我讲了一下HashMap的相关源码实现,而且咱们知道它是线程不安全的,在并发环境中使用时,HashMap在扩容的时候有可能会生成一个环形链表,从而致使get造成死循环超时。那这篇咱们就来介绍一下并发环境下使用的HashMap——ConcurrentHashMap,下面是它的类关系图。java
Segment是一种可重入锁,在ConcurrentHashMap里扮演锁的角色;HashEntry则用于存储键值对数据。node
一个ConcurrentHashMap里包含一个Segment数组。Segment的结构和HashMap相似,是一种数组和链表结构。一个Segment里包含一个HashEntry数组,每个HashEntry是一个链表结构的元素,每一个Segment守护着一个HashEntry数组里的元素,当对HashEntry数组的数据进行修改时,必须首先得到与它对应的Segment锁。算法
ConcurrentHashMap经过使用分段锁技术,将数据分红一段一段的存储,而后给每一段数据配一把锁,当一个线程占用锁访问其中一个段数据的时候,其余段的数据也能被其余线程访问,可以实现真正的并发访问。数组
static final class Segment extends ReentrantLock implements Serializable {
private static final long serialVersionUID = 2249069246763182397L;
static final int MAX_SCAN_RETRIES =
Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;
transient volatile HashEntry[] table;
transient int count;
transient int modCount;
transient int threshold;
final float loadFactor;
... ...
}复制代码
static final class HashEntry<K,V> {
final int hash;
final K key;
volatile V value;
volatile HashEntry<K,V> next;
}
复制代码
static final class Segment<K,V> extends ReentrantLock implements Serializable {
private static final long serialVersionUID = 2249069246763182397L;
static final int MAX_SCAN_RETRIES =
Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;
transient volatile HashEntry<K,V>[] table;
transient int count;
transient int modCount;
transient int threshold;
final float loadFactor;
}
final Segment<K,V>[] segments;复制代码
public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel)复制代码
static final int DEFAULT_CONCURRENCY_LEVEL = 16;复制代码
接下来咱们看一下ConcurrentHashMap中的几个关键函数,get,put,rehash(扩容), size方法,看看他是如何实现并发的。
安全
segment的put方法实现bash
/** * The number of elements. Accessed only either within locks * or among other volatile reads that maintain visibility. */
transient int count;复制代码
static final int RETRIES_BEFORE_LOCK = 2;
public int size() {
// Try a few times to get accurate count. On failure due to
// continuous async changes in table, resort to locking.
final Segment<K,V>[] segments = this.segments;
int size;
boolean overflow; // true if size overflows 32 bits
long sum; // sum of modCounts
long last = 0L; // previous sum
int retries = -1; // first iteration isn't retry
try {
for (;;) {
// 超过尝试次数,则对每一个 Segment 加锁
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();
}
}
return overflow ? Integer.MAX_VALUE : size;
}复制代码
ConcurrentHashMap如何判断统计过程当中Segment的cout发生了变化?数据结构
数据结构采用数组 + 链表 + 红黑树的方式实现。当链表中(bucket)的节点个数超过8个时,会转换成红黑树的数据结构存储,这样设计的目的是为了提升同一个链表冲突过大状况下的读取效率。并发
static class Node<K,V> implements Map.Entry<K,V> {
final int hash;
final K key;
volatile V val;
volatile Node<K,V> next;
}复制代码
ConcurrentHashMap中有三个核心的CAS操做async
//获取索引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节点(将c和table[i]比较,相同则插入v)
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);
}
//利用volatile设置节点位置i的值,仅在上锁区被调用
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
} 复制代码
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
//若是一个线程发现sizeCtl<0,意味着另外的线程
//执行CAS操做成功,当前线程只须要让出cpu时间片
if ((sc = sizeCtl) < 0)
Thread.yield();
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
//CAS方法把sizectl置为-1,表示本线程正在进行初始化
try {
if ((tab = table) == null || tab.length == 0) {
//DEFAULT_CAPACITY 默认初始容量是 16
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
//初始化数组,长度为 16 或初始化时提供的长度
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
//将这个数组赋值给 table,table 是 volatile 的
table = tab = nt;
//若是 n 为 16 的话,那么这里 sc = 12
//其实就是 0.75 * n
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}复制代码
调用initTable会判断sizeCtl的值,若值为-1则表示正在初始化,会调用yield()去等待。函数
若值为0,这时先调用CAS算法去设置为-1,再初始化。
因此执行第一次put操做的线程会执行Unsafe.compareAndSwapInt方法修改sizeCtl为-1,有且只有一个线程可以修改为功,其它线程经过Thread.yield()让出CPU时间片等待table初始化完成。
综上所述,能够知道初始化是单线程操做。
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) {
//不容许key、value为空
if (key == null || value == null) throw new NullPointerException();
//返回 (h ^ (h >>> 16)) & HASH_BITS;
int hash = spread(key.hashCode());
int binCount = 0;
//循环,直到插入成功
for (Node[] tab = table;;) {
Node f; int n, i, fh;
if (tab == null || (n = tab.length) == 0)
//table为空,初始化table
tab = initTable();
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//索引处无值
if (casTabAt(tab, i, null,
new Node(hash, key, value, null)))
break; // no lock when adding to empty bin
}
else if ((fh = f.hash) == MOVED)// MOVED=-1;
//检测到正在扩容,则帮助其扩容
tab = helpTransfer(tab, f);
else {
V oldVal = null;
//上锁(hash值相同的链表的头节点)
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) {
//遍历链表节点
binCount = 1;
for (Node e = f;; ++binCount) {
K ek;
// hash和key相同,则修改value
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
//仅putIfAbsent()方法中onlyIfAbsent为true
if (!onlyIfAbsent)
//putIfAbsent()包含key则返回get,不然put并返回
e.val = value;
break;
}
Node pred = e;
//已遍历到链表尾部,直接插入
if ((e = e.next) == null) {
pred.next = new Node(hash, key,
value, null);
break;
}
}
}
else if (f instanceof TreeBin) {// 树节点
Node p;
binCount = 2;
if ((p = ((TreeBin)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
//判断是否要将链表转换为红黑树,临界值和HashMap同样也是8
if (binCount >= TREEIFY_THRESHOLD)
//若length<64,直接tryPresize,两倍table.length;不转树
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}复制代码
static final int spread(int h) {
return (h ^ (h >>> 16)) & HASH_BITS;
}
复制代码
int index = (n - 1) & hash复制代码
4. 若是f为null,说明table中这个位置第一次插入元素,利用Unsafe.compareAndSwapObject方法插入Node节点。
6. 其他状况把新的Node节点按链表或红黑树的方式插入到合适的位置,这个过程采用同步内置锁实现并发,代码如上。
在节点f上进行同步,节点插入以前,再次利用tabAt(tab, i) == f判断,防止被其它线程修改。
private final void treeifyBin(Node[] tab, int index) {
Node b; int n, sc;
if (tab != null) {
// MIN_TREEIFY_CAPACITY 为 64
// 因此,若是数组长度小于 64 的时候,其实也就是 32 或者 16 或者更小的时候,会进行数组扩容
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
// 后面咱们再详细分析这个方法
tryPresize(n << 1);
// b 是头结点
else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
// 加锁
synchronized (b) {
if (tabAt(tab, index) == b) {
// 下面就是遍历链表,创建一颗红黑树
TreeNode hd = null, tl = null;
for (Node e = b; e != null; e = e.next) {
TreeNode p =
new TreeNode(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(hd));
}
}
}
}
}复制代码
这里的扩容也是作翻倍扩容,扩容后数组容量为原来的 2 倍。
// 首先要说明的是,方法参数 size 传进来的时候就已经翻了倍了
private final void tryPresize(int size) {
// c:size 的 1.5 倍,再加 1,再往上取最近的 2 的 n 次方。
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;
// 这个 if 分支和以前说的初始化数组的代码基本上是同样的
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); // 0.75 * n
}
} finally {
sizeCtl = sc;
}
}
}
else if (c <= sc || n >= MAXIMUM_CAPACITY)
break;
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;
// 2. 用 CAS 将 sizeCtl 加 1,而后执行 transfer 方法
// 此时 nextTab 不为 null
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);
}
// 1. 将 sizeCtl 设置为 (rs << RESIZE_STAMP_SHIFT) + 2)
// 调用 transfer 方法,此时 nextTab 参数为 null
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
}
}
}复制代码
至于transfer()方法的源码这里我就不分析了,它的大概功能就是将原来的 tab 数组的元素迁移到新的 nextTab 数组中。
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) {//tabAt(i),获取索引i处Node
// 判断头结点是否就是咱们须要的节点
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
// 若是头结点的 hash<0,说明正在扩容,或者该位置是红黑树
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;
}复制代码
Node<K,V> find(int h, Object k) {
Node<K,V> e = this;
if (k != null) {
do {
K ek;
if (e.hash == h &&
((ek = e.key) == k || (ek != null && k.equals(ek))))
return e;
} while ((e = e.next) != null);
}
return null;
}复制代码
到这里我就基本把ConcurrentHashMap在 JDK 1.7和1.8中的实现大概捋了一遍,并详细分析了几个重要的方法实现:初始化、put、get。在 JDK1.8中ConcurrentHashMap发生了较大的变化,经过使用CAS+synchronized的实现取代了原先 1.7 中的Segment分段锁机制,从而支持更高的并发量。
这只是我对ConcurrentHashMap的第二次学习,若想更好地理解掌握ConcurrentHashMap的实现之精妙,我的以为还需之后再多看几回,相信每次都会有新的收获。