Java 7 中 ConcurrentHashMap 的存储结构如下图，ConcurrnetHashMap 由很多个 Segment 组合，而每一个 Segment 内部是一个类似于 HashMap 结构的 HashEntry 数组，所以可以进行扩容。但是 Segment 的个数一旦初始化就不能改变，默认 Segment 的个数是 16 个，可以认为 ConcurrentHashMap 默认支持最多 16 个线程并发。
一个ConcurrentHashMap中只有一个Segment
/**
* 默认初始化容量
*/
static final int DEFAULT_INITIAL_CAPACITY = 16;
/**
* 默认负载因子
*/
static final float DEFAULT_LOAD_FACTOR = 0.75f;
/**
* 默认并发级别
*/
static final int DEFAULT_CONCURRENCY_LEVEL = 16;
- capacity：与小数组（HashEntry）的容量有关（小数组的初始容量 = capacity / clevel，最小是2）
- factor：负载因子，影响的是小数组的扩容
- clevel：并发度，也即Segment 的个数，大数组的个数，默认是16
@SuppressWarnings("unchecked")
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;
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
this.segmentShift = 32 - sshift;
this.segmentMask = ssize - 1;
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
// c = 容量 / ssize ，默认 16 / 16 = 1，这里是计算HashEntry的容量
int c = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
int cap = MIN_SEGMENT_TABLE_CAPACITY;
//HashEntry的容量至少是2或者2的倍数
while (cap < c)
cap <<= 1;
// create segments and segments[0]
// 创建 Segment 数组，设置 segments[0]
Segment s0 = new Segment(loadFactor, (int)(cap * loadFactor),
(HashEntry[])new HashEntry[cap]);
Segment[] ss = (Segment[])new Segment[ssize];
UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
this.segments = ss;
}
构造函数中的重点：
- 记录偏移量sshift=clevel 2的N次方中N，得到segmentShift = 32 - sshift;用来确定目标值在Segment中的位置
- 记录segmentMask = ssize - 1;使得segmentMask的每一位都为1
- Segment的容量ssize = 大于等于clevel的最小2次幂
- HashEntry的容量cap = 大于等于 ( 初始容量 / ssize ) 的最小2次幂
public V put(K key, V value) {
Segment s;
if (value == null)
throw new NullPointerException();
int hash = hash(key);
int j = (hash >>> segmentShift) & segmentMask;
if ((s = (Segment)UNSAFE.getObject // nonvolatile; recheck
(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
// 如果查找到的 Segment 为空，初始化
s = ensureSegment(j);
return s.put(key, hash, value, false);
}
@SuppressWarnings("unchecked")
private Segment ensureSegment(int k) {
final Segment[] ss = this.segments;
long u = (k << SSHIFT) + SBASE; // raw offset
Segment seg;
// 判断 u 位置的 Segment 是否为null
if ((seg = (Segment)UNSAFE.getObjectVolatile(ss, u)) == null) {
Segment proto = ss[0]; // use segment 0 as prototype
// 获取0号 segment 里的 HashEntry 初始化长度
int cap = proto.table.length;
// 获取0号 segment 里的 hash 表里的扩容负载因子，所有的 segment 的 loadFactor 是相同的
float lf = proto.loadFactor;
// 计算扩容阀值
int threshold = (int)(cap * lf);
// 创建一个 cap 容量的 HashEntry 数组
HashEntry[] tab = (HashEntry[])new HashEntry[cap];
if ((seg = (Segment)UNSAFE.getObjectVolatile(ss, u)) == null) { // recheck
// 再次检查 u 位置的 Segment 是否为null，因为这时可能有其他线程进行了 *** 作
Segment s = new Segment(lf, threshold, tab);
// 自旋检查 u 位置的 Segment 是否为null
while ((seg = (Segment)UNSAFE.getObjectVolatile(ss, u))
== null) {
// 使用CAS 赋值，只会成功一次
if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
break;
}
}
}
return seg;
} 流程如下：
- 根据segmentShift和segmentMask确定要插入的元素在segment中的位置
- 如果查找到的segment位置上为空，则进入初始化
- 根据segment[0]里HashEntry的容量和负载因子，计算阈值，创建新的HashEntry数组
- 判断segment中该位置上是否仍为null，如果仍然是则使用CAS赋值
将得到的hash值向右按位移动segmentShift位，然后再与segmentMask做&运算得到Segment的索引。
在初始化的时候，segmentShift的值等于32-sshift，例如concurrencyLevel等于16，则sshift等于4，那么segmentShift为28。hash值是一个32位的整数，将其向右移动28就变成这个样子：0000 0000 0000 0000 0000 0000 0000 XXXX，然后再用这个值和segmentMask做与运算，所得结果确定Segment的索引。
以上仅仅是初始化segment的 *** 作，在初始化完成后将进入插入 *** 作：
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
//先尝试对segment加锁，如果直接加锁成功，那么node=null；如果加锁失败，则会调用scanAndLockForPut方法去获取锁，
//在这个方法中，获取锁后会返回对应HashEntry（要么原来就有要么新建一个）
HashEntry node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
V oldValue;
try {
//这里是一个优化点，由于table自身是被volatile修饰的，然而put这一块代码本身是加锁了的，所以同一时间内只会有一个线程 *** 作这部分内容，
//所以不再需要对这一块内的变量做任何volatile修饰，因为变量加了volatile修饰后，变量无法进行编译优化等，会对性能有一定的影响
//故将table赋值给put方法中的一个局部变量，从而使得能够减少volatile带来的不必要消耗。
HashEntry[] tab = table;
int index = (tab.length - 1) & hash;
//这里有一个问题：为什么不直接使用数组下标获取HashEntry，而要用entryAt来获取链表？
//这里结合网上内容个人理解是：由于Segment继承的是ReentrantLock，所以它是一个可重入锁，那么是否存在某种场景下，
//会导致同一个线程连续两次进入put方法，而由于put最终使用的putOrderedObject只是禁止了写写重排序无法保证内存可见性，
//所以这种情况下第二次put在获取链表时必须用entryAt中的volatile语义的get来获取链表，因为这种情况下下标获取的不一定是最新数据。
HashEntry first = entryAt(tab, index);//先获取需要put的对在当前这个segment中对应的链表的表头结点。
for (HashEntry e = first;;) {//开始遍历first为头结点的链表
if (e != null) {//<1>
//e不为空，说明当前键值对需要存储的位置有hash冲突，直接遍历当前链表，如果链表中找到一个节点对应的key相同，
//依据onlyIfAbsent来判断是否覆盖已有的value值
K k;
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) { //说明键的Hash值一样，内容也一样
//进入这个条件内说明需要put的对应的key节点已经存在，直接判断是否更新并最后break退出循环。
oldValue = e.value;
if (!onlyIfAbsent) {
e.value = value;
++modCount;
}
break;
}
e = e.next;//未进入上面的if条件中，说明当前e节点对应的key不是需要的，直接遍历下一个节点。
}
else {//<2> 头插法
//进入到这个else分支，说明e为空，对应有两种情况下e可能会为空，即：
// 1>. <1>中进行循环遍历，遍历到了链表的表尾仍然没有满足条件的节点。
// 2>. e=first一开始就是null（可以理解为即一开始就遍历到了尾节点）
if (node != null) //这里有可能获取到锁是通过scanAndLockForPut方法内自旋获取到的，这种情况下依据找好或者说是新建好了对应节点，node不为空
node.setNext(first);
else // 当然也有可能是这里直接第一次tryLock就获取到了锁，从而node没有分配对应节点，即需要给依据插入的k,v来创建一个新节点
node = new HashEntry(hash, key, value, first);
int c = count + 1; //总数+1 在这里依据获取到了锁，即是线程安全的！对应了上述对count变量的使用规范说明。
if (c > threshold && tab.length < MAXIMUM_CAPACITY)//判断是否需要进行扩容
//扩容是直接重新new一个新的HashEntry数组，这个数组的容量是老数组的两倍，
//新数组创建好后再依次将老的table中的HashEntry插入新数组中，所以这个过程是十分费时的，应尽量避免。
//扩容完毕后，还会将这个node插入到新的数组中。
rehash(node);
else
setEntryAt(tab, index, node);
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
unlock();
}
return oldValue;
}
插入过程如下：
- 计算得到插入元素的目标位置，获取这个位置上的HashEntry元素。
- 如果这个位置上的头结点first!=null，遍历链表，找key值相同或者key值和value值都相同的结点，根据onlyIfAbsent判断是否覆盖oldValue
- 如果这个位置上的头结点first==null，有两种可能：1⃣️遍历到链表尾部都没有一样的结点2⃣️fist==null。使用头插法将目标结点插入到链表中，如果插入后的容量大于阈值，则扩容。
private void rehash(HashEntry node) {
HashEntry[] oldTable = table;
// 老容量
int oldCapacity = oldTable.length;
// 新容量，扩大两倍
int newCapacity = oldCapacity << 1;
// 新的扩容阀值
threshold = (int)(newCapacity * loadFactor);
// 创建新的数组
HashEntry[] newTable = (HashEntry[]) new HashEntry[newCapacity];
// 新的掩码，默认2扩容后是4，-1是3，二进制就是11。
int sizeMask = newCapacity - 1;
for (int i = 0; i < oldCapacity ; i++) {
// 遍历老数组
HashEntry e = oldTable[i];
if (e != null) {
HashEntry next = e.next;
// 计算新的位置，新的位置只可能是不变或者是老的位置+老的容量。
int idx = e.hash & sizeMask;
if (next == null) // Single node on list
// 如果当前位置还不是链表，只是一个元素，直接赋值
newTable[idx] = e;
else { // Reuse consecutive sequence at same slot
// 如果是链表了
HashEntry lastRun = e;
int lastIdx = idx;
// 新的位置只可能是不便或者是老的位置+老的容量。
// 遍历结束后，lastRun 后面的元素位置都是相同的
for (HashEntry last = next; last != null; last = last.next) {
int k = last.hash & sizeMask;
if (k != lastIdx) {
lastIdx = k;
lastRun = last;
}
}
// lastRun 后面的元素位置都是相同的，直接作为链表赋值到新位置。
newTable[lastIdx] = lastRun;
// Clone remaining nodes
for (HashEntry p = e; p != lastRun; p = p.next) {
// 遍历剩余元素，头插法到指定 k 位置。
V v = p.value;
int h = p.hash;
int k = h & sizeMask;
HashEntry n = newTable[k];
newTable[k] = new HashEntry(h, p.key, v, n);
}
}
}
}
// 头插法插入新的节点
int nodeIndex = node.hash & sizeMask; // add the new node
node.setNext(newTable[nodeIndex]);
newTable[nodeIndex] = node;
table = newTable;
}
扩容的流程：
- 新的HashEntry[]容量是原来的二倍，负载因子不变，可以求出新的阈值，以及新的sizeMask，根据新的sizeMask和新的容量可以定位元素在新的数组中的位置
- 遍历原来的数组，计算数组中元素新的hash值，定位到新的数组中，判断该元素是否有下一个结点
- 如果没有下个结点，可以直接赋值到新的数组
- 如果有下个结点，进入到第一个for循环中，目的是找到链表后面几个hash值相同并相连的结点，如果找到可以直接批量赋值到新的数组
- 然后进入到第二个for循环中，把剩下的结点再放在新的数组中
- 最后把输入的结点插入到新的数组中
public V get(Object key) {
Segment s; // manually integrate access methods to reduce overhead
HashEntry[] tab;
//1 和put *** 作一样，先通过key进行两次hash确定取哪个segment中的数据
int h = hash(key);
//2 使用UNSAFE方法获取对应的Segment，然后再进行一次&运算得到HashEntry链表的位置，然后从链表头开始遍历整个链表。
//（由于hash会碰撞，所以用一个链表保存），如果找到对应的key，则返回对应的value值，如果链表遍历完都没有找到对应的key，
// 则说明map中不包含该key，返回null
long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
if ((s = (Segment)UNSAFE.getObjectVolatile(segments, u)) != null &&
(tab = s.table) != null) {
for (HashEntry e = (HashEntry) UNSAFE.getObjectVolatile
(tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
e != null; e = e.next) {
K k;
// 如果发生哈希碰撞，判断key是否相同
if ((k = e.key) == key || (e.hash == h && key.equals(k)))
return e.value;
}
}
return null;
} 查找的流程如下：
- 求得目标元素在segment中的位置，得到该位置上的HashEntry数组
- 求得目标元素在HashEntry中的位置，遍历链表找到key相同的元素，返回value值
//最大容量是2^30
private static final int MAXIMUM_CAPACITY = 1 << 30;
//初始容量是16
private static final int DEFAULT_CAPACITY = 16;
//默认并发级别是16
private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
//默认负载因子0.75
private static final float LOAD_FACTOR = 0.75f;
//如果一个hash桶中的节点数达到8，下次添加新节点时，会把这个hash桶的所有节点用红黑树保存
//如果数组table的长度不足64，那么也不转化为红黑树，改为扩容一次
static final int TREEIFY_THRESHOLD = 8;
//如果一棵红黑树的节点减少到6，那么就把它退化为链表保存
static final int UNTREEIFY_THRESHOLD = 6;
//转化为红黑树的另一个条件。table的长度不足这个值时，不转化为红黑树，改为扩容一次
static final int MIN_TREEIFY_CAPACITY = 64;除此之外，ConcurrentHashMap还有一个重要的参数：sizeCtl
/**
* Table initialization and resizing control. When negative, the
* table is being initialized or resized: -1 for initialization,
* else -(1 + the number of active resizing threads). Otherwise,
* when table is null, holds the initial table size to use upon
* creation, or 0 for default. After initialization, holds the
* next element count value upon which to resize the table.
*/
private transient volatile int sizeCtl;（二）构造函数
- 负数代表正在进行初始化或扩容 *** 作 ,其中-1代表正在初始化 ,-N 表示有N-1个线程正在进行扩容 *** 作
- 正数或0代表hash表还没有被初始化，这个数值表示初始化或下一次进行扩容时的大小，类似于扩容阈值。它的值始终是当前ConcurrentHashMap容量的0.75倍，这与loadfactor是对应的。实际容量>=sizeCtl，则扩容。
//构造方法1
public ConcurrentHashMap() {
}
//构造方法2
public ConcurrentHashMap(int initialCapacity) {
if (initialCapacity < 0)
throw new IllegalArgumentException();
int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
MAXIMUM_CAPACITY :
tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
// sizeCtl表示初始化的数组容量
this.sizeCtl = cap;
}
//构造方法3
public ConcurrentHashMap(Map m) {
this.sizeCtl = DEFAULT_CAPACITY;
putAll(m);
}
//构造方法4
public ConcurrentHashMap(int initialCapacity, float loadFactor) {
this(initialCapacity, loadFactor, 1);
}
//构造方法5
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 bins
initialCapacity = concurrencyLevel; // as estimated threads
long size = (long)(1.0 + (long)initialCapacity / loadFactor);
int cap = (size >= (long)MAXIMUM_CAPACITY) ?
MAXIMUM_CAPACITY : tableSizeFor((int)size);
// sizeCtl表示初始化的数组容量
this.sizeCtl = cap;
} 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());
int binCount = 0;
// 相当于while(true)
for (Node[] tab = table;;) {
Node f; int n, i, fh;
// 如果tab为空则初始化tab
if (tab == null || (n = tab.length) == 0)
tab = initTable();
// tab不为空，tab相应位置上没有元素，直接cas赋值， *** 作成功就退出死循环
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
if (casTabAt(tab, i, null,
new Node(hash, key, value, null)))
break; // no lock when adding to empty bin
}
// 检查table[i]的节点的hash是否等于MOVED，如果等于，则检测到正在扩容，则帮助其扩容
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
// 不等于MOVED的情况下发生碰撞
else {
V oldVal = null;
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) {
binCount = 1;
// 遍历链表，比较每个结点的hash值和key值，如果有一样的，根据onlyIfAbsent的值决定是否覆盖value
for (Node e = f;; ++binCount) {
K ek;
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
// 如果遍历到链表尾部都没有找到一样的，就在尾部插入结点
Node pred = e;
if ((e = e.next) == null) {
pred.next = new Node(hash, key,
value, null);
break;
}
}
}
// 红黑树情况
else if (f instanceof TreeBin) {
Node p;
binCount = 2;
if ((p = ((TreeBin)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
// 判断插入结点后是否需要转换成树
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
} 插入 *** 作的流程：
- 根据给定的key求二次哈希值，得到要插入的结点在table数组中的位置index
- 如果table为空，初始化table
- table[index]==null，则没有发生碰撞，直接赋值
- table[index]处于MOVED状态，说明该位置正在扩容，则帮助它扩容
- table[index]!=null并且不是MOVED状态，说明发生碰撞。如果是链表，遍历链表，如果有相同的结点就覆盖（如果onlyIfAbsent=false），没有就插入到链表尾部；如果是树则插入到树中
- 如果是链表，判断插入结点后是否需要转换为树
public V get(Object key) {
Node[] tab; Node e, p; int n, eh; K ek;
int h = spread(key.hashCode());
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
// 头结点就是目标结点
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
// 目标结点在树中
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
// 目标结点在链表中
while ((e = e.next) != null) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
} // ConcurrentHashMap的第一个扩容方法，初始化table数组
private final Node[] initTable() {
Node[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
// sizeCtl<0表示正在初始化或正在被其他线程扩容
if ((sc = sizeCtl) < 0)
Thread.yield(); // lost initialization race; just spin
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
Node[] nt = (Node[])new Node[n];
table = tab = nt;
// sc=0.75*capacity
sc = n - (n >>> 2);
}
} finally {
// sizeCtl = 0.75*Capacity,为扩容阈值
sizeCtl = sc;
}
break;
}
}
return tab;
} // ConcurrentHashMap的第二个扩容方法
private final void tryPresize(int size) {
int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
tableSizeFor(size + (size >>> 1) + 1);
int sc;
// 只有大于等于0才表示该线程可以扩容
while ((sc = sizeCtl) >= 0) {
Node[] tab = table; int n;
if (tab == null || (n = tab.length) == 0) {
n = (sc > c) ? sc : c;
// 期间没有其他线程对表 *** 作，则CAS将SIZECTL状态置为-1，表示正在进行初始化
if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if (table == tab) {
@SuppressWarnings("unchecked")
Node[] nt = (Node[])new Node[n];
table = nt;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
}
}
// 若扩容值不大于原阀值，或现有容量>=最值，什么都不用做了
else if (c <= sc || n >= MAXIMUM_CAPACITY)
break;
// table不为空，且在此期间其他线程未修改table
else if (tab == table) {
int rs = resizeStamp(n);
if (sc < 0) {
Node[] nt;
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);
}
}
} tryPresize()方法调用的transfer()来具体实现扩容：
private final void transfer(Node[] tab, Node[] nextTab) {
int n = tab.length, stride;
//我们把一个线程一次负责转移hash槽的动作称为一个批次，n为扩容前的数组容量，
//首先根据可用CPU核心数，确定一个批次负责的hash槽的数量，
//CPU核心数有多个时，取 n/(8* NCPU)，最后与 最小单批次负责转移的hash槽数比较，取较大者。MIN_TRANSFER_STRIDE = 16.
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
if (nextTab == null) { // initiating
try {
@SuppressWarnings("unchecked")
//扩容后的数组容量为原来的2倍
Node[] nt = (Node[])new Node[n << 1];
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
//transferIndex存放的是扩容时下一次可以转移的hash槽的下标+1；即如果n=32，那么首次开始时transferIndex=32
transferIndex = n;
}
int nextn = nextTab.length;
ForwardingNode fwd = new ForwardingNode(nextTab);
// 是否可以推进，如转移了table[n-1]完成后，是否可以向前分配到table[n-2]
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
for (int i = 0, bound = 0;;) {
Node f; int fh;
while (advance) {
//下一次分配transfer段内最大hash数组索引index;即如果数组长度为32，扩容到64，
//那么首次分配时第一个transfer最大nextIndex=32，下一次就是 nextIndex=16
//下一次分配给transfer段内最小hash数组索引index;即如果数组长度为32，扩容到64，
//那么首次分配时第一个transfer最大nextBound=16，下一次就是 nextBound=0
int nextIndex, nextBound;
// 判断当前转移的位置是否已经到分配的transfer段内下临界值
if (--i >= bound || finishing)
advance = false;
// 判断是否已经没有可以分配的transfer，如果相等，说明扩容时所有的hash槽都已经被执行扩容的线程领取完毕
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
// 分配下一个段transfer
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
bound = nextBound;
i = nextIndex - 1;
advance = false;
}
}
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {
nextTable = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1);
return;
}
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
finishing = advance = true;
i = n; // recheck before commit
}
}
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {
synchronized (f) {
if (tabAt(tab, i) == f) {
// ln, hn ：这里定义了 lowerNode 和 higherNode两节点，
// 分别用来存储后续rehash之后哪些应该放在低位（nextTab 低位index），哪些应该放在高位（nextTab 高位index）
// 因为这些节点放在原数组同一位置，说明hash & (n-1) 相同，如果n=32，n-1=00001111，即它们的hash值后几位都是一样的
// 所以runBit = fh & n 只有两种结果：0或32，根据结果分为ln和hn
Node ln, hn;
if (fh >= 0) {
int runBit = fh & n;
Node lastRun = f;
// 这一步是遍历整个链表，找到最后几个结点，它们 hash & n 的结果相同，循环结束后直接批量赋值到ln或hn
// 这些结点被重用
for (Node p = f.next; p != null; p = p.next) {
int b = p.hash & n;
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
if (runBit == 0) {
ln = lastRun;
hn = null;
}
else {
hn = lastRun;
ln = null;
}
// 再一次遍历链表，只遍历剩下的结点，根据ph & n 的结果赋值给ln或hn
// 这些节点是重新创建的，使用头插法
for (Node 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(ph, pk, pv, ln);
else
hn = new Node(ph, pk, pv, hn);
}
// 最后将ln和hn放入新的数组中，链表的顺序是重用的结点和原始数组中的顺序一样
// 但是重新创建的结点和原本的顺序相反，因为使用了头插法
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
else if (f instanceof TreeBin) {
TreeBin t = (TreeBin)f;
TreeNode lo = null, loTail = null;
TreeNode hi = null, hiTail = null;
int lc = 0, hc = 0;
for (Node e = t.first; e != null; e = e.next) {
int h = e.hash;
TreeNode p = new TreeNode
(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;
}
}
ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
(hc != 0) ? new TreeBin(lo) : t;
hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
(lc != 0) ? new TreeBin(hi) : t;
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
}
}
}
}
}
写吐了 就这样吧
参考https://javaguide.cn/java/collection/concurrent-hash-map-source-code.html#_1-concurrenthashmap-1-7
https://blog.csdn.net/fuyuwei2015/article/details/72630365
https://blog.csdn.net/qq_37659383/article/details/82924833
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）