并发编程之 ConcurrentHashMap

ConcurrentHashMap 是线程安全且高效的 HashMap。

HashMap 概述

HashMap 是基于哈希表的 Map 接口的非同步实现。此实现提供所有可选的映射操作，并允许使用 null 值和 null 键（除了不同步和允许使用 null 之外，HashMap 类与 Hashtable 大致相同）。此类不保证映射的顺序，特别是它不保证该顺序恒久不变。

值得注意的是 HashMap 不是线程安全的，如果想要线程安全的 HashMap，可以通过 Collections 类的静态方法 synchronizedMap 获得线程安全的 HashMap。

Map map = Collections.synchronizedMap(new HashMap());

HashMap 的数据结构

HashMap 的底层主要是基于数组和链表来实现的，它之所以有相当快的查询速度主要是因为它是通过计算散列码来决定存储的位置。HashMap 中主要是通过 key 的 hashCode 来计算 hash 值的，只要 hashCode 相同，计算出来的 hash 值就一样。如果存储的对象对多了，就有可能不同的对象所算出来的 hash 值是相同的，这就出现了所谓的 hash 冲突。学过数据结构的同学都知道，解决 hash 冲突的方法有很多，HashMap 底层是通过链表来解决 hash 冲突的。也就是说，其链表结果主要是用来解决 hash 冲突的。

hashmap 结构：哈希表是由数组+链表组成的，数组的默认长度为 16（可以自动变长。在构造 HashMap 的时候也可以指定一个长度），数组里每个元素存储的是一个链表的头结点。

JDK1.8 中：

• 使用一个 Node 数组来存储数据，但这个 Node 可能是链表结构，也可能是红黑树结构
• 如果插入的 key 的 hashcode 相同，那么这些 key 也会被定位到 Node 数组的同一个格子里。
• 如果同一个格子里的 key 不超过 8 个，使用链表结构存储。
• 如果超过了 8 个，那么会调用 treeifyBin 函数，将链表转换为红黑树。
• 那么即使 hashcode 完全相同，由于红黑树的特定，查找某个特定元素，也只需要 O(log n)的开销
• 也就是说 put/get 的操作的时间复杂度只有 O(log n)

备注：当数组大小已经超过 64 并且链表中的元素个数超过默认设定（8 个）时，将链表转化为红黑树

为什么要使用 ConcurrentHashMap？

在并发编程中使用 HashMap 可能导致程序死循环。而使用线程安全的 HashTable 效率又非常低下，基于以上两个原因，便有了 ConcurrentHashMap 的登场机会

1.7 中 HashMap 死循环

在多线程环境下，使用 HashMap 进行 put 操作会引起死循环，导致 CPU 利 用率接近 100%，HashMap 在并发执行 put 操作时会引起死循环，是因为多线程会导致 HashMap 的 Entry 链表形成环形数据结构，一旦形成环形数据结构，Entry 的 next 节点永远不为空，就会产生死循环获取 Entry。

HashMap 一次扩容的过程及发生死循环的原因

1、取当前 table 的 2 倍作为新 table 的大小 2、根据算出的新 table 的大小 new 出一个新的 Entry 数组来，名为 newTable 3、轮询原 table 的每一个位置，将每个位置上连接的 Entry，算出在新 table 上的位置，并以链表形式连接 4、原 table 上的所有 Entry 全部轮询完毕之后，意味着原 table 上面的所有 Entry 已经移到了新的 table 上，HashMap 中的 table 指向 newTable

总结：

HashMap 之所以在并发下的扩容造成死循环，是因为，多个线程并发进行 时，因为一个线程先期完成了扩容，将原 Map 的链表重新散列到自己的表中，并且链表变成了倒序，后一个线程再扩容时，又进行自己的散列，再次将倒序链表变为正序链表。于是形成了一个环形链表，当 get 表中不存在的元素时，造成死循环。

线程不安全的 HashMap

在多线程环境下，使用 HashMap 进行 put 操作会引起死循环，导致 CPU 利用率接近 100%，所以在并发情况下不能使用 HashMap。HashMap 在并发执行 put 操作时会引起死循环，是因为多线程会导致 HashMap 的 Entry 链表形成环形数据结构，一旦形成环形数据结构，Entry 的 next 节点永远不为空，就会产生死循环获取 Entry。

效率低下的 HashTable

HashTable 容器使用 synchronized 来保证线程安全，但在线程竞争激烈的情况下 HashTable 的效率非常低下。因为当一个线程访问 HashTable 的同步方法，其他线程也访问 HashTable 的同步方法时，会进入阻塞或轮询状态。如线程 1 使用 put 进行元素添加，线程 2 不但不能使用 put 方法添加元素，也不能使用 get 方法来获取元素，所以竞争越激烈效率越低。

ConcurrentHashMap 的锁分段技术可有效提升并发访问率

HashTable 容器在竞争激烈的并发环境下表现出效率低下的原因是所有访问 HashTable 的线程都必须竞争同一把锁，假如容器里有多把锁，每一把锁用于锁容器其中一部分数据，那么当多线程访问容器里不同数据段的数据时，线程间就不会存在锁竞争，从而可以有效提高并发访问效率，这就是 ConcurrentHashMap 所使用的锁分段技术。首先将数据分成一段一段地存储，然后给每一段数据配一把锁，当一个线程占用锁访问其中一个段数据的时候，其他段的数据也能被其他线程访问。

ConcurrentHashMap 的结构

jdk 1.7 中采用 Segment + HashEntry 的方式进行实现，java 8 取消了分段锁的设计，采用数组 + 单向链表 + 红黑树。

JDK 1.7 的实现方式

jdk 1.7 中，ConcurrentHashMap 是由 Segment 数组结构和 HashEntry 数组结构组成。Segment 是一种可重入锁（ReentrantLock），在 ConcurrentHashMap 里扮演锁的角色；HashEntry 则用于存储键值对数据。一个 ConcurrentHashMap 里包含一个 Segment 数组。Segment 的结构和 HashMap 类似，是一种数组和链表结构。一个 Segment 里包含一个 HashEntry 数组，每个 HashEntry 是一个链表结构的元素，每个 Segment 守护着一个 HashEntry 数组里的元素，当对 HashEntry 数组的数据进行修改时，必须首先获得与它对应的 Segment 锁。

ConcurrentHashMap 初始化时，计算出 Segment 数组的大小 size 和每个 Segment 中 HashEntry 数组的大小 cap，并初始化 Segment 数组的第一个元素；其中 size 大小为 2 的幂次方，默认为 16，cap 大小也是 2 的幂次方，最小值为 2，最终结果根据根据初始化容量 initialCapacity 进行计算，其中 Segment 在实现上继承了 ReentrantLock，这样就自带了锁的功能。

put 实现

当执行 put 方法插入数据时，根据 key 的 hash 值，在 Segment 数组中找到相应的位置，如果相应位置的 Segment 还未初始化，则通过 CAS 进行赋值，接着执行 Segment 对象的 put 方法通过加锁机制插入数据。

1. 线程 A 执行 tryLock()方法成功获取锁，则把 HashEntry 对象插入到相应的位置；
2. 线程 B 获取锁失败，则执行 scanAndLockForPut()方法，在 scanAndLockForPut 方法中，会通过重复执行 tryLock()方法尝试获取锁，在多处理器环境下，重复次数为 64，单处理器重复次数为 1，当执行 tryLock()方法的次数超过上限时，则执行 lock()方法挂起线程 B；
3. 当线程 A 执行完插入操作时，会通过 unlock()方法释放锁，接着唤醒线程 B 继续执行；

size 实现

因为 ConcurrentHashMap 是可以并发插入数据的，所以在准确计算元素时存在一定的难度，一般的思路是统计每个 Segment 对象中的元素个数，然后进行累加，但是这种方式计算出来的结果并不一样的准确的，因为在计算后面几个 Segment 的元素个数时，已经计算过的 Segment 同时可能有数据的插入或则删除，在 1.7 的实现中，采用了如下方式：

先采用不加锁的方式，连续计算元素的个数，最多计算 3 次：

1. 如果前后两次计算结果相同，则说明计算出来的元素个数是准确的；
2. 如果前后两次计算结果都不同，则给每个 Segment 进行加锁，再计算一次元素的个数；

JDK 1.8 的实现方式

ConcurrentHashMap 在 1.8 中的实现，相比于 1.7 的版本基本上全部都变掉了。首先，取消了 Segment 分段锁的数据结构，取而代之的是数组+链表（红黑树）的结构。而对于锁的粒度，调整为对每个数组元素加锁（Node）。然后是定位节点的 hash 算法被简化了，这样带来的弊端是 Hash 冲突会加剧。因此在链表节点数量大于 8 时，会将链表转化为红黑树进行存储。这样一来，查询的时间复杂度就会由原先的 O(n)变为 O(logN)。

1.8 中放弃了 Segment 臃肿的设计，取而代之的是采用 Node + CAS + Synchronized 来保证并发安全进行实现，

只有在执行第一次 put 方法时才会调用 initTable()初始化 Node 数组。

put 实现

当执行 put 方法插入数据时，根据 key 的 hash 值，在 Node 数组中找到相应的位置，实现如下：

1. 如果相应位置的 Node 还未初始化，则通过 CAS 插入相应的数据；
2. 如果相应位置的 Node 不为空，且当前该节点不处于移动状态，则对该节点加 synchronized 锁，如果该节点的 hash 不小于 0，则遍历链表更新节点或插入新节点；
3. 如果该节点是 TreeBin 类型的节点，说明是红黑树结构，则通过 putTreeVal 方法往红黑树中插入节点；
4. 如果 baseCount 不为 0，说明 put 操作对数据产生了影响，如果当前链表的个数达到 8 个，则通过 treeifyBin 方法转化为红黑树，如果 oldVal 不为空，说明是一次更新操作，没有对元素个数产生影响，则直接返回旧值；
5. 如果插入的是一个新节点，则执行 addCount()方法尝试更新元素个数 baseCount；

size 实现

1.8 中使用一个 volatile 类型的变量 baseCount 记录元素的个数，当插入新数据或则删除数据时，会通过 addCount()方法更新 baseCount

1. 初始化时 counterCells 为空，在并发量很高时，如果存在两个线程同时执行 CAS 修改 baseCount 值，则失败的线程会继续执行方法体中的逻辑，使用 CounterCell 记录元素个数的变化；
2. 如果 CounterCell 数组 counterCells 为空，调用 fullAddCount()方法进行初始化，并插入对应的记录数，通过 CAS 设置 cellsBusy 字段，只有设置成功的线程才能初始化 CounterCell 数组
3. 如果通过 CAS 设置 cellsBusy 字段失败的话，则继续尝试通过 CAS 修改 baseCount 字段，如果修改 baseCount 字段成功的话，就退出循环，否则继续循环插入 CounterCell 对象；所以在 1.8 中的 size 实现比 1.7 简单多，因为元素个数保存 baseCount 中，部分元素的变化个数保存在 CounterCell 数组中
4. 通过累加 baseCount 和 CounterCell 数组中的数量，即可得到元素的总个数；