看了大半天才看懂。总结一下,
putVal方法中6个步骤:
- 没有初始化就
initTable - 没有hash冲突直接CAS插入
- 如果要扩容, 当前位置有
fwd;ForwardingNode记录nextTab的指针，从而当前线程helpTransfer到哪里 - hash冲突,
synchronized(f)的Node对象锁, 分为linkedlist和红黑树讨论 - 长度>8, len>64: treeify
- 添加成功
addCount, 统计size, 检查是否需要扩容
size()方法, 用到了LongAdder的思想; addCount:CAS式加到baseCount, 如果过多就CounterCell, 分治思想, 最后fullAddCount合并
treeify之后ConcurrentHashMap每个元素是个TreeBin, HashMap是个TreeNode, TreeBin实际上是把整个红黑树看作一个元素，而不是TreeNode那样只看在slot中的Node,这是因为红黑树平衡的时候这个slot的root可能变了，那么synchronized(f)就失效；但是不应该失效，所以TreeBin看作整体就没有这个问题了。 TreeBin很重要的fields是root
扩容过程中重要的字段:
-
transferIndex: 总共要迁移当前tab哪些index -
stride: 每个线程负责迁移的步长 -
i,bound: while循环具体每个线程迁移的当前index和终止index -
advance: flag: 要不要往前找index迁移 -
finishing: 当前线程迁移任务结束没有 -
sizeCtl:private transient volatile int sizeCtl: 记录多线程下迁移元素的线程数量 - 用于迁移找高位链和低位链:
-
runBit := fh & n: 很巧妙，对比hash的index:= fh & (n-1);runBit只用来表示高X位，是为了知道当前链表的Node是在高位还是低位链 -
lastRun: 对应if(b != runBit):lastRun后面的所有Node的runBit都相同，可以提前结束计算;
图片来源
lastRun.png
-
- 未迁移的slot要进行
put: 是可以的，如果这个slot在迁移：synchronized(f): 想要put的线程阻塞; - 迁移的hash slot的
get:hn,ln高低位是复制引用的，所以get不受影响 -
get不需要加锁，因为Node:volatile val
transfer
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
//根据机器CPU核心数来计算，一条线程负责Node数组中多长的迁移量
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
//本线程分到的迁移量
//假设为16（默认也为16）
stride = MIN_TRANSFER_STRIDE; // subdivide range
//nextTab若为空代表线程是第一个进行迁移的
//初始化迁移后的新Node数组
if (nextTab == null) { // initiating
try {
@SuppressWarnings("unchecked")
//这里n为旧数组长度，左移一位相当于乘以2
//例如原数组长度16，新数组长度则为32
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变量为新数组
nextTable = nextTab;
//假设为16
transferIndex = n;
}
//假设为32
int nextn = nextTab.length;
//标示Node对象，此对象的hash变量为-1
//在get或者put时若遇到此Node，则可以知道当前Node正在迁移
//传入nextTab对象
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
while (advance) {
int nextIndex, nextBound;
//i为当前正在处理的Node数组下标，每次处理一个Node节点就会自减1
if (--i >= bound || finishing)
advance = false;
//假设nextIndex=16
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
//由以上假设，nextBound就为0
//且将nextIndex设置为0
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
//bound=0
bound = nextBound;
//i=16-1=15
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
}
}
//此时i=15，取出Node数组下标为15的那个Node，若为空则不需要迁移
//直接设置占位标示，代表此Node已处理完成
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
//检测此Node的hash是否为MOVED，MOVED是一个常量-1，也就是上面说的占位Node的hash
//如果是占位Node，证明此节点已经处理过了，跳过i=15的处理，继续循环
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {
//锁住这个Node
synchronized (f) {
//确认Node是原先的Node
if (tabAt(tab, i) == f) {
//ln为lowNode，低位Node，hn为highNode，高位Node
//这两个概念下面以图来说明
Node<K,V> ln, hn;
if (fh >= 0) {
//此时fh与原来Node数组长度进行与运算
//如果高X位为0，此时runBit=0
//如果高X位为1，此时runBit=1
int runBit = fh & n;
Node<K,V> lastRun = f;
for (Node<K,V> p = f.next; p != null; p = p.next) {
//这里的Node，都是同一Node链表中的Node对象
int b = p.hash & n;
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
//正如上面所说，runBit=0，表示此Node为低位Node
if (runBit == 0) {
ln = lastRun;
hn = null;
}
else {
//Node为高位Node
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;
//若hash和n与运算为0，证明为低位Node，原理同上
if ((ph & n) == 0)
ln = new Node<K,V>(ph, pk, pv, ln);
//这里将高位Node与地位Node都各自组成了两个链表
else
hn = new Node<K,V>(ph, pk, pv, hn);
}
//将低位Node设置到新Node数组中，下标为原来的位置
setTabAt(nextTab, i, ln);
//将高位Node设置到新Node数组中，下标为原来的位置加上原Node数组长度
setTabAt(nextTab, i + n, hn);
//将此Node设置为占位Node，代表处理完成
setTabAt(tab, i, fwd);
//继续循环
advance = true;
}
....
}
}
}
}
}
put:
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
//对key的hashCode进行散列
int hash = spread(key.hashCode());
int binCount = 0;
//一个无限循环，直到put操作完成后退出循环
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
//当Node数组为空时进行初始化
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//Unsafe类volatile的方式取出hashCode散列后通过与运算得出的Node数组下标值对应的Node对象
//此时的Node对象若为空，则代表还未有线程对此Node进行插入操作
else if ((f = tabAt(tab, i = (n - 1) & hash)) == 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);
else {
V oldVal = null;
//对Node对象进行加锁
synchronized (f) {
//二次确认此Node对象还是原来的那一个
if (tabAt(tab, i) == f) {
if (fh >= 0) {
binCount = 1;
//无限循环，直到完成put
for (Node<K,V> e = f;; ++binCount) {
K ek;
//和HashMap一样，先比较hash，再比较equals
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) {
//和链表头Node节点不冲突，就将其初始化为新Node作为上一个Node节点的next
//形成链表结构
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
...
}
