Doug Lea大神在j.u.c包下给我们提供了一个适用于多线程并发环境使用的集合类ConcurrentHashMap。而如果在多线程环境，不考虑任何线程安全的防范的话，使用HashMap会带来诸多问题。
HashMap的并发问题
HashMap集合是非线程安全，在多线程环境下容易出现问题。HashMap在数据更新的时候，会带来很多问题：
- 数据丢失
多线程环境下，如两个线程同时在一个bucket下put元素，有可能会造成元素写入被覆盖，从而丢失数据。
- fail-fast机制
当多个线程对同一个集合进行操作时候，就会触发fail-fast机制并抛出ConcurrentModificationException。
- 死循环
当HashMap中table的大小不够时，如果两个线程同时进行resize的操作。如果某一bucket下元素是用链表记录，在resize过程中，链表在多线程的环境下有可能会形成闭合回链，get请求就会造成死循环，使得CPU飙升。详细可以看看这篇博文：疫苗：JAVA HASHMAP的死循环。不过JDK1.8中已经改写resize方法，应该不会出现这种问题，但这并不是我们可以在多线程环境下使用HashMap理由。
...
线程安全类 HashTable
HashTable是线程安全的，底层是通过synchronized来保证线程安全。当多线程竞争激烈的时候，没有获得锁的线程都将会阻塞。synchronized修饰所有针对HashTable集合的操作。这样一旦有线程获得锁，其他的线程都只能等待锁的释放，然后再去竞争锁，这样一来HashTable的效率必定会受到影响。
ConcurrentHashMap 锁机制
相对于低效的HashTable，ConcurrentHashMap在锁机制层面上做了优化。锁优化的思路一般有一下几种：
- 减少锁持有时间
- 减小锁粒度
- 锁分离
- 锁粗化
- 锁消除
JDK1.6 ConcurrentHashMap
ConcurrentHashMap中存储的元素是通过静态内部类HashEntry封装实现的。
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 型，保证其在内存中可见性。key，hash 和 next 都被声明为 final 型。ConcurrentHashMap存储数据根据key的hash值将数据元素散列到哈希表中每一个bucket中。当发生哈希碰撞时候，会将元素封装成HashEntry构成链表。由于next是final类型的，链表中添加元素都将从表头添加。
减小锁粒度-分段锁
在JDK1.7中，ConcurrentHashMap在锁优化过程通过减小锁粒度的实现了对集合的高效并发操作。ConcurrentHashMap包含一个静态内部类Segment，是用来充当锁的角色。
每个Segment将维护若干个bucket。而锁只针对Segment而不是整张表，从而实现减小锁的粒度。
Segment类
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 初始化
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 方法
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是继承ReentrantLock。put操作在开始之前会通过调用lock获取锁，添加元素完毕后，调用unlock释放锁。从此可以看出，ConcurrentHashMap在锁方面优化点之一，引入segment，将锁分成N段，每次操作集合，只会锁住对应的segment而非整张表，减小锁粒度，支持一定数量并发写入，提升了并发效率。
读写锁分离-完全并发读
get 方法
ConcurrentHashMap中的读操作如get方法是没有加锁的。在更新操作中，最后都会更新count变量。count是volatile类型，在不加锁的前提下，也可以保证被准确读取。而在读的时候也会去首先判断count的值。如果写入过程读取值，就要加锁等待其他操作释放锁之后再去读取。
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采用底层的CPU的CAS指令和synchronized来实现锁机制。数据存储和HashMap一致，采用数组、链表和红黑树实现。
sizeCtl 变量
private transient volatile int sizeCtl;
sizeCtl是控制标识符，不同的值代表不同的意义。
-
-1表示正在初始化 -
-N表示N-1个线程正在进行扩容 -
0代表尚未初始化 -
>0扩容阈值
table 初始化
table初始化是在put操作过程中进行的。可以从源码角度看一下initTable是如何保证在多线程环境下，只会初始化一次。
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;
}
CAS（Compare and Swap），从字面意思来看就是比较并交换。CAS有3个操作数，原始值V，预期值A，要修改的值B，当且仅当原始值V等于预期值A的时候，才会将V修改为B。Java中通过sun.misc.Unsafe类调用JNI代码来实现CPU的CAS指令。
这里通过借助CAS实现了区别于内部悲观独占锁synchronized的乐观锁来实现ConcurrentHashMap的并发安全。
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;
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;
}
-
key或value为空,抛出NP异常，表明ConcurrentHashMap不允许key或value为空 - 调用
spread方法计算出key的哈希值 - 遍历
table- 如果
table为空，进行初始化工作 - 当前
index没有其他元素，调用casTabAt通过CAS更新元素值 - 检测到其他线程正在扩容，会调用
helpTransfer方法协助其调用 - 当发生哈希碰撞，无论是链表还是红黑树，添加元素的操作都需要上锁
synchronized
- 如果
get 方法
ConcurrentHashMap的get方法，没有上锁，表明ConcurrentHashMap在读操作上是支持完全并发的。效率层面不受加锁机制的影响。
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。
网友评论