【Java 并发笔记】ConcurrentHashMap（1.7
2019-01-06 本文已影响4人
58bc06151329
文前说明
作为码农中的一员，需要不断的学习，我工作之余将一些分析总结和学习笔记写成博客与大家一起交流，也希望采用这种方式记录自己的学习之旅。
本文仅供学习交流使用，侵权必删。
不用于商业目的，转载请注明出处。
1. ConcurrentHashMap
- ConcurrentHashMap 是线程安全且高效的 HashMap。
- 多线程环境下，使用 HashMap 进行 put 操作会引起死循环，如下例，导致 CPU 利用率接近 100%，所以在并发情况下不能使用 HashMap。
public static void main(String[] arg) throws InterruptedException {
final HashMap<String, String> map = new HashMap<String, String>();
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
for (int i = 0; i < 10000; i++) {
new Thread(new Runnable() {
@Override
public void run() {
map.put(UUID.randomUUID().toString(), "");
}
}, "thread" + i).start();
}
}
}, "threadMain");
thread.start();
thread.join();
}
- 因为多线程会导致 HashMap 的 Entry 链表形成环形数据结构。
- Entry 的 next 结点永远不为空，产生死循环获取 Entry。
1.1 HaspMap（JDK 1.7）
JDK 1.7 HashMap
- HashMap 里面是一个数组（table），数组中每个元素（Entry）是一个单向链表。
- 是基于哈希表的 Map 接口的实现，以 Key-Value 的形式存在，即存储的对象是 Entry （同时包含了 Key 和 Value）。
- 根据 hash 算法 来计算 Key-Value 的存储位置并进行快速存取。
- 最多只允许一条 Entry 的键为 Null（多条覆盖），但允许多条 Entry 的值为 Null。
- 是 Map 的一个非同步的实现。
| 术语 | 英文 | 解释 |
|---|---|---|
| hash 算法 | hash algorithm | 是一种将任意内容的输入转换成相同长度输出的加密方式，其输出被称为 hash（哈希）值。 |
| hash 表 | hash table | 根据设定的哈希函数 H(key) 和处理冲突方法将一组关键字映象到一个有限的地址区间上，并以关键字在地址区间中的象作为记录在表中的存储位置，这种表称为 hash 表或散列，所得存储位置称为 hash 地址或散列地址。 |
| 属性 | 说明 |
|---|---|
| DEFAULT_INITIAL_CAPACITY | 初始化桶（数组）大小，始终保持 2^n，可以扩容，扩容后数组大小为当前的 2 倍。默认值 1 << 4（2^4，16）。 |
| MAXIMUM_CAPACITY | 桶最大值 1 << 30（2^30，1073741824）。 |
| DEFAULT_LOAD_FACTOR | 默认的负载因子（0.75）。 |
| Entry<K,V>[] table | HaspMap 中的数组。根据需要可调整大小，长度必须是 2^n。 |
| size | 映射中包含的键值映射的数量。 |
| loadFactor | 负载因子，可在初始化时显式指定。用于衡量的是一个散列表的空间使用程度。负载因子越大，对空间的利用越充分，查找效率越低；若负载因子越小，数据越稀疏，对空间造成的浪费越严重。 |
| threshold | 扩容的阈值，可在初始化时显式指定。值等于 HashMap 的容量乘以负载因子。 |
public HashMap() {
this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR);
}
public HashMap(int initialCapacity, float loadFactor) {
if (initialCapacity < 0)
throw new IllegalArgumentException("Illegal initial capacity: " + initialCapacity);
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
if (loadFactor <= 0 || Float.isNaN(loadFactor))
throw new IllegalArgumentException("Illegal load factor: " + loadFactor);
this.loadFactor = loadFactor;
threshold = initialCapacity;
init();
}
- 默认容量为 2^4（16），负载因子为 0.75。
- Map 在使用过程中不断存放数据，当数量达到了 16 * 0.75 = 12 就需要将当前 16 的容量进行扩容，而扩容这个过程涉及到 rehash、复制数据等操作，非常消耗性能。
- 能提前预估 HashMap 的大小最好，给定一个大小，尽量的减少扩容带来的性能损耗。
transient Entry<K,V>[] table = (Entry<K,V>[]) EMPTY_TABLE;
- Entry 是 HashMap 中的一个内部类。
| Entry 属性 | 说明 |
|---|---|
| key | 写入时的键。 |
| value | 值 |
| next | 下一个结点对象，用于实现链表结构。 |
| hash | 当前 key 的 hashcode。 |
put 方法过程
public V put(K key, V value) {
// 当插入第一个元素的时候，需要先初始化数组大小
if (table == EMPTY_TABLE) {
inflateTable(threshold);
}
// 如果 key 为 null，将 entry 放到 table[0] 中。
if (key == null)
return putForNullKey(value);
// 1. 求出 key 的 hash 值
int hash = hash(key);
// 2. 找到对应的数组下标
int i = indexFor(hash, table.length);
// 3. 遍历一下对应下标处的链表，看是否有重复的 key 已经存在，
// 如果有，直接覆盖，put 方法返回旧值就结束。
for (Entry<K,V> e = table[i]; e != null; e = e.next) {
Object k;
if (e.hash == hash && ((k = e.key) == key || key.equals(k))) {
V oldValue = e.value;
e.value = value;
e.recordAccess(this);
return oldValue;
}
}
modCount++;
// 4. 不存在重复的 key，将此 entry 添加到链表中。
addEntry(hash, key, value, i);
return null;
}
- put 方法的流程。
- 判断当前数组是否需要初始化。
- 如果 key 为空，则 put 一个空值进去。
- 根据 key 计算出 hashcode。
- 根据计算出的 hashcode 定位出所在链表。
- 遍历判断链表里的 hashcode、key 是否和传入 key 相等，如果相等则进行覆盖，并返回原来的值。
- 如果定位位置为空，说明当前位置没有数据存入，新增一个 Entry 对象写入当前位置。
数组初始化
- 第一个元素插入 HashMap 时做一次数组初始化，先确定初始的数组大小，并计算数组扩容的阈值。
private void inflateTable(int toSize) {
// 保证数组大小一定是 2 的 n 次方。
// 比如初始化：new HashMap(20)，那么处理成初始数组大小是 32。
int capacity = roundUpToPowerOf2(toSize);
// 计算扩容阈值：capacity * loadFactor
threshold = (int) Math.min(capacity * loadFactor, MAXIMUM_CAPACITY + 1);
// 初始化数组
table = new Entry[capacity];
initHashSeedAsNeeded(capacity); //ignore
}
计算具体数组位置
- 使用 key 的 hash 值和数组长度进行与计算，从而计算出所在数组的具体位置。
- 例如数组长度 32，减 1 （11111）与 hash 进行与计算，取的就是 key 的 hash 值的低 5 位，作为在数组中的下标位置。
static int indexFor(int hash, int length) {
// assert Integer.bitCount(length) == 1 : "length must be a non-zero power of 2";
return hash & (length-1);
}
添加结点到链表中
- 找到数组下标后，先进行 key 判重，没有重复，就准备将新值 放入到链表的表头。
- 主要逻辑就是先判断是否需要扩容，需要的话先扩容，然后再将这个新的数据插入到扩容后的数组的相应位置处的链表的表头。
void addEntry(int hash, K key, V value, int bucketIndex) {
// 如果当前 HashMap 大小已经达到了阈值，并且新值要插入的数组位置已经有元素了，那么进行扩容。
if ((size >= threshold) && (null != table[bucketIndex])) {
// 扩容
resize(2 * table.length);
// 扩容后，重新计算 hash 值。
hash = (null != key) ? hash(key) : 0;
// 重新计算扩容后的新下标。
bucketIndex = indexFor(hash, table.length);
}
createEntry(hash, key, value, bucketIndex);
}
// 将新值放到链表的表头，然后 size++
void createEntry(int hash, K key, V value, int bucketIndex) {
Entry<K,V> e = table[bucketIndex];
table[bucketIndex] = new Entry<>(hash, key, value, e);
size++;
}
数组扩容
- 在插入新值的时候，如果当前的 size 已经达到了阈值，并且要插入的数组位置上已经有元素，那么就会触发扩容，扩容后，数组大小为原来的 2 倍。
void resize(int newCapacity) {
Entry[] oldTable = table;
int oldCapacity = oldTable.length;
if (oldCapacity == MAXIMUM_CAPACITY) {
threshold = Integer.MAX_VALUE;
return;
}
// 新的数组
Entry[] newTable = new Entry[newCapacity];
// 将原来数组中的值迁移到新的数组中。
transfer(newTable, initHashSeedAsNeeded(newCapacity));
table = newTable;
threshold = (int)Math.min(newCapacity * loadFactor, MAXIMUM_CAPACITY + 1);
}
get 方法过程
public V get(Object key) {
// key 为 null，会被放到 table[0]，所以只要遍历下 table[0] 处的链表就可以。
if (key == null)
return getForNullKey();
Entry<K,V> entry = getEntry(key);
return null == entry ? null : entry.getValue();
}
- get 方法的流程。
- 根据 key 计算 hash 值。
- 找到相应的数组下标：hash & (length – 1)。
- 遍历该数组位置处的链表，直到找到相等（== 或 equals）的 key。
final Entry<K,V> getEntry(Object key) {
if (size == 0) {
return null;
}
int hash = (key == null) ? 0 : hash(key);
// 确定数组下标，然后从头开始遍历链表，直到找到为止
for (Entry<K,V> e = table[indexFor(hash, table.length)];
e != null;
e = e.next) {
Object k;
if (e.hash == hash &&
((k = e.key) == key || (key != null && key.equals(k))))
return e;
}
return null;
}
1.2 HashTable
- HashTable 容器使用 synchronized 来保证线程安全，但在线程竞争激烈的情况下 HashTable 的效率非常低下。
- 当一个线程访问 HashTable 的同步方法时，其他线程访问 HashTable 的同步方法可能会进入阻塞或轮询状态。
- HashTable 容器在竞争激烈的并发环境下表现出效率低下的原因，是因为所有访问它的线程都必须竞争同一把锁，假如容器里有多把锁，每一把锁用于锁容器其中一部分数据，那么当多线程访问容器里不同数据段的数据时，线程间就不会存在锁竞争，从而可以有效的提高并发访问效率，这就是 ConcurrentHashMap（JDK 1.7） 使用的 锁分段技术。
- ConcurrentHashMap 将数据分成一段一段的存储，然后给每一段数据配一把锁，当一个线程占用锁访问其中一个段数据的时候，其他段的数据也能被其他线程访问。
- 有些方法需要跨段，比如 size() 和 containsValue()，它们可能需要锁定整个表而不仅仅是某个段，这需要按顺序锁定所有段，操作完毕后，又按顺序释放所有段的锁。
- 按顺序 很重要，否则极有可能出现死锁，在 ConcurrentHashMap 内部，段数组是 final 的，并且其成员变量实际也是 final 的，但是，仅仅是将数组声明为 final 的并不保证数组成员也是 final 的，需要实现上的保证。这可以确保不会出现死锁，因为获得锁的顺序是固定的。
- HashTable 的迭代器是强一致性的，而 ConcurrentHashMap 是弱一致的。-
- ConcurrentHashMap 的 get，clear，iterator 方法都是弱一致性的。
1.3 ConcurrentHashMap（JDK 1.7）
JDK 1.7 ConcurrentHashMap
- ConcurrentHashMap 是由 Segment 数组 结构和 HashEntry 数组 结构组成。
- Segment 是一种可重入锁 ReentrantLock，在 ConcurrentHashMap 里扮演锁的角色，HashEntry 则用于存储键值对数据。
- ConcurrentHashMap 里包含一个 Segment 数组，Segment 的结构和 HashMap 类似，一个 Segment 里包含一个 HashEntry 数组，每个 HashEntry 是一个链表结构的元素， 每个 Segment 守护者一个 HashEntry 数组里的元素，当对 HashEntry 数组的数据进行修改时，必须首先获得它对应的 Segment 锁。
| 属性 | 说明 |
|---|---|
| concurrencyLevel | 并发度，程序运行时能够同时更新 ConcurrentHashMap 且不产生锁竞争的最大线程数，分段锁个数，即 Segment[] 的数组长度，默认为 16。用户也可以在构造函数中设置并发度。 |
| initialCapacity | 初始容量，指的是整个 ConcurrentHashMap 的初始容量，实际操作的时候需要平均分给每个 Segment。 |
| loadFactor | 负载因子，Segment 数组不可以扩容，负载因子供每个 Segment 内部使用。 |
- 并发度设置过小，会带来严重的锁竞争问题，设置过大，原本位于同一个 Segment 内的访问会扩散到不同的 Segment 中，CPU cache 命中率会下降，从而引起程序性能下降。
- 和 JDK 1. 6 不同，JDK 1. 7 中除了第一个 Segment 之外，剩余的 Segments 采用的是 延迟初始化 机制：每次 put 之前都需要检查 key 对应的 Segment 是否为 null，如果是则调用
ensureSegment()以确保对应的 Segment 被创建。 -
ensureSegment()可能在并发环境下被调用，但并未使用锁来控制竞争，而是使用了 Unsafe 对象的getObjectVolatile()提供的原子读语义结合 CAS 来确保 Segment 创建的原子性。
private ConcurrentHashMap.Segment<K, V> ensureSegment(int k) {
ConcurrentHashMap.Segment[] ss = this.segments;
long u = (long)(k << SSHIFT) + SBASE;
ConcurrentHashMap.Segment seg;
if((seg = (ConcurrentHashMap.Segment)UNSAFE.getObjectVolatile(ss, u)) == null) {
ConcurrentHashMap.Segment proto = ss[0];
int cap = proto.table.length;
float lf = proto.loadFactor;
int threshold = (int)((float)cap * lf);
ConcurrentHashMap.HashEntry[] tab = (ConcurrentHashMap.HashEntry[])(new ConcurrentHashMap.HashEntry[cap]);
if((seg = (ConcurrentHashMap.Segment)UNSAFE.getObjectVolatile(ss, u)) == null) {
ConcurrentHashMap.Segment s = new ConcurrentHashMap.Segment(lf, threshold, tab);
while((seg = (ConcurrentHashMap.Segment)UNSAFE.getObjectVolatile(ss, u)) == null) {
seg = s;
if(UNSAFE.compareAndSwapObject(ss, u, (Object)null, s)) {
break;
}
}
}
}
return seg;
}
- HashEntry 代表每个 hash 链中的一个结点。
static final class HashEntry<K, V> {
final int hash;
final K key;
volatile V value;
volatile ConcurrentHashMap.HashEntry<K, V> next;
static final Unsafe UNSAFE;
static final long nextOffset;
HashEntry(int hash, K key, V value, ConcurrentHashMap.HashEntry<K, V> next) {
this.hash = hash;
this.key = key;
this.value = value;
this.next = next;
}
final void setNext(ConcurrentHashMap.HashEntry<K, V> n) {
UNSAFE.putOrderedObject(this, nextOffset, n);
}
static {
try {
UNSAFE = Unsafe.getUnsafe();
Class e = ConcurrentHashMap.HashEntry.class;
nextOffset = UNSAFE.objectFieldOffset(e.getDeclaredField("next"));
} catch (Exception var1) {
throw new Error(var1);
}
}
}
| HashEntry 属性 | 说明 |
|---|---|
| key | 写入时的键。 |
| value | 值 |
| next | 下一个结点对象，用于实现链表结构。 |
| hash | 当前 key 的 hashcode。 |
- 为了确保读操作能够看到最新的值，value 设置成 volatile，避免了加锁。
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
if (concurrencyLevel > MAX_SEGMENTS)
concurrencyLevel = MAX_SEGMENTS;
// Find power-of-two sizes best matching arguments
int sshift = 0;
int ssize = 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 个
int c = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
// 默认 MIN_SEGMENT_TABLE_CAPACITY 是 2，这个值也是有讲究的，因为这样的话，对于具体的槽上，
// 插入一个元素不至于扩容，插入第二个的时候才会扩容
int cap = MIN_SEGMENT_TABLE_CAPACITY;
while (cap < c)
cap <<= 1;
// 创建 Segment 数组，
// 并创建数组的第一个元素 segment[0]
Segment<K,V> s0 =
new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
(HashEntry<K,V>[])new HashEntry[cap]);
Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
// 往数组写入 segment[0]
UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
this.segments = ss;
}
- Segment 数组长度为 16，不可以扩容。
- Segment[i] 的默认大小为 2，负载因子是 0.75，得出初始阈值为 1.5，也就是以后插入第一个元素不会触发扩容，插入第二个会进行第一次扩容。
- 只初始化了 segment[0]，其他位置仍然是 null。
- 当前 segmentShift 的值为 32 - 4 = 28，segmentMask 为 16 - 1 = 15，为移位数和掩码。
put 方法过程
public V put(K key, V value) {
Segment<K,V> s;
if (value == null)
throw new NullPointerException();
// 1. 计算 key 的 hash 值
int hash = hash(key);
// 2. 根据 hash 值找到 Segment 数组中的位置 j
// hash 是 32 位，无符号右移 segmentShift(28) 位，剩下低 4 位，
// 然后和 segmentMask(15) 做一次与操作，也就是说 j 是 hash 值的最后 4 位，也就是槽的数组下标
int j = (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 ensureSegment
s = ensureSegment(j);
// 3. 插入新值到 槽 s 中
return s.put(key, hash, value, false);
}
- 通过 key 定位到 Segment，在对应的 Segment 中进行具体的 put 操作。
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
// 在往该 segment 写入前，需要先获取该 segment 的独占锁
HashEntry<K,V> node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
V oldValue;
try {
// segment 内部的数组
HashEntry<K,V>[] tab = table;
// 利用 hash 值，求应该放置的数组下标
int index = (tab.length - 1) & hash;
// first 是数组该位置处的链表的表头
HashEntry<K,V> first = entryAt(tab, index);
for (HashEntry<K,V> e = first;;) {
if (e != null) {
K k;
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);
else
node = new HashEntry<K,V>(hash, key, value, first);
int c = 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();
}
return oldValue;
}
初始化槽
- ConcurrentHashMap 初始化的时候会初始化第一个槽 segment[0]，对于其他槽，在插入第一个值的时候再进行初始化。
- 该方法考虑了并发情况。
- 多个线程同时进入初始化同一个槽 segment[k]，但只要有一个成功就可以了。
private Segment<K,V> ensureSegment(int k) {
final Segment<K,V>[] ss = this.segments;
long u = (k << SSHIFT) + SBASE; // raw offset
Segment<K,V> seg;
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
// 使用当前 segment[0] 处的数组长度和负载因子来初始化 segment[k]，这就是之前要初始化 segment[0] 的原因。
// 为什么要用 " 当前 "，因为 segment[0] 可能早就扩容过了。
Segment<K,V> proto = ss[0];
int cap = proto.table.length;
float lf = proto.loadFactor;
int threshold = (int)(cap * lf);
// 初始化 segment[k] 内部的数组
HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) { // 再次检查一遍该槽是否被其他线程初始化。
Segment<K,V> s = new Segment<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;
}
}
}
return seg;
}
- 未能获取到独占锁，则利用
scanAndLockForPut()自旋获取锁。
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
HashEntry<K,V> first = entryForHash(this, hash);
HashEntry<K,V> e = first;
HashEntry<K,V> node = null;
int retries = -1; // negative while locating node
// 循环获取锁
while (!tryLock()) {
HashEntry<K,V> f; // to recheck first below
if (retries < 0) {
if (e == null) {
if (node == null) // speculatively create node
// 进到这里说明数组该位置的链表是空的，没有任何元素
// 当然，进到这里的另一个原因是 tryLock() 失败，所以该槽存在并发，不一定是该位置
node = new HashEntry<K,V>(hash, key, value, null);
retries = 0;
}
else if (key.equals(e.key))
retries = 0;
else
// 顺着链表往下走
e = e.next;
}
// 重试次数如果超过 MAX_SCAN_RETRIES（单核 1 次多核 64 次），那么不抢了，进入到阻塞队列等待锁
// lock() 是阻塞方法，直到获取锁后返回
else if (++retries > MAX_SCAN_RETRIES) {
lock();
break;
}
else if ((retries & 1) == 0 &&
// 进入这里，说明有新的元素进到了链表，并且成为了新的表头
// 这边的策略是，重新执行 scanAndLockForPut 方法
(f = entryForHash(this, hash)) != first) {
e = first = f; // re-traverse if entry changed
retries = -1;
}
}
return node;
}
- 尝试自旋获取锁。
- 重试的次数达到了 MAX_SCAN_RETRIES 则改为阻塞锁获取，保证能获取成功。
- MAX_SCAN_RETRIES 单核 CPU，重试 1 次，多核 CPU 重试 64 次。
数组扩容
- segment 数组不能扩容，是对 segment 数组某个位置内部的数组 HashEntry[] 进行扩容，扩容后容量为原来的 2 倍，该方法没有考虑并发，因为执行该方法之前已经获取了锁。
// 方法参数上的 node 是这次扩容后，需要添加到新的数组中的数据。
private void rehash(HashEntry<K,V> node) {
HashEntry<K,V>[] oldTable = table;
int oldCapacity = oldTable.length;
// 2 倍
int newCapacity = oldCapacity << 1;
threshold = (int)(newCapacity * loadFactor);
// 创建新数组
HashEntry<K,V>[] newTable =
(HashEntry<K,V>[]) new HashEntry[newCapacity];
// 新的掩码，如从 16 扩容到 32，那么 sizeMask 为 31，对应二进制 ‘000...00011111’
int sizeMask = newCapacity - 1;
// 遍历原数组，将原数组位置 i 处的链表拆分到 新数组位置 i 和 i+oldCap 两个位置
for (int i = 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 = 19
int idx = e.hash & sizeMask;
if (next == null) // 该位置处只有一个元素
newTable[idx] = e;
else { // Reuse consecutive sequence at same slot
// e 是链表表头
HashEntry<K,V> lastRun = e;
// idx 是当前链表的头结点 e 的新位置
int lastIdx = idx;
// for 循环找到一个 lastRun 结点，这个结点之后的所有元素是将要放到一起的
for (HashEntry<K,V> last = next;
last != null;
last = last.next) {
int k = 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) {
V v = p.value;
int h = p.hash;
int k = h & sizeMask;
HashEntry<K,V> n = newTable[k];
newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
}
}
}
}
// 将新来的 node 放到新数组中刚刚的 两个链表之一 的 头部
int nodeIndex = node.hash & sizeMask; // add the new node
node.setNext(newTable[nodeIndex]);
newTable[nodeIndex] = node;
table = newTable;
}
- 代码中包含两个紧挨着的 for 循环，没有第一个 for 循环，也是可以工作的，如果 lastRun 的后面还有比较多的结点，那么循环是值得的，但是如果每次 lastRun 都是链表的最后一个元素或者很靠后的元素，那么这次遍历就浪费了，不过 Doug Lea 经过测试结果统计，如果使用默认的阈值，大约只有 1/6 的结点需要克隆。
put 方法的流程。
- 将当前 Segment 中的 table 通过 key 的 hashcode 定位到 HashEntry。
- 遍历该 HashEntry，如果不为空则判断传入的 key 和当前遍历的 key 是否相等，相等则覆盖旧的 value。
- 不为空则需要新建一个 HashEntry 并加入到 Segment 中，同时会先判断是否需要扩容。
- 最后再解除在第 1 步中所获取当前 Segment 的锁。
get 方法过程
public V get(Object key) {
Segment<K,V> s; // manually integrate access methods to reduce overhead
HashEntry<K,V>[] tab;
// 1. hash 值
int h = hash(key);
long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
// 2. 根据 hash 找到对应的 segment
if ((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) {
K k;
if ((k = e.key) == key || (e.hash == h && key.equals(k)))
return e.value;
}
}
return null;
}
- get 方法的流程。
- 计算 hash 值，找到 segment 数组中的具体位置，获使用的槽。
- 槽中也是一个数组，根据 hash 找到数组中具体的位置。
- 顺着链表进行查找即可。
- 因为 get 过程中没有加锁，因此需要考虑并发问题。
put 操作的线程安全性
- 初始化槽，使用了 CAS 来初始化 Segment 中的数组。
- 添加结点到链表的操作是插入到表头的，如果这个时候 get 操作在链表遍历的过程已经到了中间，是不会受影响的。
- get 操作如果在 put 之后，那么依赖 setEntryAt 方法中使用的 UNSAFE.putOrderedObject，可以保证刚刚插入表头的结点被读取。
- 扩容是新创建了数组，然后进行迁移数据，最后再将 newTable 设置给属性 table。
- 如果 get 操作此时也在进行并且先行，那么就是在旧的 table 上做查询操作。
- 如果 put 先行，那么 table 使用了 volatile 关键字，保证了 put 操作的可见性。
remove 操作的线程安全性
- 通过 key 定位到 Segment，在对应的 Segment 中进行具体的 remove 操作。
public V remove(Object key) {
int hash = this.hash(key);
ConcurrentHashMap.Segment s = this.segmentForHash(hash);
return s == null?null:s.remove(key, hash, (Object)null);
}
- 如果 get 先执行再 remove 操作，那么不存在任何问题。
- 如果 remove 先执行再 get 操作。
- 如果此结点是头结点，那么需要将头结点的 next 设置为数组该位置的元素，使用了 UNSAFE 操作数组保证可见性。
- 如果要删除的结点不是头结点，那么需要将要删除结点的后继结点接到前驱结点中，这里的并发保证是 next 使用了 volatile 关键字。
final V remove(Object key, int hash, Object value) {
if(!this.tryLock()) {
this.scanAndLock(key, hash);
}
Object oldValue = null;
try {
ConcurrentHashMap.HashEntry[] tab = this.table;
int index = tab.length - 1 & hash;
ConcurrentHashMap.HashEntry e = ConcurrentHashMap.entryAt(tab, index);
ConcurrentHashMap.HashEntry next;
for(ConcurrentHashMap.HashEntry pred = null; e != null; e = next) {
next = e.next;
Object k = e.key;
if(e.key == key || e.hash == hash && key.equals(k)) {
Object v = e.value;
if(value != null && value != v && !value.equals(v)) {
break;
}
if(pred == null) {
ConcurrentHashMap.setEntryAt(tab, index, next);
} else {
pred.setNext(next);
}
++this.modCount;
--this.count;
oldValue = v;
break;
}
pred = e;
}
} finally {
this.unlock();
}
return oldValue;
}
size
-
要统计整个 ConcurrentHashMap 里元素的大小，就必须统计所有 Segment 里元素的大小后求和。
- Segment 里的全局变量 count 是一个 volatile 变量。
-
ConcurrentHashMap 的做法是先尝试 2 次通过不锁住 Segment 的方式统计各个 Segment 大小，如果统计的过程中，容器的 count 发生了变化，则再采用加锁的方式来统计所有 Segment 的大小。
- 使用 modCount 变量，在 put、remove 和 clean 方法里操作元素前都会将变量 modCount 进行加 1，在统计 size 前后比较 modCount 是否发生变化，从而得知容器的大小是否发生变化。
