参考：http://www.cnblogs.com/skywang12345/p/3498537.html
1、ConcurrentHashMap介绍
ConcurrentHashMap是线程安全的哈希表。
HashMap, Hashtable, ConcurrentHashMap之间的关联如下：
-
HashMap是非线程安全的哈希表，常用于单线程程序中。
-
Hashtable是线程安全的哈希表，它是通过synchronized来保证线程安全的。Hashtable在线程竞争激烈时，效率比较低(此时建议使用ConcurrentHashMap)！因为当一个线程访问Hashtable的同步方法时，其它线程访问Hashtable的同步方法，可能会进入阻塞状态。
-
ConcurrentHashMap是线程安全的哈希表，JDK1.7中，它是通过“分段锁”来保证线程安全的。
ConcurrentHashMap将哈希表分成许多片段(Segment)，每一个片段除了保存哈希表之外，本质上也是一个“可重入的互斥锁”(ReentrantLock)。多线程对同一个片段的访问，是互斥的；但是，对于不同片段的访问，却是可以同步进行的。
这里写图片描述
2、要点
这里写图片描述
-
ConcurrentHashMap继承于AbstractMap抽象类。
-
Segment是ConcurrentHashMap中的内部类，它就是ConcurrentHashMap中的“分段锁”对应的存储结构。ConcurrentHashMap与Segment是组合关系，1个ConcurrentHashMap对象包含若干个Segment对象。在代码中，这表现为ConcurrentHashMap类中存在“Segment数组”成员。
-
Segment类继承于ReentrantLock类，所以Segment本质上是一个可重入的互斥锁。
-
HashEntry也是ConcurrentHashMap的内部类，是单向链表节点，存储着key-value键值对。Segment与HashEntry是组合关系，Segment类中存在“HashEntry数组”成员，“HashEntry数组”中的每个HashEntry就是一个单向链表。
对于多线程访问对一个“哈希表对象”竞争资源，Hashtable是通过一把锁来控制并发； 而ConcurrentHashMap则是将哈希表分成许多片段，对于每一个片段分别通过一个互斥锁来控制并发。 ConcurrentHashMap对并发的控制更加细腻，它也更加适应于高并发场景！
3、Segment（片段）
ConcurrentHashMap维护一个segment数组，将元素分成若干段，它通过“segments数组”对象来保存各个分段，第一次hash找到对应的Segment。
final Segment<K,V>[] segments;
Segment是ConcurrentHashMap的内部类：
static final class Segment<K,V> extends ReentrantLock implements Serializable {
...
//每一个Segment有一个保存HashEntry的数组（第二次hash）
transient volatile HashEntry<K,V>[] table;
// threshold阈，是哈希表在其容量自动增加之前可以达到多满的一种尺度。
transient int threshold;
// loadFactor是加载因子
final float loadFactor;
Segment(float lf, int threshold, HashEntry<K,V>[] tab) {
this.loadFactor = lf;
this.threshold = threshold;
this.table = tab;
}
...
}
说明：
（1）Segment 包含 HashEntry数组，HashEntry保存了哈希表中的键值对。
（2）Segment 继承于 ReentrantLock 。这意味着，Segment本质上就是可重入的互斥锁。
4、HashEntry
HashEntry是链表中的单个节点。
JDK1.7中，ConcurrentHashMap是链式哈希表，它是通过“拉链法”来解决哈希冲突的。
static final class HashEntry<K,V> {
final int hash; // 哈希值
final K key; // 键
volatile V value; // 值
volatile HashEntry<K,V> next; // 下一个HashEntry节点
HashEntry(int hash, K key, V value, HashEntry<K,V> next) {
this.hash = hash;
this.key = key;
this.value = value;
this.next = next;
}
...
}
5、ConcurrentHashMap的创建，添加，获取，删除
5.1 创建
下面以ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel)来进行说明。
@SuppressWarnings("unchecked")
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
// 参数有效性判断
if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
// concurrencyLevel是“用来计算segments的容量”
if (concurrencyLevel > MAX_SEGMENTS)
concurrencyLevel = MAX_SEGMENTS;
int sshift = 0;
int ssize = 1;
// ssize=“大于或等于concurrencyLevel的最小的2的N次方值”
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
// 初始化segmentShift和segmentMask
this.segmentShift = 32 - sshift;
this.segmentMask = ssize - 1;
// 哈希表的初始容量
// 哈希表的实际容量=“segments的容量” x “segments中数组的长度”
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
// “哈希表的初始容量” / “segments的容量”
int c = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
// cap就是“segments中的HashEntry数组的长度”
int cap = MIN_SEGMENT_TABLE_CAPACITY;
while (cap < c)
cap <<= 1;
// segments
Segment<K,V> s0 =
new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
(HashEntry<K,V>[])new HashEntry[cap]);
Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
this.segments = ss;
}
5.2 添加
下面以put(K key, V value)来对ConcurrentHashMap中增加键值对来进行说明。
public V put(K key, V value) {
Segment<K,V> s;
if (value == null)
throw new NullPointerException();
// 获取key对应的哈希值
int hash = hash(key);
int j = (hash >>> segmentShift) & segmentMask;
// 如果找不到该Segment，则新建一个。
if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck
(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
s = ensureSegment(j);
//Segment中的put函数见下面的分析
return s.put(key, hash, value, false);
}
Segment中的put函数：
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
// tryLock()获取锁，成功返回true，失败返回false。
// 获取锁失败的话，则通过scanAndLockForPut()“自旋”获取锁，并返回”要插入的key-value“对应的”HashEntry链表。
HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value);
V oldValue;
try {
// tab代表”当前Segment中的HashEntry数组“
HashEntry<K,V>[] tab = table;
// 根据”hash值“获取”HashEntry数组中对应的HashEntry链表“
int index = (tab.length - 1) & hash;
HashEntry<K,V> first = entryAt(tab, index);
for (HashEntry<K,V> e = first;;) {
// 如果”HashEntry链表中的当前HashEntry节点“不为null，
if (e != null) {
K k;
// 当”要插入的key-value键值对“已经存在于”HashEntry链表中“时，先保存原有的值。
// 若”onlyIfAbsent“为true，即”要插入的key不存在时才插入”，则直接退出；
// 否则，用新的value值覆盖原有的原有的值。
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
oldValue = e.value;
if (!onlyIfAbsent) {
e.value = value;
++modCount;
}
break;
}
e = e.next;
}
else {
// 如果node非空，则将first设置为“node的下一个节点”。
// 否则，新建HashEntry链表
if (node != null)
node.setNext(first);
else
node = new HashEntry<K,V>(hash, key, value, first);
int c = count + 1;
// 如果添加key-value键值对之后，Segment中的元素超过阈值(并且，HashEntry数组的长度没超过限制)，
// 则rehash；
// 否则，直接添加key-value键值对。
if (c > threshold && tab.length < MAXIMUM_CAPACITY)
rehash(node); //扩容
else
setEntryAt(tab, index, node);
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
// 释放锁
unlock();
}
return oldValue;
}
从上边观察，总共进行了两次hash计算：
第一次，调用hash()函数，计算key的hash值，从而找到对应的Segment片段：
private int hash(Object k) {
int h = hashSeed;
if ((0 != h) && (k instanceof String)) {
return sun.misc.Hashing.stringHash32((String) k);
}
h ^= k.hashCode();
// Spread bits to regularize both segment and index locations,
// using variant of single-word Wang/Jenkins hash.
h += (h << 15) ^ 0xffffcd7d;
h ^= (h >>> 10);
h += (h << 3);
h ^= (h >>> 6);
h += (h << 2) + (h << 14);
return h ^ (h >>> 16);
}
第二次，在Segment中，根据key的hash值，计算插入位置index：
int index = (tab.length - 1) & hash;
5.3 自旋函数scanAndLockForPut()
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
// 第一个HashEntry节点
HashEntry<K,V> first = entryForHash(this, hash);
// 当前的HashEntry节点
HashEntry<K,V> e = first;
HashEntry<K,V> node = null;
// 重复计数(自旋计数器)
int retries = -1; // negative while locating node
// 查找”key-value键值对“在”HashEntry链表上对应的节点“；
// 若找到的话，则不断的自旋；在自旋期间，若通过tryLock()获取锁成功则返回；否则自旋MAX_SCAN_RETRIES次数之后，强制获取”锁“并退出。
// 若没有找到的话，则新建一个HashEntry链表。然后不断的自旋。
// 此外，若在自旋期间，HashEntry链表的表头发生变化；则重新进行查找和自旋工作！
while (!tryLock()) {
HashEntry<K,V> f; // to recheck first below
// 1. retries<0的处理情况
if (retries < 0) {
// 1.1 如果当前的HashEntry节点为空(意味着，在该HashEntry链表上上没有找到”要插入的键值对“对应的节点)，而且node=null；
// 则新建HashEntry链表。
if (e == null) {
if (node == null) // speculatively create node
node = new HashEntry<K,V>(hash, key, value, null);
retries = 0;
}
// 1.2 如果当前的HashEntry节点是”要插入的键值对在该HashEntry上对应的节点“，则设置retries=0
else if (key.equals(e.key))
retries = 0;
// 1.3 设置为下一个HashEntry。
else
e = e.next;
}
// 2. 如果自旋次数超过限制，则获取“锁”并退出
else if (++retries > MAX_SCAN_RETRIES) {
lock();
break;
}
// 3. 当“尝试了偶数次”时，就获取“当前Segment的第一个HashEntry”，即f。
// 然后，通过f!=first来判断“当前Segment的第一个HashEntry是否发生了改变”。
// 若是的话，则重置e，first和retries的值，并重新遍历。
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first) {
e = first = f; // re-traverse if entry changed
retries = -1;
}
}
return node;
}
scanAndLockForPut()的目标是获取锁。流程如下：
它首先会调用entryForHash()，根据hash值获取”当前Segment中对应的HashEntry节点(first)，即找到对应的HashEntry链表“。
紧接着进入while循环。在while循环中，它会遍历”HashEntry链表(e)“，查找”要插入的key-value键值对“在”该HashEntry链表上对应的节点“。
若找到的话，则不断的自旋，即不断的执行while循环。在自旋期间，若通过tryLock()获取锁成功则返回；否则，在自旋MAX_SCAN_RETRIES次数之后，强制获取锁并退出。
若没有找到的话，则新建一个HashEntry链表，然后不断的自旋。在自旋期间，若通过tryLock()获取锁成功则返回；否则，在自旋MAX_SCAN_RETRIES次数之后，强制获取锁并退出。
此外，若在自旋期间，HashEntry链表的表头发生变化；则重新进行查找和自旋工作！
5.4 扩容函数rehash()
private void rehash(HashEntry<K,V> node) {
HashEntry<K,V>[] oldTable = table;
// ”Segment中原始的HashEntry数组的长度“
int oldCapacity = oldTable.length;
// ”Segment中新HashEntry数组的长度“
int newCapacity = oldCapacity << 1; //新容量 = 旧容量 * 2
// 新的阈值
threshold = (int)(newCapacity * loadFactor);
// 创建一个新的HashEntry数组
HashEntry<K,V>[] newTable =
(HashEntry<K,V>[]) new HashEntry[newCapacity];
int sizeMask = newCapacity - 1;
// 遍历”原始的HashEntry数组“，
// 将”原始的HashEntry数组“中的每个”HashEntry链表“的值，都复制到”新的HashEntry数组的HashEntry元素“中。
for (int i = 0; i < oldCapacity ; i++) {
// 获取”原始的HashEntry数组“中的”第i个HashEntry链表“
HashEntry<K,V> e = oldTable[i];
if (e != null) {
HashEntry<K,V> next = e.next;
int idx = e.hash & sizeMask;
if (next == null) // Single node on list
newTable[idx] = e;
else { // Reuse consecutive sequence at same slot
HashEntry<K,V> lastRun = e;
int lastIdx = idx;
for (HashEntry<K,V> last = next;
last != null;
last = last.next) {
int k = last.hash & sizeMask;
if (k != lastIdx) {
lastIdx = k;
lastRun = last;
}
}
newTable[lastIdx] = lastRun;
// 将”原始的HashEntry数组“中的”HashEntry链表(e)“的值，都复制到”新的HashEntry数组的HashEntry“中。
for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
V v = p.value;
int h = p.hash;
int k = h & sizeMask;
HashEntry<K,V> n = newTable[k];
newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
}
}
}
}
// 将新的node节点添加到“Segment的新HashEntry数组(newTable)“中。
int nodeIndex = node.hash & sizeMask; // add the new node
node.setNext(newTable[nodeIndex]);
newTable[nodeIndex] = node;
table = newTable;
}
5.5 获取
下面以get(Object key)为例，对ConcurrentHashMap的获取方法进行说明。
public V get(Object key) {
Segment<K,V> s; // manually integrate access methods to reduce overhead
HashEntry<K,V>[] tab;
int h = hash(key);
long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
// 获取key对应的Segment片段。
// 如果Segment片段不为null，则在“Segment片段的HashEntry数组中”中找到key所对应的HashEntry列表；
// 接着遍历该HashEntry链表，找到于key-value键值对对应的HashEntry节点。
if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
(tab = s.table) != null) {
for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
(tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
e != null; e = e.next) {
K k;
if ((k = e.key) == key || (e.hash == h && key.equals(k)))
return e.value;
}
}
return null;
}
5.6 删除
下面以remove(Object key)来对ConcurrentHashMap中的删除操作来进行说明。
public V remove(Object key) {
int hash = hash(key);
// 根据hash值，找到key对应的Segment片段。
Segment<K,V> s = segmentForHash(hash);
return s == null ? null : s.remove(key, hash, null);
}
Segment中的remove函数：
final V remove(Object key, int hash, Object value) {
// 尝试获取Segment对应的锁。
// 尝试失败的话，则通过scanAndLock()来获取锁。
if (!tryLock())
scanAndLock(key, hash);
V oldValue = null;
try {
// 根据“hash值”找到“Segment的HashEntry数组”中对应的“HashEntry节点(e)”，该HashEntry节点是一HashEntry个链表。
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash;
HashEntry<K,V> e = entryAt(tab, index);
HashEntry<K,V> pred = null;
// 遍历“HashEntry链表”，删除key-value键值对
while (e != null) {
K k;
HashEntry<K,V> next = e.next;
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
V v = e.value;
if (value == null || value == v || value.equals(v)) {
if (pred == null)
setEntryAt(tab, index, next);
else
pred.setNext(next);
++modCount;
--count;
oldValue = v;
}
break;
}
pred = e;
e = next;
}
} finally {
// 释放锁
unlock();
}
return oldValue;
}
说明：remove()的目的就是删除key-value键值对。在删除之前，它会获取到Segment的互斥锁，在删除之后，再释放锁。
它的删除过程也比较简单，它会先根据hash值，找到“Segment的HashEntry数组”中对应的“HashEntry”节点。根据Segment的数据结构，我们知道Segment中包含一个HashEntry数组对象，而每一个HashEntry本质上是一个单向链表。 在找到“HashEntry”节点之后，就遍历该“HashEntry”节点对应的链表，找到key-value键值对对应的节点，然后删除。
5.7 scanAndLock()函数
scanAndLock()的目标是获取锁。它的实现与scanAndLockForPut()类似。
private void scanAndLock(Object key, int hash) {
// 第一个HashEntry节点
HashEntry<K,V> first = entryForHash(this, hash);
HashEntry<K,V> e = first;
int retries = -1;
// 查找”key-value键值对“在”HashEntry链表上对应的节点“；
// 无论找没找到，最后都会不断的自旋；在自旋期间，若通过tryLock()获取锁成功则返回；否则自旋MAX_SCAN_RETRIES次数之后，强制获取”锁“并退出。
// 若在自旋期间，HashEntry链表的表头发生变化；则重新进行查找和自旋！
while (!tryLock()) {
HashEntry<K,V> f;
if (retries < 0) {
// 如果“遍历完该HashEntry链表，仍然没找到”要删除的键值对“对应的节点”
// 或者“在该HashEntry链表上找到”要删除的键值对“对应的节点”，则设置retries=0
// 否则，设置e为下一个HashEntry节点。
if (e == null || key.equals(e.key))
retries = 0;
else
e = e.next;
}
// 自旋超过限制次数之后，获取锁并退出。
else if (++retries > MAX_SCAN_RETRIES) {
lock();
break;
}
// 当“尝试了偶数次”时，就获取“当前Segment的第一个HashEntry”，即f。
// 然后，通过f!=first来判断“当前Segment的第一个HashEntry是否发生了改变”。
// 若是的话，则重置e，first和retries的值，并重新遍历。
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first) {
e = first = f;
retries = -1;
}
}
}
6、总结
-
ConcurrentHashMap是线程安全的哈希表，它是通过“锁分段”来实现的。ConcurrentHashMap中包括了“Segment(锁分段)数组”，每个Segment就是一个哈希表，而且也是可重入的互斥锁。
-
第一，Segment是哈希表表现在，Segment包含了“HashEntry数组”，而“HashEntry数组”中的每一个HashEntry元素是一个单向链表。即Segment是通过链式哈希表。
-
第二，Segment是可重入的互斥锁表现在，Segment继承于ReentrantLock，而ReentrantLock就是可重入的互斥锁。
-
-
对于ConcurrentHashMap的添加，删除操作，在操作开始前，线程都会获取Segment的互斥锁；操作完毕之后，才会释放。而对于读取操作，它是通过volatile去实现的，HashEntry数组是volatile类型的，而volatile能保证“即对一个volatile变量的读，总是能看到（任意线程）对这个volatile变量最后的写入”，即我们总能读到其它线程写入HashEntry之后的值。 以上这些方式，就是ConcurrentHashMap线程安全的实现原理。
网友评论