ConcurrentHashMap

HashTable是一个线程安全的类，它使用synchronized对put、get等方法进行加锁，故它加锁的对象是当前HashTable的实例，相当于锁住了整张哈希表。在多线程并发的情况下，如果一个线程通过get或者put等方法获取到了这个HashTable锁，那么意味在上一个线程释放锁前其他线程不能再对哈希表进行get、put等操作，这个等待锁的线程只能被迫阻塞，导致HashTable的存取效率非常低下。

ConcurrentHashMap可以做到读取数据不加锁，并且其内部的结构可以让其在进行写操作的时候能够将锁的粒度保持地尽量地小，允许多个修改操作并发进行，其关键在于使用了锁分段技术（JDK 1.7）。它使用了多个锁来控制对hash表的不同部分进行的修改。对于JDK1.7版本的实现, ConcurrentHashMap内部使用段(Segment)来表示这些不同的部分，每个段其实就是一个小的Hashtable，它们有自己的锁。只要多个修改操作发生在不同的段上，它们就可以并发进行。JDK1.8的实现降低锁的粒度，JDK1.7版本锁的粒度是基于Segment的，包含多个HashEntry，而JDK1.8锁的粒度就是HashEntry（首节点）。

JDK 1.7中的 CurrentHashMap

整个CurrentHashMap 由一个个Segment组成，segment 意为”部分“或者”一段“的意思，所以很多时候被称为分段锁。简单理解CurrentHashMap是一个Segment数组，Segment通过继承ReentrantLock来进行加锁，所以每次需要加锁的操作锁住的是一个Segment，这样保证了每个Segment是线程安全的，同时也实现了全局的线程安全。

1. CurrentHashMap 的构造方法

ConcurrentHashMap的构造方法如下：

staticfinalintDEFAULT_INITIAL_CAPACITY = 16;
staticfinalfloatDEFAULT_LOAD_FACTOR = 0.75f;
staticfinalintDEFAULT_CONCURRENCY_LEVEL = 16;
// 无参构造方法publicConcurrentHashMap() {
// 初始化容量16，装载因子0.75，默认并发级别16this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
}
publicConcurrentHashMap(intinitialCapacity,
floatloadFactor, intconcurrencyLevel) {
if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
thrownewIllegalArgumentException();
if (concurrencyLevel > MAX_SEGMENTS)
concurrencyLevel = MAX_SEGMENTS;
// Find power-of-two sizes best matching argumentsintsshift = 0;
intssize = 1;
// 计算并行级别 ssize，因为要保持并行级别是 2 的 n 次方while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
// 我们这里先不要那么烧脑，用默认值，concurrencyLevel 为 16，sshift 为 4// 那么计算出 segmentShift 为 28，segmentMask 为 15，后面会用到这两个值this.segmentShift = 32 - sshift;
this.segmentMask = ssize - 1;
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
// initialCapacity 是设置整个 map 初始的大小，// 这里根据 initialCapacity 计算 Segment 数组中每个位置可以分到的大小// 如 initialCapacity 为 64，那么每个 Segment 或称之为"槽"可以分到 4 个intc = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
// 默认 MIN_SEGMENT_TABLE_CAPACITY 是 2，这个值也是有讲究的，因为这样的话，对于具体的槽上，// 插入一个元素不至于扩容，插入第二个的时候才会扩容intcap = MIN_SEGMENT_TABLE_CAPACITY; while (cap < c)
cap <<= 1;
// 创建 Segment 数组，// 并创建数组的第一个元素 segment[0]Segment<K,V> s0 =
newSegment<K,V>(loadFactor, (int)(cap * loadFactor),
(HashEntry<K,V>[])newHashEntry[cap]);
Segment<K,V>[] ss = (Segment<K,V>[])newSegment[ssize];
// 往数组写入 segment[0]UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]this.segments = ss;
} 

ConcurrentHashMap 有多个构造方法，这里给出一个无参构造方法和一个3参数构造方法，所有构造方法最终都会调用这个三参数的构造方法。其中构造方法中的三个参数代表的含义如下：

• initialCapacity 表示初始容量，指整个 ConcurrentHashMap 的初始化容量，实际操作的时候回分配给每个Segment。
• loadFactor 表示负载因子，由于Segment数组不可扩容，所以这里的负载因子是指每个Segment的负载因子。
• concurrencyLevel （或称作并行级别、并发数、Segment数）默认是16，也就是说ConcurrentHashMap 有 16 个Segment。所以理论上只要线程不在同一个Segment上，最多可以同时支持16个线程并发存取。这个值可以在初始化的时候设置，但是一旦初始化以后由于它是不可被扩容的所以便不能被更改。

2. put过程分析

publicVput(Kkey, Vvalue) {
Segment<K,V> s;
if (value == null)
thrownewNullPointerException();
// 1. 计算 key 的 hash 值inthash = hash(key);
// 2. 根据 hash 值找到 Segment 数组中的位置 j// hash 是 32 位，无符号右移 segmentShift(28) 位，剩下高 4 位，// 然后和 segmentMask(15) 做一次与操作，也就是说 j 是 hash 值的高 4 位，也就是Segment的数组下标intj = (hash >>> segmentShift) & segmentMask;
// 刚刚说了，初始化的时候初始化了 segment[0]，但是其他位置还是 null，// ensureSegment(j) 对 segment[j] 进行初始化if ((s = (Segment<K,V>)UNSAFE.getObject// nonvolatile; recheck
(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegments = ensureSegment(j);
// 3. 插入新值到 Segment s 中returns.put(key, hash, value, false);
}

上边的代码根据 hash 值很快就能找到相应的 Segment，之后就是 Segment 内部的 put 操作了。

Segment 内部是一个类似HashMap的结构，由 数组+链表 组成的。

finalVput(Kkey, inthash, Vvalue, booleanonlyIfAbsent) {
// 在往该 segment 写入前，需要先获取该 segment 的独占锁// 先看主流程，后面还会具体介绍这部分内容HashEntry<K,V> node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
VoldValue;
try {
// 这个是 segment 内部的数组HashEntry<K,V>[] tab = table;
// 再利用 hash 值，求应该放置的数组下标intindex = (tab.length - 1) & hash;
// first 是数组该位置处的链表的表头HashEntry<K,V> first = entryAt(tab, index);
// 下面这串 for 循环虽然很长，不过也很好理解，想想该位置没有任何元素和已经存在一个链表这两种情况for (HashEntry<K,V> e = first;;) {
if (e != null) {
Kk;
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 到底是不是 null，这个要看获取锁的过程，不过和这里都没有关系。// 如果不为 null，那就直接将它设置为链表表头；如果是null，初始化并设置为链表表头。if (node != null)
node.setNext(first);
elsenode = newHashEntry<K,V>(hash, key, value, first);
intc = count + 1;
// 如果超过了该 segment 的阈值，这个 segment 需要扩容if (c > threshold && tab.length < MAXIMUM_CAPACITY)
rehash(node); // 扩容后面也会具体分析else// 没有达到阈值，将 node 放到数组 tab 的 index 位置，// 其实就是将新的节点设置成原链表的表头setEntryAt(tab, index, node);
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
// 解锁unlock();
}
returnoldValue;
}

由于有独占锁的保护，所以 segment 内部的操作并不复杂。至于这里面的并发问题，我们稍后再进行介绍。

到这里 put 操作就结束了，接下来，我们说一说其中几步关键的操作。

（1）ensureSegment 初始化Segment

ConcurrentHashMap 的构造方法中会初始化第一个Segment 段 segment[0],其他Segment段则是在插入第一个值的时候进行初始化。这里需要考虑并发，因为可能会有多个线程同时进来初始化同一个Segment[k]，不过只要有一个成功就可以了。

privateSegment<K,V> ensureSegment(intk) {
finalSegment<K,V>[] ss = this.segments;
longu = (k << SSHIFT) + SBASE; // raw offsetSegment<K,V> seg;
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
// 这里看到为什么之前要初始化 segment[0] 了，// 使用当前 segment[0] 处的数组长度和负载因子来初始化 segment[k]// 为什么要用“当前”，因为 segment[0] 可能早就扩容过了Segment<K,V> proto = ss[0];
intcap = proto.table.length;
floatlf = proto.loadFactor;
intthreshold = (int)(cap * lf);
// 初始化 segment[k] 内部的数组HashEntry<K,V>[] tab = (HashEntry<K,V>[])newHashEntry[cap];
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) { // 再次检查一遍该槽是否被其他线程初始化了。Segment<K,V> s = newSegment<K,V>(lf, threshold, tab);
// 使用 while 循环，内部用 CAS，当前线程成功设值或其他线程成功设值后，退出while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) {
if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
break;
}
}
}
returnseg;
}

总的来说，ensureSegment(int k) 比较简单，对于并发操作使用 CAS 进行控制。

（2）scanAndLockForPut 获取写入锁

前面我们看到，在往某个 segment 中 put 的时候，首先会调用 node = tryLock() ? null : scanAndLockForPut(key, hash, value)，也就是说先进行一次 tryLock() 快速获取该 segment 的独占锁，如果失败，那么进入到 scanAndLockForPut 这个方法来获取锁。

下面我们来具体分析这个方法中是怎么控制加锁的。

privateHashEntry<K,V> scanAndLockForPut(Kkey, inthash, Vvalue) {
HashEntry<K,V> first = entryForHash(this, hash);
HashEntry<K,V> e = first;
HashEntry<K,V> node = null;
intretries = -1; // negative while locating node// 循环获取锁while (!tryLock()) {
HashEntry<K,V> f; // to recheck first belowif (retries < 0) {
if (e == null) {
if (node == null) // speculatively create node// 进到这里说明数组该位置的链表是空的，没有任何元素// 当然，进到这里的另一个原因是 tryLock() 失败，所以该槽存在并发，不一定是该位置node = newHashEntry<K,V>(hash, key, value, null);
retries = 0;
}
elseif (key.equals(e.key))
retries = 0;
else// 顺着链表往下走e = e.next;
}
// 重试次数如果超过 MAX_SCAN_RETRIES(单核1多核64)，那么不抢了，进入到阻塞队列等待锁// lock() 是阻塞方法，直到获取锁后返回elseif (++retries > MAX_SCAN_RETRIES) {
lock();
break;
}
elseif ((retries & 1) == 0 &&
// 这个时候是有大问题了，那就是有新的元素进到了链表，成为了新的表头// 所以这边的策略是，相当于重新走一遍这个 scanAndLockForPut 方法
(f = entryForHash(this, hash)) != first) {
e = first = f; // re-traverse if entry changedretries = -1;
}
}
returnnode;
}

这个方法有两个出口，一个是 tryLock() 成功了，循环终止，另一个就是重试次数超过了 MAX_SCAN_RETRIES，进到 lock() 方法，此方法会阻塞等待，直到成功拿到独占锁。

这个方法就是看似复杂，但是其实就是做了一件事，那就是获取该 segment 的独占锁，如果需要的话顺便实例化了一下 node。

（3）扩容与Rehash

重复一下，segment 数组不能扩容，扩容是 segment 数组某个位置内部的数组 HashEntry<K,V>[] 进行扩容，扩容后，容量为原来的 2 倍。

首先，我们要回顾一下触发扩容的地方，put 的时候，如果判断该值的插入会导致该 segment 的元素个数超过阈值，那么先进行扩容，再插值，读者这个时候可以回去 put 方法看一眼。

该方法不需要考虑并发，因为到这里的时候，是持有该 segment 的独占锁的。

// 方法参数上的 node 是这次扩容后，需要添加到新的数组中的数据。privatevoidrehash(HashEntry<K,V> node) {
HashEntry<K,V>[] oldTable = table;
intoldCapacity = oldTable.length;
// 2 倍intnewCapacity = oldCapacity << 1;
threshold = (int)(newCapacity * loadFactor);
// 创建新数组HashEntry<K,V>[] newTable =
(HashEntry<K,V>[]) newHashEntry[newCapacity];
// 新的掩码，如从 16 扩容到 32，那么 sizeMask 为 31，对应二进制 ‘000...00011111’intsizeMask = newCapacity - 1;
// 遍历原数组，老套路，将原数组位置 i 处的链表拆分到 新数组位置 i 和 i+oldCap 两个位置for (inti = 0; i < oldCapacity ; i++) {
// e 是链表的第一个元素HashEntry<K,V> e = oldTable[i];
if (e != null) {
HashEntry<K,V> next = e.next;
// 计算应该放置在新数组中的位置，// 假设原数组长度为 16，e 在 oldTable[3] 处，那么 idx 只可能是 3 或者是 3 + 16 = 19intidx = e.hash & sizeMask;
if (next == null) // 该位置处只有一个元素，那比较好办newTable[idx] = e;
else { // Reuse consecutive sequence at same slot// e 是链表表头HashEntry<K,V> lastRun = e;
// idx 是当前链表的头结点 e 的新位置intlastIdx = idx;
// 下面这个 for 循环会找到一个 lastRun 节点，这个节点之后的所有元素是将要放到一起的for (HashEntry<K,V> last = next;
last != null;
last = last.next) {
intk = last.hash & sizeMask;
if (k != lastIdx) {
lastIdx = k;
lastRun = last;
}
}
// 将 lastRun 及其之后的所有节点组成的这个链表放到 lastIdx 这个位置newTable[lastIdx] = lastRun;
// 下面的操作是处理 lastRun 之前的节点，// 这些节点可能分配在另一个链表中，也可能分配到上面的那个链表中for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
Vv = p.value;
inth = p.hash;
intk = h & sizeMask;
HashEntry<K,V> n = newTable[k];
newTable[k] = newHashEntry<K,V>(h, p.key, v, n);
}
}
}
}
// 将新来的 node 放到新数组中刚刚的 两个链表之一 的 头部intnodeIndex = node.hash & sizeMask; // add the new nodenode.setNext(newTable[nodeIndex]);
newTable[nodeIndex] = node;
table = newTable;
}

这里的扩容比之前的 HashMap 要复杂一些，代码难懂一点。上面有两个挨着的 for 循环，第一个 for 有什么用呢?

仔细一看发现，如果没有第一个 for 循环，也是可以工作的，但是，这个 for 循环下来，如果 lastRun 的后面还有比较多的节点，那么这次就是值得的。因为我们只需要克隆 lastRun 前面的节点，后面的一串节点跟着 lastRun 走就是了，不需要做任何操作。

我觉得 Doug Lea 的这个想法也是挺有意思的，不过比较坏的情况就是每次 lastRun 都是链表的最后一个元素或者很靠后的元素，那么这次遍历就有点浪费了。不过 Doug Lea 也说了，根据统计，如果使用默认的阈值，大约只有 1/6 的节点需要克隆。

3. get 过程分析

相对于 put 来说，get 就很简单了。

• 计算 hash 值，找到 segment 数组中的具体位置，或我们前面用的“槽”
• 槽中也是一个数组，根据 hash 找到数组中具体的位置
• 到这里是链表了，顺着链表进行查找即可

publicVget(Objectkey) {
Segment<K,V> s; // manually integrate access methods to reduce overheadHashEntry<K,V>[] tab;
// 1. hash 值inth = hash(key);
longu = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
// 2. 根据 hash 找到对应的 segmentif ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
(tab = s.table) != null) {
// 3. 找到segment 内部数组相应位置的链表，遍历for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
(tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
e != null; e = e.next) {
Kk;
if ((k = e.key) == key || (e.hash == h && key.equals(k)))
returne.value;
}
}
returnnull;
}

4. 并发问题分析

现在我们已经说完了 put 过程和 get 过程，我们可以看到 get 过程中是没有加锁的，那自然我们就需要去考虑并发问题。

添加节点的操作 put 和删除节点的操作 remove 都是要加 segment 上的独占锁的，所以它们之间自然不会有问题，我们需要考虑的问题就是 get 的时候在同一个 segment 中发生了 put 或 remove 操作。

• put 操作的线程安全性。
  • 初始化槽，这个我们之前就说过了，使用了 CAS 来初始化 Segment 中的数组。
  • 添加节点到链表的操作是插入到表头的，所以，如果这个时候 get 操作在链表遍历的过程已经到了中间，是不会影响的。当然，另一个并发问题就是 get 操作在 put 之后，需要保证刚刚插入表头的节点被读取，这个依赖于 setEntryAt 方法中使用的 UNSAFE.putOrderedObject。
  • 扩容。扩容是新创建了数组，然后进行迁移数据，最后面将 newTable 设置给属性 table。所以，如果 get 操作此时也在进行，那么也没关系，如果 get 先行，那么就是在旧的 table 上做查询操作；而 put 先行，那么 put 操作的可见性保证就是 table 使用了 volatile 关键字。
• remove 操作的线程安全性。
  • remove 操作我们没有分析源码，所以这里说的读者感兴趣的话还是需要到源码中去求实一下的。
  • get 操作需要遍历链表，但是 remove 操作会"破坏"链表。
  • 如果 remove 破坏的节点 get 操作已经过去了，那么这里不存在任何问题。
  • 如果 remove 先破坏了一个节点，分两种情况考虑。 1、如果此节点是头结点，那么需要将头结点的 next 设置为数组该位置的元素，table 虽然使用了 volatile 修饰，但是 volatile 并不能提供数组内部操作的可见性保证，所以源码中使用了 UNSAFE 来操作数组，请看方法 setEntryAt。2、如果要删除的节点不是头结点，它会将要删除节点的后继节点接到前驱节点中，这里的并发保证就是 next 属性是 volatile 的。

二、JDK 1.8 中的 ConcurrentHashMap

在JDK1.7之前，ConcurrentHashMap是通过分段锁机制来实现的，所以其最大并发度受Segment的个数限制。因此，在JDK1.8中，ConcurrentHashMap的实现原理摒弃了这种设计，而是选择了与HashMap类似的数组+链表+红黑树的方式实现，而加锁则采用CAS和synchronized实现。

结构上和 Java8 的 HashMap 基本上一样，不过它要保证线程安全性，所以在源码上确实要复杂一些。

来源：https://www.pdai.tech/md/java/thread/java-thread-x-juc-collection-ConcurrentHashMap.html