共计 3806 个字符，预计需要花费 10 分钟才能阅读完成。
jdk1.7 的 ConcurrentHashMap
jdk1.7 采用的是 Segment+HashEntry 数组 + 链表来实现的
ConcurrentHashMap 初始化时有 16 个 Segment，每个 Segment 中又包含了 HashEntry 数组，每个 HashEntry 又是一个链表，可以看出，Segment 的结构和 HashMap 类似，你可以这样理解：ConcurrentHashMap 由多个 HashMap 组成。
线程安全的实现原理
ConcurrentHashMap 中的 Segment 在实现上继承了 ReentrantLock，这样就实现了分段锁的功能。下面我们通过 put()方法来看看它是如何保证线程安全的。
public V put(K key, V value) {
Segment<K,V> s;
// value 不能为 null
if (value == null)
throw new NullPointerException();
// 计算 key 的 hash 值
int hash = hash(key);
// 定位到哪个 Segment
int j = (hash >>> segmentShift) & segmentMask;
if ((s = (Segment<K,V>)UNSAFE.getObject
(segments, (j << SSHIFT) + SBASE)) == null)
// 如果此 Segment 为 null，则初始化 Segment
s = ensureSegment(j);
// 往 Segment 中 put 值
return s.put(key, hash, value, false);
}final V put(K key, int hash, V value, boolean onlyIfAbsent) {
// 尝试获取锁，如果获取不到，则进入 scanAndLockForPut 方法
// scanAndLockForPut 中会再次 tryLock，如果还是获取不到，则 lock()挂起线程
HashEntry<K,V> node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
V oldValue;
try {
// table 通过 volatile 修饰，保证拿到的都是最新的 table
HashEntry<K,V>[] tab = table;
// 得到数组下标，& 取模作用，详细介绍可以看我写的 jdk1.8HashMap 文章
int index = (tab.length - 1) & hash;
// 得到此下标下的 HashEntry
HashEntry<K,V> first = entryAt(tab, index);
// 遍历链表
for (HashEntry<K,V> e = first;;) {if (e != null) {
K k;
// 如果 key 相同，则覆盖原值
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
oldValue = e.value;
if (!onlyIfAbsent) {
e.value = value;
// 修改次数 +1，用于限制遍历中删除
++modCount;
}
break;
}
// 继续遍历下个值
e = e.next;
}
else {if (node != null)
// 头插法（头插法的问题看我写的 jdk1.7HashMap 文章）node.setNext(first);
else
// new 新的 HashEntry，里面也是采用头插法
node = new HashEntry<K,V>(hash, key, value, first);
int c = count + 1;
// 如果大于阈值，开始扩容
if (c > threshold && tab.length < MAXIMUM_CAPACITY)
rehash(node);
else
// 将链表放到数组下标处
setEntryAt(tab, index, node);
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
// 释放锁
unlock();}
return oldValue;
}通过上面源码可以看出 jdk1.7 的 ConcurrentHashMap 通过 Segment 的 ReentrantLock 锁来实现线程安全。要注意的是上锁操作是this.lock()，也就是每个 Segment 的锁是分开的，其中一个上锁并不影响其他 Segment 操作，说明可以同时 16 个线程同时进来对不同 Segment 操作，这样可以有效的提高并发度。
这种 Segment+ReentrantLock 看起来可以很好的解决并发问题了，那为什么 jdk1.8 又不继续使用此方法了呢？下面我们看看 jdk1.8 的 ConcurrentHashMap
jdk1.8 的 ConcurrentHashMap
jdk1.8 采用的是 Node 数组 + 链表 / 红黑树实现
jdk1.8 的 ConcurrentHashMap 去除了外层的 Segment，直接使用数组加链表或红黑树，数据结构和 1.8 的 HashMap 一样，不同之处在于使用了 CAS+synchronized 保证了线程安全。
线程安全的实现原理
我们还是通过 put()方法看看它是如何保证线程安全的
public V put(K key, V value) {return putVal(key, value, false);
}
final V putVal(K key, V value, boolean onlyIfAbsent) {
// 同样 value 不能为 null
if (key == null || value == null) throw new NullPointerException();
// 新的 hash 算法，和 1.7 的不同
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)
// 如果 table 为空，初始化
tab = initTable();
// f= 此下标的头节点，i= 下标
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
// 如果此下标节点为空，new 新节点放在此节点下
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
// fh= 头节点的 hash 值
else if ((fh = f.hash) == MOVED)
// 协助扩容
tab = helpTransfer(tab, f);
else {
V oldVal = null;
// 重点来了，synchronized 锁当前 node 节点
synchronized (f) {if (tabAt(tab, i) == f) {
// 头节点 hash 值大于 0，说明是链表
if (fh >= 0) {
// 链表长度
binCount = 1;
// 遍历链表
for (Node<K,V> e = f;; ++binCount) {
K ek;
// key 相同，覆盖原值
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;
// 到链表末端，new 新 node 挂在末端
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;
}通过上面源码可以看出，1.8 的 ConcurrentHashMap 采用的是 synchronized 来实现线程安全。synchronized (f)可以看出锁的是每一个 Node 对象，这样可以将锁细化，除非有多个线程同时操作同一个 Node，否则不会出现争抢锁，这样锁也不会升级，就算出现争抢锁，也可以通过自旋来获取锁，达到一定次数才会升级为重量级锁，这种情况是很少的。那为什么 1.7 的 ReentrantLock 不细化到 Node 上呢？
如果 1.7 的 ReentrantLock 细化到 Node 上，它只会在线程没有抢到锁的时候，新建节点再尝试一次，如果还没抢到，不会自旋，而是直接挂起，这样就增加了线程上下文开销的代价。当然，ReentrantLock 的 tryLock()也是可以自旋的，但是确定不了 tryLock 的时间。
所以，在锁细化到如此程度的情况下，使用 synchronized 是最好的选择。
<center> 扫一扫，关注我 </center>
