ConcurrentHashMap集合学习

一、JDK1.7 和 1.8 版本ConcurrenHashMap对比分析

JDK 1.7版本

在JDK 1.7版本ConcurrentHashMap使用了分段锁的方式（对Segment进行加锁），其实际结构为：Segment数组 + HashEntry数组 + 链表。由很多个 Segment 组合，而每一个 Segment 是一个类似于 HashMap 的结构，其可以进行扩容。但是 Segment 的个数一旦初始化就不能改变，默认 Segment 的个数是 16 个，你也可以认为 ConcurrentHashMap 默认支持最多 16 个并发线程。
（Segment 继承了 ReentrantLock，所以 Segment 内部可以很方便的获取锁）

JDK 1.8版本

JDK1.8版本 中的 ConcurrentHashMap 使用的 Synchronized 锁 + CAS 的机制。结构也由 Java7 中的 Segment 数组 + HashEntry 数组 + 链表 进化成了 Node 数组 + 链表 / 红黑树，Node 是类似于一个 HashMap 的结构。它的冲突再达到一定大小时会转化成红黑树，在冲突小于一定数量时又退回链表。

Java 1.8 在链表长度超过一定阈值 8 时将链表（寻址时间复杂度为 O(N)）转换为红黑树（寻址时间复杂度为 O(log(N))）。Java 8 中，锁粒度更细，synchronized 只锁定当前链表或红黑二叉树的首节点，这样只要 hash 不冲突，就不会产生并发，就不会影响其他 Node 的读写，效率大幅提升。

二、JDK 1.8版本源码分析

（一）put 方法

• 根据 key 计算出 hashcode 。便于之后进一步确定Node的位置。
• 判断是否需要进行初始化。 ConcurrentHashMap 的初始化是通过自旋和 CAS 操作完成的。
• 即为当前 key 定位出的Node，如果为空表示当前位置可以写入数据，利用 CAS 尝试写入，失败则自旋保证成功。
• 如果当前位置的 hashcode ==MOVED == -1,则需要进行扩容。
• 如果都不满足，则利synchronized 锁写入数据。（synchronized锁当前Node节点，相比与JDK1.7版本的segment分段锁，锁粒度小）
• 如果数量大于一定长度TREEIFY_THRESHOLD 则要执行树化方法，在 treeifyBin 中会首先判断当前数组长度 ≥64时才会将链表转换为红黑树。

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();int hash = spread(key.hashCode());int binCount = 0;for (Node<K,V>[] tab = table;;) {// f = 目标位置元素Node<K,V> f; int n, i, fh;// fh 后面存放目标位置的元素 hash 值if (tab == null || (n = tab.length) == 0)// 数组桶为空，初始化数组桶（自旋+CAS)tab = initTable();else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {// 桶内为空，CAS 放入，不加锁，成功了就直接 break 跳出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 加锁加入节点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;
}

（二）get方法

总结一下 get 过程：

• 根据 hash 值计算位置。
• 查找到指定位置，如果头节点就是要找的，直接返回它的 value。
• 如果头节点 hash 值小于 0，说明正在扩容或者是红黑树，查找之。
• 如果是链表，遍历查找之。

public V get(Object key) {Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;// key 所在的 hash 位置int h = spread(key.hashCode());if ((tab = table) != null && (n = tab.length) > 0 &&(e = tabAt(tab, (n - 1) & h)) != null) {// 如果指定位置元素存在，头结点hash值相同if ((eh = e.hash) == h) {if ((ek = e.key) == key || (ek != null && key.equals(ek)))// key hash 值相等，key值相同，直接返回元素 valuereturn e.val;}else if (eh < 0)// 头结点hash值小于0，说明正在扩容或者是红黑树，find查找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;
}

三、JDK 1.7 和 JDK 1.8 的 ConcurrentHashMap 实现有什么不同？

• 线程安全实现方式：JDK 1.7 采用 Segment 分段锁来保证安全， Segment 是继承自 ReentrantLock。JDK1.8 放弃了 Segment 分段锁的设计，采用 Node + CAS + synchronized 保证线程安全，锁粒度更细，synchronized 只锁定当前链表或红黑二叉树的首节点。
• Hash 碰撞解决方法 : JDK 1.7 采用拉链法，JDK1.8 采用拉链法结合红黑树（链表长度超过一定阈值时，将链表转换为红黑树）。
• 并发度：JDK 1.7 最大并发度是 Segment 的个数，默认是 16。JDK 1.8 最大并发度是 Node 数组的大小，并发度更大。