源码分析之ConcurrentHashMap

Doug Lea大神在j.u.c包下给我们提供了一个适用于多线程并发环境使用的集合类ConcurrentHashMap。而如果在多线程环境,不考虑任何线程安全的防范的话,使用HashMap会带来诸多问题。

HashMap的并发问题

HashMap集合是非线程安全,在多线程环境下容易出现问题。HashMap在数据更新的时候,会带来很多问题:

  • 数据丢失

多线程环境下,如两个线程同时在一个bucketput元素,有可能会造成元素写入被覆盖,从而丢失数据。

  • fail-fast机制

当多个线程对同一个集合进行操作时候,就会触发fail-fast机制并抛出ConcurrentModificationException

  • 死循环

HashMaptable的大小不够时,如果两个线程同时进行resize的操作。如果某一bucket下元素是用链表记录,在resize过程中,链表在多线程的环境下有可能会形成闭合回链,get请求就会造成死循环,使得CPU飙升。详细可以看看这篇博文:疫苗:JAVA HASHMAP的死循环。不过JDK1.8中已经改写resize方法,应该不会出现这种问题,但这并不是我们可以在多线程环境下使用HashMap理由。

线程安全类 HashTable

HashTable是线程安全的,底层是通过synchronized来保证线程安全。当多线程竞争激烈的时候,没有获得锁的线程都将会阻塞。synchronized修饰所有针对HashTable集合的操作。这样一旦有线程获得锁,其他的线程都只能等待锁的释放,然后再去竞争锁,这样一来HashTable的效率必定会受到影响。

ConcurrentHashMap 锁机制

相对于低效的HashTableConcurrentHashMap在锁机制层面上做了优化。锁优化的思路一般有一下几种:

  • 减少锁持有时间
  • 减小锁粒度
  • 锁分离
  • 锁粗化
  • 锁消除

JDK1.6 ConcurrentHashMap

ConcurrentHashMap中存储的元素是通过静态内部类HashEntry封装实现的。

1
2
3
4
5
6
7
8
9
10
11
12
static final class HashEntry<K,V> {
final K key;
final int hash;
volatile V value;
final HashEntry<K,V> next;
HashEntry(K key, int hash, HashEntry<K,V> next, V value) {
this.key = key;
this.hash = hash;
this.next = next;
this.value = value;
}
}

其中value 字段被声明为 volatile 型,保证其在内存中可见性。keyhashnext 都被声明为 final 型。ConcurrentHashMap存储数据根据keyhash值将数据元素散列到哈希表中每一个bucket中。当发生哈希碰撞时候,会将元素封装成HashEntry构成链表。由于nextfinal类型的,链表中添加元素都将从表头添加。

减小锁粒度-分段锁

JDK1.7中,ConcurrentHashMap在锁优化过程通过减小锁粒度的实现了对集合的高效并发操作。ConcurrentHashMap包含一个静态内部类Segment,是用来充当锁的角色。

每个Segment将维护若干个bucket。而锁只针对Segment而不是整张表,从而实现减小锁的粒度。

Segment类
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
static final class Segment<K,V> extends ReentrantLock implements Serializable {
// 包含HashEntry的个数
transient volatile int count;
// table 更新次数
transient int modCount;
// table resize 阈值
transient int threshold;
// HashEntry数组用于存储元素
transient volatile HashEntry<K,V>[] table;
// 加载因子
final float loadFactor;
// 构造函数
Segment(int initialCapacity, float lf) {
loadFactor = lf;
setTable(HashEntry.<K,V>newArray(initialCapacity));
}
void setTable(HashEntry<K,V>[] newTable) {
threshold = (int)(newTable.length * loadFactor);
table = newTable;
}
HashEntry<K,V> getFirst(int hash) {
HashEntry<K,V>[] tab = table;
return tab[hash & (tab.length - 1)];
}
}
ConcurrentHashMap 初始化
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
if(!(loadFactor > 0) || initialCapacity < 0 ||
concurrencyLevel <= 0)
throw new IllegalArgumentException();
if(concurrencyLevel > MAX_SEGMENTS)
concurrencyLevel = MAX_SEGMENTS;
int sshift = 0;
int ssize = 1;
while(ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
segmentShift = 32 - sshift;
segmentMask = ssize - 1;
this.segments = Segment.newArray(ssize);
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
int c = initialCapacity / ssize;
if(c * ssize < initialCapacity)
++c;
int cap = 1;
while(cap < c)
cap <<= 1;
for(int i = 0; i < this.segments.length; ++i) {
this.segments[i] = new Segment<K,V>(cap, loadFactor);
}

ConcurrentHashMap在初始化的过程中,创建了segments数组。ConcurrentHashMap 的结构示意图如下所示:

put 方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
public V put(K key, V value) {
// 不允许value为null
if (value == null)
throw new NullPointerException();
int hash = hash(key.hashCode());
// 根据哈希值找到在segments数组中对一个的segment
return segmentFor(hash).put(key, hash, value, false);
}
V put(K key, int hash, V value, boolean onlyIfAbsent) {
// segment.put 加锁,锁对象是segment而非整个table
lock();
try {
int c = count;
// 动态扩容
if (c++ > threshold)
rehash();
// 找出table中key index处的元素
HashEntry<K,V>[] tab = table;
int index = hash & (tab.length - 1);
HashEntry<K,V> first = tab[index];
HashEntry<K,V> e = first;
while (e != null && (e.hash != hash || !key.equals(e.key)))
e = e.next;
V oldValue;
if (e != null) {
oldValue = e.value;
// key处已有值,根据onlyIfAbsent觉得是否需要覆盖
if (!onlyIfAbsent) {
e.value = value;
}
else {
// 元素封装成 HashEntry,添加至表头
oldValue = null;
++modCount;
tab[index] = new HashEntry<K,V>(key, hash, first, value);
count = c;
}
return oldValue;
} finally {
// 释放锁
unlock();
}
}

segment是继承ReentrantLockput操作在开始之前会通过调用lock获取锁,添加元素完毕后,调用unlock释放锁。从此可以看出,ConcurrentHashMap在锁方面优化点之一,引入segment,将锁分成N段,每次操作集合,只会锁住对应的segment而非整张表,减小锁粒度,支持一定数量并发写入,提升了并发效率。

读写锁分离-完全并发读
get 方法

ConcurrentHashMap中的读操作如get方法是没有加锁的。在更新操作中,最后都会更新count变量。countvolatile类型,在不加锁的前提下,也可以保证被准确读取。而在读的时候也会去首先判断count的值。如果写入过程读取值,就要加锁等待其他操作释放锁之后再去读取。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
V get(Object key, int hash) {
if(count != 0) {
HashEntry<K,V> e = getFirst(hash);
while(e != null) {
if(e.hash == hash && key.equals(e.key)) {
V v = e.value;
if(v != null)
return v;
// 写入完成前读取需加锁
return readValueUnderLock(e);
}
e = e.next;
}
}
return null;
}

ConcurrentHashMap中,在不加锁的前提下可以成功读取值,这种读写分离锁的实现,减少了请求获取锁的频次,使得并发效率进一步提高。

JDK1.8 ConcurrentHashMap

JDK1.8中,ConcurrentHashMap的实现不再使用Segment做锁分段方法。新版本中ConcurrentHashMap采用底层的CPUCAS指令和synchronized来实现锁机制。数据存储和HashMap一致,采用数组、链表和红黑树实现。

sizeCtl 变量
1
private transient volatile int sizeCtl;

sizeCtl是控制标识符,不同的值代表不同的意义。

  • -1 表示正在初始化
  • -N 表示N-1个线程正在进行扩容
  • 0 代表尚未初始化
  • >0 扩容阈值
table 初始化

table初始化是在put操作过程中进行的。可以从源码角度看一下initTable是如何保证在多线程环境下,只会初始化一次。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
if ((sc = sizeCtl) < 0)
// sizeCtl 小于0表明有其他线程正在操作table 初始化或者扩容,当前线程让出CPU
Thread.yield(); // lost initialization race; just spin
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
// 通过CAS机制讲更新sizeCtl为-1,保证线程安全。
try {
if ((tab = table) == null || tab.length == 0) {
// table 初始化
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;
}

CASCompare and Swap),从字面意思来看就是比较并交换。CAS有3个操作数,原始值V,预期值A,要修改的值B,当且仅当原始值V等于预期值A的时候,才会将V修改为BJava中通过sun.misc.Unsafe类调用JNI代码来实现CPUCAS指令。

这里通过借助CAS实现了区别于内部悲观独占锁synchronized的乐观锁来实现ConcurrentHashMap的并发安全。

put 方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
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;
if (tab == null || (n = tab.length) == 0)
tab = initTable();
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
}
else if ((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;
}
}
}
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;
}
  • keyvalue为空,抛出NP异常,表明ConcurrentHashMap不允许keyvalue为空
  • 调用spread方法计算出key的哈希值
  • 遍历table
    • 如果table为空,进行初始化工作
    • 当前index没有其他元素,调用casTabAt通过CAS更新元素值
    • 检测到其他线程正在扩容,会调用helpTransfer方法协助其调用
    • 当发生哈希碰撞,无论是链表还是红黑树,添加元素的操作都需要上锁synchronized
get 方法

ConcurrentHashMapget方法,没有上锁,表明ConcurrentHashMap在读操作上是支持完全并发的。效率层面不受加锁机制的影响。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
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)
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;
}

总结

在多线程并发环境下,HashMap是肯定不能用。我们要选择适用于多线程高并发场景的集合类。

ConcurrentHashMap支持完全并发读操作,从效率上来说是优于HashTable,但由于ConcurrentHashMap在读操作中存在弱一致性,所以还是需要结合场景来决定是否用ConcurrentHashMap替代HashTable

参考

探索 ConcurrentHashMap 高并发性的实现机制