【集合Map系列四】ConcurrentHashMap

???欢迎来到我的博客，很高兴能够在这里和您见面！希望您在这里可以感受到一份轻松愉快的氛围，不仅可以获得有趣的内容和知识，也可以畅所欲言、分享您的想法和见解。

• 推荐:kuan 的首页,持续学习,不断总结,共同进步,活到老学到老
• 导航
  • 檀越剑指大厂系列:全面总结 java 核心技术点,如集合,jvm,并发编程 redis,kafka,Spring,微服务,Netty 等
  • 常用开发工具系列:罗列常用的开发工具,如 IDEA,Mac,Alfred,electerm,Git,typora,apifox 等
  • 数据库系列:详细总结了常用数据库 mysql 技术点,以及工作中遇到的 mysql 问题等
  • 懒人运维系列:总结好用的命令,解放双手不香吗?能用一个命令完成绝不用两个操作
  • 数据结构与算法系列:总结数据结构和算法,不同类型针对性训练,提升编程思维,剑指大厂

非常期待和您一起在这个小小的网络世界里共同探索、学习和成长。??? ?? 欢迎订阅本专栏 ??

1.jdk1.7 中 CHM 数据结构？

ConcurrentHashMap 和 HashMap 结构差不多，不过 ConcurrentHashMap 支持并发操作。所以结构更加复杂一些。

整个 ConcurrentHashMap 由一个个 segment 组成。segment 代表一段的意思。所以 ConcurrentHashMap 也叫分段锁。简单理解，ConcurrentHashMap 是由 segment 数组组成。segment 继承自 reentrantlock 来进行加锁，所以每个 segment 是线程安全的，整个 ConcurrentHashMap 就是线程安全的。

ConcurrentHashMap 与 HashMap 和 Hashtable 最大的不同在于：put 和 get 两次 Hash 到达指定的 HashEntry，第一次 hash 到达 Segment,第二次到达 Segment 里面的 Entry,然后在遍历 entry 链表.

2.CHM 的构造函数有几个？

public ConcurrentHashMap() {
    
}public ConcurrentHashMap(int initialCapacity) {
    if (initialCapacity < 0)throw new IllegalArgumentException();int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?MAXIMUM_CAPACITY :tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));this.sizeCtl = cap;
}public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
    this.sizeCtl = DEFAULT_CAPACITY;putAll(m);
}public ConcurrentHashMap(int initialCapacity, float loadFactor) {
    this(initialCapacity, loadFactor, 1);
}
//可设置初始容量和阈值和并发级别
public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel) {
    if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)throw new IllegalArgumentException();if (initialCapacity < concurrencyLevel)   // Use at least as many binsinitialCapacity = concurrencyLevel;   // as estimated threadslong size = (long)(1.0 + (long)initialCapacity / loadFactor);int cap = (size >= (long)MAXIMUM_CAPACITY) ?MAXIMUM_CAPACITY : tableSizeFor((int)size);this.sizeCtl = cap;
}

concurrencyLevel:并发级别，并发数，segment 数。默认是 16，也就是说默认是 16 个 segments。

理论上来说，最多支持 16 个线程并发写。操作分布在不同的 segment 上，对单独的 segment 进行加锁处理，可以做到线程安全，可以在初始化的时候设置此值，设置之后不支持扩容。

3.jdk1.7 的 CHM 是如何进行锁操作的？

数据结构：ReentrantLock+Segment+HashEntry 组成，写的时候对单个 segment 加锁

4.jdk1.8 的 CHM 是如何保证并发的？

数据结构为数组+链表/红黑树，内部大量采用 cas 来实现。JDK8 中 ConcurrentHashMap 参考了 JDK8 HashMap 的实现，采用了数组+链表/红黑树的实现方式来设计，内部大量采用 CAS 操作。

CAS 是 compare and swap 的缩写，即我们所说的比较交换。cas 是一种基于锁的操作，而且是乐观锁。在 java 中锁分为乐观锁和悲观锁。悲观锁是将资源锁住，等一个之前获得锁的线程释放锁之后，下一个线程才可以访问。而乐观锁采取了一种宽泛的态度，通过某种方式不加锁来处理资源，比如通过给记录加 version 来获取数据，性能较悲观锁有很大的提高。

JDK8 中彻底放弃了 Segment 转而采用的是 Node，其设计思想也不再是 JDK1.7 中的分段锁思想。

Node:保存 key，value 及 key 的 hash 值的数据结构。其中 value 和 next 都用 volatile 修饰，保证并发的

可见性。

static class Node<K,V> implements Map.Entry<K,V> {
    final int hash;final K key;volatile V val;//并发可见性volatile Node<K,V> next;//并发可见性Node(int hash, K key, V val, Node<K,V> next) {
    this.hash = hash;this.key = key;this.val = val;this.next = next;}
}

5.jdk1.8 的 CHM 和 jdk1.7 的区别？

数据结构上的区别，在 jdk1.7 中使用的是 ReentrantLock+Segment+HashEntry ，在 jdk1.8 中使用的是 Node+CAS+synchronized+红黑树。

• JDK1.8 的实现降低锁的粒度，JDK1.7 版本锁的粒度是基于 Segment 的，包含多个 HashEntry，而 JDK1.8 锁的粒度就是 HashEntry（首节点）
• 取消了 segment 分段锁，采用数组+链表+红黑树。
• 1.7 中用 ReentrantLock+Segment 加锁，1.8 中使用的是 CAS+synchronized 加锁,对数组元素 Node 加锁。
• 在链表节点大于 8 时,且数组长度大于等于 64 时，会转为红黑树。数据量大时，hash 冲突加剧，性能下降。
• 查询时间复杂度，1.7 最坏时是单链表 O(n),1.8 是红黑树 O(logn)
• JDK1.8 版本的数据结构变得更加简单，使得操作也更加清晰流畅，因为已经使用 synchronized 来进行同步，所以不需要分段锁的概念，也就不需要 Segment 这种数据结构了，由于粒度的降低，实现的复杂度也增加了
• JDK1.8 使用红黑树来优化链表，基于长度很长的链表的遍历是一个很漫长的过程，而红黑树的遍历效率是很快的，代替一定阈值的链表，这样形成一个最佳拍档

6.Hashtable 和 CHM 在多线程下区别？

多线程环境下都是线程安全的，ConcurrentHashMap 的效率更高。

HashTable 使用一把锁处理并发问题，在多线程情况下，多个线程竞争同一个锁，效率较低，导致阻塞。

ConcurrentHashMap 分两个版本

• 1.7 使用分段锁，相当于把 HashMap 分成多段，每一段都拥有各自的锁，这样可以实现多线程访问。
• 1.8 采用了 cas 和 synchronized 加锁，锁粒度细化到元素本身，理论上是最高级别的并发。

7.CHM 的 get()为什么不需要加锁?

ConcurrentHashMap 的 get 操作可以无锁是由于 Node 的元素 val 和指针 next 是用 volatile 修饰的，在多线程环境下线程 A 修改结点的 val 或者新增节点的时候是对线程 B 可见的。

在 Node 上加 volatile 的目的是什么呢？
其实就是为了使得 Node 数组在扩容的时候对其他线程具有可见性而加的 volatile。

get 操作全程不需要加锁是因为 Node 的成员 val 是用 volatile 修饰的和 Node 用 volatile 修饰没有关系。
Node 用 volatile 修饰主要是保证在数组扩容的时候保证可见性。

8.CHM 是如何进行扩容的?

private transient volatile int sizeCtl;

通过构造函数可以发现 sizeCtl 变量经常出现，该变量通过查看 jdk 源码注释可知该变量主要控制初始化或扩容：

• -1，表示线程正在进行初始化操作。
• -(1+nThreads)，表示 n 个线程正在进行扩容。
• 0，默认值，后续在真正初始化的时候使用默认容量。
• 大于 0，初始化或扩容完成后下一次的扩容门槛。

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;// 单核不拆分，下面讨论多核的情况// 计算步长，拆分任务n >>> 3 = n / 2^3// 先将n分为8份，然后等分给每个cpu，若最后计算的步长小于最小步长16，则设置为16if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)stride = MIN_TRANSFER_STRIDE; // subdivide rangeif (nextTab == null) {
                // initiatingtry {
    // 扩容 2倍@SuppressWarnings("unchecked")Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];nextTab = nt;} catch (Throwable ex) {
          // try to cope with OOMEsizeCtl = Integer.MAX_VALUE;return;}nextTable = nextTab;// transferIndex 记录迁移进度transferIndex = n;}int nextn = nextTab.length;ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);// 从后面的迁移逻辑看到 迁移复制元素是逆序迁移// advance= true 则代表可继续向前一个位置迁移复制元素boolean advance = true;// 是否所有线程都全部迁移完毕，true则可以将nextTab赋值给table了boolean finishing = false; // to ensure sweep before committing nextTab// i 代表当前线程正在迁移的数组位置，bound代表它本次可以迁移的范围下限for (int i = 0, bound = 0;;) {
    Node<K,V> f; int fh;while (advance) {
    int nextIndex, nextBound;// （1）两种情况不需要继续向前一个位置迁移复制元素（逆序）：// ①i每次自减1，i>=bound说明本批次迁移未完成，不需要继续向前推进。// ②finishing标志为true，说明所有线程分配的迁移任务都已经完成了，则不需要向前推进。// 若 --i < bound，说明当前批次的迁移任务完成，可继续分配新范围的任务// 也就是一个线程可以多次分到任务，能者多劳。if (--i >= bound || finishing)// 向前一个位置迁移复制元素advance = false;//(2) 每次执行，都会把 transferIndex 最新的值同步给 nextIndexelse if ((nextIndex = transferIndex) <= 0) {
    //若 transferIndex小于等于0，则说明原数组中所有位置的迁移任务都分配完毕（不代表所有位置都迁移完毕）//于是，需要跳出while循环，并把 i设为 -1，// 以跳到（4）判断正在处理的线程是否完成自己负责范围内迁移工作。i = -1;advance = false;}else if (U.compareAndSwapInt(this, TRANSFERINDEX, nextIndex,nextBound = (nextIndex > stride ?nextIndex - stride : 0))) {
    //（3）cas 设置TRANSFERINDEX，分配任务范围[nextBound,nextIndex)，任务的长度是stride// 举例，假设 n=64，即初始的transferIndex=64，stride=16// nextIndex=transferIndex=64，nextBound=nextIndex-stride=48// bound=48// i=63// 从后往前复制bound = nextBound;i = nextIndex - 1;advance = false;  // 本次任务分配完成，结束循环}}// （4）i已经越界了，整个数组的迁移任务已经全部分配完毕if (i < 0 || i >= n || i + n >= nextn) {
    int sc;if (finishing) {
    // 扩容完毕// nextTable置为空nextTable = null;// 新数组赋值给旧数组table = nextTab;// sizeCtl 设置为新的数组长度的 3/4.即 3/4 *2nsizeCtl = (n << 1) - (n >>> 1);return;}// 到这，说明所有的迁移任务都分配完了// 当前线程也已经完成了自己的迁移任务（无论参与了几次迁移），// 则sc-1，表明参与扩容的线程数减1if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
    // 迁移开始时，会设置 sc=(rs << RESIZE_STAMP_SHIFT) + 2// 每当有一个线程参与迁移，sc 就会加 1。// 因此，这里就是去校验当前 sc 是否和初始值相等。if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)// 不相等，当前线程扩容任务结束。return;// 相等，说明还有一个线程还在扩容迁移（不一定是触发扩容的第一个线程）// 则当前线程会从后向前检查一遍，哪些位置的节点没有复制完，就帮忙一起复制。// 一圈扫描下来，肯定是全部迁移完毕了，则finishing可提前设置为true。finishing = advance = true;i = n; // recheck before commit}}else if ((f = tabAt(tab, i)) == null)// （5）若i的位置元素为空，就把占位节点设置为fwd标志。// 设置成功，advance置为true，向前推进复制advance = casTabAt(tab, i, null, fwd);else if ((fh = f.hash) == MOVED)// （6）若当前位置的头结点是 ForwardingNode ，则说明这个位置的所有节点已经迁移完成，// 可以继续向前迁移复制其他位置的节点advance = true; // already processedelse {
    // （7）对tab[i]进行迁移，可能是链表 or 红黑树synchronized (f) {
    if (tabAt(tab, i) == f) {
    Node<K,V> ln, hn;if (fh >= 0) {
    // 链表int runBit = fh & n;Node<K,V> lastRun = f;// lastRun并不是一条链表的最后一个，一条链表的节点可以分为两类，// 在循环中寻找lastRun的满足条件是链表中最后一个与前一个节点runBit不相等的节点作为lastRun，// 而此时lastRun后面可能还有节点，但runBit都是和lastRun相等的节点。// 这里找lastRun和java7是一样的for (Node<K,V> p = f.next; p != null; p = p.next) {
    // 计算p的位置int b = p.hash & n;if (b != runBit) {
    // 和runBit不是同一位置runBit = b;lastRun = p;}}// hash & n=0为低位节点，hash & n!=0为高位节点。// 判断找到的lastRun是低位节点还是高位节点if (runBit == 0) {
    ln = lastRun;hn = null;}else {
    hn = lastRun;ln = null;}// lastRun之前的结点因为fh&n不确定，所以全部需要再hash分配。for (Node<K,V> p = f; p != lastRun; p = p.next) {
    int ph = p.hash; K pk = p.key; V pv = p.val;if ((ph & n) == 0)ln = new Node<K,V>(ph, pk, pv, ln);elsehn = new Node<K,V>(ph, pk, pv, hn);}setTabAt(nextTab, i, ln);setTabAt(nextTab, i + n, hn);setTabAt(tab, i, fwd);advance = true;}else if (f instanceof TreeBin) {
    // 是红黑树，// 原理上和链表迁移的过程差不多，也是将节点分成高位节点和低位节点TreeBin<K,V> t = (TreeBin<K,V>)f;// lo低位树头节点，loTail低位树尾节点// hi高位树头节点，hiTail高位树尾节点TreeNode<K,V> lo = null, loTail = null;TreeNode<K,V> hi = null, hiTail = null;int lc = 0, hc = 0;for (Node<K,V> e = t.first; e != null; e = e.next) {
    int h = e.hash;TreeNode<K,V> p = new TreeNode<K,V>(h, e.key, e.val, null, null);if ((h & n) == 0) {
    if ((p.prev = loTail) == null)lo = p;elseloTail.next = p;// 尾插法loTail = p;++lc;}else {
    if ((p.prev = hiTail) == null)hi = p;elsehiTail.next = p;hiTail = p;++hc;}}// 低位节点的个数 <= UNTREEIFY_THRESHOLD=6, 则树退为链表// 否则判断是否有高位节点，无，则原先那棵树t就是一棵低位树，直接赋值给ln// 有高位节点，则低位节点重新树化。// 高位节点的判断同理ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :(hc != 0) ? new TreeBin<K,V>(lo) : t;hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :(lc != 0) ? new TreeBin<K,V>(hi) : t;setTabAt(nextTab, i, ln);setTabAt(nextTab, i + n, hn);setTabAt(tab, i, fwd);advance = true;}}}}}
}

9.CHM 的 put 方法源码?

public V put(K key, V value) {
    return putVal(key, value, false);
}final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();// 1. 哈希值高低位扰动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)// 2. tab 为空 初始化 懒汉模式tab = initTable();else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
    // 3. tab不为null，则通过(n - 1) & hash 计算 tab对应索引下标，找到node// node为null说明没有发生hash冲突，cas 设置新节点node到tab的对应位置，成功则结束循环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)// 4. 发现哈希值为MOVED时，// 说明数组正在扩容，帮助扩容，这个节点只可能是ForwardingNodetab = helpTransfer(tab, f);else {
    // 5.正常情况下发生哈希冲突V oldVal = null;synchronized (f) {
    // 再次检查i位置的节点是否还是f// 如果有变动则重新循环if (tabAt(tab, i) == f) {
    if (fh >= 0) {
    // 6. 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)))) {
    // 链表中已经有hash相等且（key地址相等 or key值相等）// 则判断是否需要替换// put onlyIfAbsent=false，新值替换旧值// putIfAbsent onlyIfAbsent=true,新值不替换旧值oldVal = e.val;if (!onlyIfAbsent)e.val = value;break;}// 解决hash冲突的方式// 链表法，新节点放在了链表尾部（尾插法），这里和jdk1.7不一样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) {
    // 7.红黑树Node<K,V> p;binCount = 2;if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,value)) != null) {
    // putTreeVal的返回值是已经存在的节点// p != null 说明 key已经存在，看是否需要替换valueoldVal = p.val;if (!onlyIfAbsent)p.val = value;}}}}if (binCount != 0) {
    // 8. binCount，链表的长度>=8时 可能变为红黑树，也可能是扩容// 数组长度小于64时，是扩容数组if (binCount >= TREEIFY_THRESHOLD)treeifyBin(tab, i);if (oldVal != null)// 若旧值不为null，则说明是替换，不需要后面的addCountreturn oldVal;break;}}}// 9. 元素数量+1addCount(1L, binCount);return null;
}

10.key 和 value 为什么不能为 null?

ConcurrentHashMap 中 key 和 value 为什么不能为 null 呢?下面从 HashMap 和 ConcurrentHashMap 的对比,详细说明.根据 HashMap 的源码可以知道,当 key 为 null 时，计算出的 hash 值为 0，value 放置在第 0 个桶上

//HashMap
static final int hash(Object key) {
    int h;return (key == null) ? 0 : (h = key.hashCode()) ^ (h >>> 16);
}

根据 ConcurrentHashMap 源码可以知道,ConcurrentHashMap 没有像 HashMap 一样先计算 hash,先进行了判断 key 和 value 是否为 null,为 null 会抛出空值针异常,主要是因为存在二义性问题且 ConcurrentHashMap 没法解决

• 可能没有这个 key
• 可能有这个 key，只不过 value 为 null

因为 ConcurrentHashMap 是线程安全的，一般使用在并发环境下，你一开始 get 方法获取到 null 之后，再去调用 containsKey 方法，没法确保 get 方法和 containsKey 方法之间，没有别的线程来捣乱，刚好把你要查询的对象设置了进去或者删除掉了。

//ConcurrentHashMap
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;......
}

11.jdk1.8 中 CHM 最大并发量

在 JDK1.8 中，ConcurrentHashMap 的并发性能得到了大幅提升，最大支持的并发量也有所增加。根据官方文档的描述，JDK1.8 中的 ConcurrentHashMap 最大可以同时支持并发读写操作的线程数是理论上的最大值，即 CPU 核心数乘以 2。

觉得有用的话点个赞 ?? 呗。
??????本人水平有限，如有纰漏，欢迎各位大佬评论批评指正！???

???如果觉得这篇文对你有帮助的话，也请给个点赞、收藏下吧，非常感谢!? ? ?

???Stay Hungry Stay Foolish 道阻且长,行则将至,让我们一起加油吧！???