Java Map 之 ConcurrentHashMap浅析
2018-09-11 本文已影响0人
KIDNG_LGJ
总所周知HashMap是非线程安全的类。若使用需要线程安全可对应使用Hashtable或者ConcurrentHashMap。因为Hashtable为遗留类（从命名上也略知一二），所以需要线程安全时推荐使用ConcurrentHashMap。Hashtable实现与HashMap基本一致，在公开方法中通过synchronized描述符保证线程安全，且不接受null的key值和value值（HashMap对null做了特殊处理）。
关于HashMap的解析可以参考我的文章Java Map 浅析之 HashMap
HashMap在多线程并发时可能死循环,可以参考文章疫苗：JAVA HASHMAP的死循环
ConcurrentHashMap
ConcurrentHashMap性能上比Hashtable高，是因为ConcurrentHashMap采用的是分段锁。
在ConcurrentHashMap结构上，采用分段(segment)存储桶结构，通过锁Segment保证每段并发时的线程安全。默认支持的最大并发量为16（由concurrencyLevel确定，初始化后不可更改）
基于jdk1.7.0_51
在java7中hash获取生成也是经过多次扰动的。
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; //真正决定segments数组的大小
// 计算并行级别 ssize，因为要保持并行级别是 2 的 n 次方
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
//若用默认值，concurrencyLevel 为 16(ssize也为16)，sshift 为 4
// 那么计算出 segmentShift 为 28，segmentMask 为 15，都是用于定位
this.segmentShift = 32 - sshift;//segment定位右移量，后面说明
this.segmentMask = ssize - 1;//segment定位掩码位数，后面说明
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
int c = initialCapacity / ssize;//计算每个segment容量
if (c * ssize < initialCapacity)
++c;
int cap = MIN_SEGMENT_TABLE_CAPACITY;//默认为2，保证第一次插入不会扩容
while (cap < c)
cap <<= 1;
// create segments and segments[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];
UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
this.segments = ss;
}
可以看到使用无参构造函数，得到的为
- Segment 数组长度为 16，不可以扩容
- Segment[i] 的默认大小为 2，负载因子是 0.75，得出初始阈值为 1.5。保证第二次插入才会扩容
- 当前 segmentShift 的值为 32 – 4 = 28（hash值的位数[32]减去对segments的大小取以2为底的对数值），segmentMask 为 16 – 1 = 15(Segments数组的大小减 1)，姑且把它们简单翻译为移位数和掩码
- 初始化了 segment[0]，其他位置还是 null。后面会说明的。
/**
* Segments are specialized versions of hash tables. This
* subclasses from ReentrantLock opportunistically, just to
* simplify some locking and avoid separate construction.
*/
static final class Segment<K,V> extends ReentrantLock implements Serializable {
/**
* The per-segment table. Elements are accessed via
* entryAt/setEntryAt providing volatile semantics.
*/
transient volatile HashEntry<K,V>[] table;
/**
* The number of elements. Accessed only either within locks
* or among other volatile reads that maintain visibility.
*/
transient int count;// Segment中元素的数量，可见的
/**
* The total number of mutative operations in this segment.
* Even though this may overflows 32 bits, it provides
* sufficient accuracy for stability checks in CHM isEmpty()
* and size() methods. Accessed only either within locks or
* among other volatile reads that maintain visibility.
*/
transient int modCount;//对count的大小造成影响的操作的次数
/**
* The table is rehashed when its size exceeds this threshold.
* (The value of this field is always <tt>(int)(capacity *
* loadFactor)</tt>.)
*/
transient int threshold; // 阈值，段中元素的数量超过这个值就会对Segment进行扩容
/**
* The load factor for the hash table. Even though this value
* is same for all segments, it is replicated to avoid needing
* links to outer object.
* @serial
*/
final float loadFactor;// 段的负载因子，其值等同于ConcurrentHashMap的负载因子
}
static final class HashEntry<K,V> {
final int hash;
final K key;
volatile V value;
volatile HashEntry<K,V> next;
}
Segment 类继承于 ReentrantLock 类，从而使得 Segment 对象能充当锁的角色。每个 Segment 对象用来守护它的成员对象 table 中包含的若干个桶。
通过锁分段技术，它使用了不同的锁来控制对哈希表的不同部分进行的修改(写)，而 ConcurrentHashMap 内部每个段(Segment)实质上就是一个小的哈希表，每个段都有自己的锁(Segment 类继承了 ReentrantLock 类)。这样，只要多个修改(写)操作发生在不同的段上，它们就可以并发进行。
public V put(K key, V value) {
Segment<K,V> s;
if (value == null)
throw new NullPointerException();
int hash = hash(key);//计算hash
//定位到对应segment（以默认值为例即高4位）
int j = (hash >>> segmentShift) & segmentMask;
if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck
(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
s = ensureSegment(j); //确保对应segment已初始化
return s.put(key, hash, value, false);//Segment.put()
}
可以看到ConcurrentHashMap不同于HashMap，它既不允许key值为null，也不允许value值为null。
假设ConcurrentHashMap一共分为2^n 个段，每个段中有2^m (Segment.table.length)个桶，那么段的定位方式是将key的hash值的高n位与(2^n -1)相与。在定位到某个段后，再将key的hash值的低m位与(2^m -1)相与，定位到具体的桶位。
以默认值为例，Segment数组长度为16(2^4)，这时segmentShift为28(32-4)，segmentMask为15（16-1）。即定位Segment为高4位决定。
而所在桶位置由Segment中table长度决定，该index与HashMap计算一致，即(Segment.table.length - 1) & hash。
//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 {
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash; //计算出对应桶位置
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 {//e为空，即链表为null或链表无对应元素
// 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 的阈值
if (c > threshold && tab.length < MAXIMUM_CAPACITY)
rehash(node);//扩容，后面分析
else
setEntryAt(tab, index, node);//将新的节点设置成原链表的表头（UNSAFE.putOrderedObject保证并发安全）
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
unlock();//解锁
}
return oldValue;
}
整体put流程不算复杂。因为是通过对Segment加锁操作，在理想状态下，ConcurrentHashMap 可以支持 16 个线程执行并发写操作（如果并发级别设置为 16）。
在添加值时要确保Segment已经初始化，通过ensureSegment初始化对用Segment。
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<K,V> proto = ss[0]; // use segment 0 as prototype
int cap = proto.table.length;
float lf = proto.loadFactor;
int threshold = (int)(cap * lf);
HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) { // recheck
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;
}
可以看出通过Segment[0]去初始化对应Segment[k]，并通过CAS确保线程安全。
关于CAS算法，可以参考并发策略-CAS算法和java Unsafe类中compareAndSwap相关介绍
在Segment.put中看到一开始HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value);的代码片段。通过 tryLock()快速获取一次锁，如果失败则通过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;
}
//超过重试次数则阻塞
else if (++retries > MAX_SCAN_RETRIES) {//重试次数，单核为1，多核为64
lock(); //阻塞方法，直到获取锁后返回
break;
}
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first) {
e = first = f; // re-traverse if entry changed
retries = -1;
}
}
return node;
}
scanAndLockForPut方法只有两个出口，一个为tryLock成功，第二种重试超过MAX_SCAN_RETRIES次数，阻塞lock()直到获得锁。如果需要顺便创建Segment中该桶链表。
最后put方法里面在插入值之前会判断插入后是否会导致扩容（插入后大小是否超过阈值），会则进行扩容(rehash)。
private void rehash(HashEntry<K,V> node) {
HashEntry<K,V>[] oldTable = table;
int oldCapacity = oldTable.length;
int newCapacity = oldCapacity << 1;//扩容为两倍
threshold = (int)(newCapacity * loadFactor);
HashEntry<K,V>[] newTable =
(HashEntry<K,V>[]) new HashEntry[newCapacity];
int sizeMask = newCapacity - 1;
for (int i = 0; i < oldCapacity ; i++) {
HashEntry<K,V> e = oldTable[i];
if (e != null) {
HashEntry<K,V> next = e.next;
//该idx只可能为原位置i或者为原位置i+oldCapacity
int idx = e.hash & sizeMask;//计算元素在新table中的位置
if (next == null) // Single node on list
newTable[idx] = e; //仅一个元素直接插入对应newTable[idx]
else { // Reuse consecutive sequence at same slot
HashEntry<K,V> lastRun = e;
int lastIdx = idx;
//循环找到一个lastRun节点，该节点后面元素都放置于newTable同一位置。
//（其实就是为了拿到链表最后的相同index元素的链表头）
for (HashEntry<K,V> last = next;
last != null;
last = last.next) {
int k = last.hash & sizeMask;
if (k != lastIdx) {
lastIdx = k;
lastRun = last;
}
}
newTable[lastIdx] = lastRun;
// Clone remaining nodes
//再将lastRun前面的节点根据新的Index插入newTable并作为链表头
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插入newTable中
int nodeIndex = node.hash & sizeMask; // add the new node
node.setNext(newTable[nodeIndex]);//重设node.next为newTable的链头
newTable[nodeIndex] = node;//插入newTable
table = newTable;
}
因为rehash过程是在已经或者锁的情况下执行的，所以不用考虑并发问题。
仔细看其实两个紧挨的for中第一个for不要也是没有关系的，如果 lastRun 的后面还有比较多的节点，那么就是值得的。Doug Lea 说了，根据统计，如果使用默认的阈值，大约只有 1/6 的节点需要克隆。
剩下remove方法
public V remove(Object key) {
int hash = hash(key);
Segment<K,V> s = segmentForHash(hash);
return s == null ? null : s.remove(key, hash, null);
}
final V remove(Object key, int hash, Object value) {
//获取锁
if (!tryLock())
scanAndLock(key, hash);
V oldValue = null;
try {
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash;
HashEntry<K,V> e = entryAt(tab, index);
HashEntry<K,V> pred = null; //要删除节点的先驱节点
while (e != null) {
K k;
HashEntry<K,V> next = e.next;
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
V v = e.value;
if (value == null || value == v || value.equals(v)) {
if (pred == null)
//删除节点为链头，将删除节点next节点设为链头
setEntryAt(tab, index, next);
else
pred.setNext(next);//重定向先驱节点的next节点
++modCount;
--count;
oldValue = v;
}
break;
}
pred = e;
e = next;
}
} finally {
unlock(); // 释放锁
}
return oldValue;
}
//和scanAndLockForPut差不多，只是没有创建node步骤
private void scanAndLock(Object key, int hash) {
// similar to but simpler than scanAndLockForPut
HashEntry<K,V> first = entryForHash(this, hash);
HashEntry<K,V> e = first;
int retries = -1;
while (!tryLock()) {
HashEntry<K,V> f;
if (retries < 0) {
if (e == null || key.equals(e.key))
retries = 0;
else
e = e.next;
}
else if (++retries > MAX_SCAN_RETRIES) {
lock();
break;
}
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first) {
e = first = f;
retries = -1;
}
}
}
分析过put方法，remove方法也就相对易懂很多。
分析完put、remove，get方法就相对简单了。
public V get(Object key) {
Segment<K,V> s; // manually integrate access methods to reduce overhead
HashEntry<K,V>[] tab;
int h = hash(key);
//找到对应segment位置
long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
(tab = s.table) != null) {
//找到对应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;
//找到对应key的元素返回value
if ((k = e.key) == key || (e.hash == h && key.equals(k)))
return e.value;
}
}
return null;
}
可以看到get方法没有加锁，所以需要考虑线程安全问题。
put方法：
- 初始化->通过CAS算法保证初始化成功
- 添加节点时分为插入表头和扩容。
- 插入表头：get已经遍历至链表中间则无影响。get在put之后则依赖于 setEntryAt 方法中使用的 UNSAFE.putOrderedObject
- 扩容：如果 get 先行，那么就是在旧的 table 上做查询操作；而 put 先行，那么 put 操作的可见性保证就是 table 使用了 volatile 关键字。
remove方法：
- 删除节点已完成，则无影响。
- 删除节点为头节点则依赖于 setEntryAt 方法。
- 删除节点为中间节点则依赖于HashEntry.next的volatile修饰符。
java8 ConcurrentHashMap
在java8中ConcurrentHashMap也引入了红黑树。结构上和 Java8 的 HashMap 基本上一样，因为要保证线程安全源码上会复杂很多。
/**
* Creates a new, empty map with the default initial table size (16).
*/
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;
}
可以看到默认容器大小为16。传入initialCapacity最终生成1.5*initialCapacity+1的向上取整2次幂大小的容器。即10->16，11->32。
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) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode()); //和java8一样进行一次扰动，高位参与运算
int binCount = 0; //记录桶长度
//循环尝试获取最新的table
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
if (tab == null || (n = tab.length) == 0)
tab = initTable(); //初始化table
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//位置为null ，通过CAS插入新值。插入失败，再次循环
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); //帮助扩容，扩容完成返回扩容后table
else {
V oldVal = null;
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) {
binCount = 1;
for (Node<K,V> e = f;; ++binCount) {
K ek;
//找到对应key，直接覆盖
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;
}
}
else if (f instanceof ReservationNode)
throw new IllegalStateException("Recursive update");
}
}
if (binCount != 0) { // binCount != 0 说明上面在做链表操作
// 判断是否要将链表转换为红黑树，临界值和 HashMap 一样，也是 8
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i); //但是数组小于64则是扩容，大于64则转为红黑树
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}
put主流程看完，大概有初始化、扩容、帮助数据迁移这三个点是要说明的。
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
if ((sc = sizeCtl) < 0) //别的线程已经在初始化
Thread.yield(); // lost initialization race; just spin
// 通过CAS将sizeCtl置为-1，即代表抢到了锁
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;//DEFAULT_CAPACITY==16
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;//table修饰符为volatile
sc = n - (n >>> 2); //sc = 0.75 * n
}
} finally {
sizeCtl = sc;//sizeCtl = 0.75 * n。默认16则为12
}
break;
}
}
return tab;
}
可以看到初始化时通过CAS保证初始化创建的。
这里也看到对sizeCtl得操作，我门要知道一点：
- sizeCtl = 0：默认值
- sizeCtl = -1：代表哈希表正在进行初始化
- sizeCtl > 0：相当于 HashMap 中的 threshold，表示阈值
- sizeCtl < -1：代表有多个线程正在进行扩容
treeifyBin()并不一定转化为红黑树，仅在数组大小大于64时转化，小于64进行数组扩容
private final void treeifyBin(Node<K,V>[] tab, int index) {
Node<K,V> b; int n;
if (tab != null) {
//table长度小于64，翻倍扩容
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
tryPresize(n << 1);
//获取table对应位置的头节点
else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
synchronized (b) { //加锁
if (tabAt(tab, index) == b) {
TreeNode<K,V> hd = null, tl = null;
for (Node<K,V> e = b; e != null; e = e.next) {//遍历建立红黑树
TreeNode<K,V> p =
new TreeNode<K,V>(e.hash, e.key, e.val,
null, null);
if ((p.prev = tl) == null)
hd = p;
else
tl.next = p;
tl = p;
}
setTabAt(tab, index, new TreeBin<K,V>(hd));//将树放入table中
}
}
}
}
}
//传入的size已经翻倍
private final void tryPresize(int size) {
// c：1.5 * size +1 且往上取整的2次幂
int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
tableSizeFor(size + (size >>> 1) + 1);
int sc;
while ((sc = sizeCtl) >= 0) {
Node<K,V>[] tab = table; int n;
if (tab == null || (n = tab.length) == 0) {//和初始化基本一致
n = (sc > c) ? sc : c;
if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if (table == tab) {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = nt;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
}
}
else if (c <= sc || n >= MAXIMUM_CAPACITY)
break;
else if (tab == table) {
int rs = resizeStamp(n);//返回一个 16 位长度的扩容校验标识
if (sc < 0) {
Node<K,V>[] nt;
//前 16 位是数据校验标识，后 16 位是当前正在扩容的线程总数
//判断校验标识是否相等，如果校验符不等或者扩容操作已经完成了，直接退出循环，不用协助它们扩容了
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt); //迁移扩容
}
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);//迁移扩容
}
}
}
在putVal方法中当桶的头节点的hash值为MOVE时进行helpTransfer。helpTransfer方法中会判断对应扩容标志位，若在扩容期间则进行协助扩容。
static final class ForwardingNode<K,V> extends Node<K,V> {
final Node<K,V>[] nextTable;
ForwardingNode(Node<K,V>[] tab) {
super(MOVED, null, null, null);
this.nextTable = tab;
}
}
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
Node<K,V>[] nextTab; int sc;
if (tab != null && (f instanceof ForwardingNode) &&
(nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
int rs = resizeStamp(tab.length);//返回一个 16 位长度的扩容校验标识
while (nextTab == nextTable && table == tab &&
(sc = sizeCtl) < 0) {
//判断校验标识是否相等，如果校验符不等或者扩容操作已经完成了，直接退出循环，不用协助扩容
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || transferIndex <= 0)
break;
//sizeCtl 标识增加一，标识增加一个线程进行扩容
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}
当hash==MOVED，则当前节点为ForwardingNode类型。
这个节点内部保存了nextTable 引用，它指向一张 hash 表。在扩容操作中，我们需要对每个桶中的结点进行分离和转移，如果某个桶结点中所有节点都已经迁移完成了（已经被转移到新表 nextTable 中了），那么会在原 table 表的该位置挂上一个 ForwardingNode 结点，说明此桶已经完成迁移。
所以，我们在 putVal()中遍历整个 hash 表的桶结点，如果遇到 hash==MOVED，说明已经有线程正在扩容 rehash 操作，不过我们要插入的桶的位置已经完成了所有节点的迁移。且由于检测到当前哈希表正在扩容，于是让当前线程去协助扩容。
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
//计算单个线程处理table的桶数，最小为16
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
//如果 nextTab 为 null，即刚开始扩容，先进行一次初始化
//外围方法已保证第一个发起迁移的线程调用时 nextTab 为 null
if (nextTab == null) { // initiating
try {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];//容量翻倍
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
// transferIndex 是 ConcurrentHashMap 的属性，用于控制迁移的位置，
// 指向最后一个桶，方便从后向前遍历
transferIndex = n;
}
int nextn = nextTab.length;
//用于标记对应桶位置已迁移完成
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
//下面部分将时并发扩容的核心---
//advance 为 true 表示可以进行下一个位置的迁移
boolean advance = true;
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;
if (--i >= bound || finishing)
advance = false;
// transferIndex <= 0，说明所有桶都有相应的线程去处理或已经处理完成
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {//nextBound为当前线程处理边界
//标识出边界及位置
bound = nextBound;
i = nextIndex - 1;
advance = false;
}
}
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {//所有迁移已完成
nextTable = null;
table = nextTab; //nextTab赋值给table，完成迁移
sizeCtl = (n << 1) - (n >>> 1);//重新计算阈值，为新数组长度0.75
return;
}
//没有一个线程协助扩容则sizeCtl+1，这里当前线程已经完成扩容所以通过CAS操作sizeCtl-1
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)//任务结束，方法退出
return;
//(sc - 2) == resizeStamp(n) << RESIZE_STAMP_SHIFT，则为所有任务结束，进入上面 if (finishing) {}
finishing = advance = true;
i = n; // recheck before commit
}
}
else if ((f = tabAt(tab, i)) == null) //待迁移桶为空，放入ForwardingNode节点标记已处理
advance = casTabAt(tab, i, null, fwd);
else if ((fh = f.hash) == MOVED)//节点为ForwardingNode，跳过
advance = true; // already processed
else {
synchronized (f) {//对对应桶进行加锁，开始处理对应桶
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
if (fh >= 0) {//头结点的 hash 大于 0，说明是链表的 Node 节点
int runBit = fh & n;
Node<K,V> lastRun = f;
//与java7差不多，lastRun为最后相同节点的链头
for (Node<K,V> p = f.next; p != null; p = p.next) {
int b = p.hash & n;
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
//因为hash扰动规则修改了，高位参与低位计算
//runBit == 0表示节点所在桶位置不用更新
if (runBit == 0) {
ln = lastRun;
hn = null;
}
else {
hn = lastRun;
ln = null;
}
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);
else
hn = new Node<K,V>(ph, pk, pv, hn);
}
setTabAt(nextTab, i, ln);//ln低位链表放回原桶位置
setTabAt(nextTab, i + n, hn);//hn高位链表放入i+n的桶位置
setTabAt(tab, i, fwd);//标识旧table的i位置的桶已经处理
advance = true;//可以继续先前遍历
}
else if (f instanceof TreeBin) {//红黑树复制迁移
TreeBin<K,V> t = (TreeBin<K,V>)f;
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;
else
loTail.next = p;
loTail = p;
++lc;
}
else {
if ((p.prev = hiTail) == null)
hi = p;
else
hiTail.next = p;
hiTail = p;
++hc;
}
}
// 如果一分为二后，节点数少于 8，将红黑树转换回链表
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);//标识旧table的i位置的桶已经处理
advance = true;
}
}
}
}
}
}
transfer中有几个需要注意的
- 形参nextTab在传入时，第一发起迁移的传入为null，后续都是为新表即ForwardingNode.nextTable
- 循环中i为当前处理桶的index，bound为处理的上边界，遍历时从后往前遍历
- for (int i = 0, bound = 0;;) {}先去理解内部循环再理解整个大循环会清晰很多。
可以看到java8对应java7有相当多的改动
- hash扰动减少，但是高位也参与低位的运算。
- 去除了Segment，但是通过锁链头方式实现分段锁。
- 多线程竞争修改表时没有拒绝其他线程的修改（java7在进行写操作时segment上锁，其他线程只能等待），而是让其他线程协助扩容工作。
