ConcurrentHashMap源码解析_03 put方法源码分析
Posted 兴趣使然の草帽路飞
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了ConcurrentHashMap源码解析_03 put方法源码分析相关的知识，希望对你有一定的参考价值。
put()方法是并发HashMap源码分析的重点方法，这里涉及到并发扩容，桶位寻址等等…- JDK1.8 ConcurrentHashMap结构图：
1、put方法源码解析
// 向并发Map中put一个数据
public V put(K key, V value) {
return putVal(key, value, false);
}
// 向并发Map中put一个数据
// Key: 数据的键
// value：数据的值
// onlyIfAbsent：是否替换数据：
// 如果为false，则当put数据时，遇到Map中有相同K,V的数据，则将其替换
// 如果为true，则当put数据时，遇到Map中有相同K,V的数据，则不做替换，不插入
final V putVal(K key, V value, boolean onlyIfAbsent) {
// 控制k 和 v 不能为null
if (key == null || value == null) throw new NullPointerException();
// 通过spread方法，可以让高位也能参与进寻址运算，使最终得到的hash值更分散
int hash = spread(key.hashCode());
// binCount表示当前k-v 封装成node插入到指定桶位后，在桶位中的所属链表的下标位置
// =0 表示当前桶位为null，node可以直接放入
// >0 表示当前桶位中有节点，遍历链表，将封装k-v的新node放入链表末尾，并记录链表末尾的下标binCount
// =2 表示当前桶位**可能**已经树化为红黑树了
int binCount = 0;
// tab 引用map对象的table
// 自旋
for (Node<K,V>[] tab = table;;) {
// f 表示桶位的头结点
// n 表示散列表数组的长度
// i 表示key通过寻址计算后，得到的桶位下标
// fh 表示桶位头结点的hash值
Node<K,V> f; int n, i, fh;
// -----------------------------------------------------------------------------
// CASE1：成立，表示当前map中的table尚未初始化...
if (tab == null || (n = tab.length) == 0)
// 对map中的table进行初始化
tab = initTable();
// -----------------------------------------------------------------------------
// CASE2：table已经初始化，此时根据寻址算法确定一个桶，并且桶的头结点f为null
// i 表示key使用路由寻址算法得到key对应table数组的下标位置，tabAt方法获取指定桶位i的头结点f
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
// 这时候就可以将封装k-v的结点直接放入桶
// casTabAt通过CAS的方式去向Node数组指定位置i设置节点值，设置成功返回true 否则返回false
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break;
}
// -----------------------------------------------------------------------------
// CASE3：table已经初始化，寻址得到的桶位中的头结点f不是null，如果该头结点f的hash值fh=-1：
// 则，表示当前节点是FWD(forwarding)节点
// 如果CASE3条件成立：表示当前桶位的头结点为 FWD结点，表示目前map正处于扩容过程中..
else if ((fh = f.hash) == MOVED)
// 发现f是FWD结点后，当前结点有义务去帮助当前map对象完成数据迁移工作...
// 等学完扩容再来分析这里~
tab = helpTransfer(tab, f);
// -----------------------------------------------------------------------------
// CASE4：CASE1-3都不满足时，那么当前桶位存放的可能是链表也可能是红黑树代理结点TreeBin
else {
// 保留替换之前的数据引用：当新插入的key存在后，会将旧值赋给OldVal，返回给put方法调用
V oldVal = null;
// 使用synchronized加锁“头节点”（**理论上是“头结点”**)
synchronized (f) {
// -----------------------------------------------------------------------
// CASE5：tabAt(tab, i) == f
// 对比一下当前桶位的头节点是否为之前获取的头结点：为什么又要对比一下？
// 如果其它线程在当前线程之前将该桶位的头结点修改掉了，当前线程再使用sync对该节点f加锁就有问题了（锁本身加错了地方~）
if (tabAt(tab, i) == f) {// 如果条件成立，说明加锁的对象f没有问题，持有锁！
// ------------------------------------------------------------------
// CASE6：fh >= 0)
// 如果条件成立，说明当前桶位就是普通链表桶位,这里回顾下常量属性字段：
// static final int MOVED = -1; 表示当前节点是FWD(forwarding)节点
// static final int TREEBIN = -2; 表示当前节点已经树化
if (fh >= 0) {
// 1.当前插入key与链表当中所有元素的key都不一致时，当前的插入操作是追加到链表的末尾，binCount此时表示链表长度
// 2.当前插入key与链表当中的某个元素的key一致时，当前插入操作可能就是替换了。binCount表示冲突位置（binCount - 1）
binCount = 1;
// 迭代循环当前桶位的链表，e是每次循环处理节点。
for (Node<K,V> e = f;; ++binCount) {
// 当前循环遍历节点的key
K ek;
// --------------------------------------------------------
// CASE7：
// 条件一：e.hash == hash
// 成立：表示循环的当前元素的hash值与插入节点的hash值一致,需要进一步判断
// 条件二：((ek = e.key) == key ||(ek != null && key.equals(ek)))
// 成立：说明循环的当前节点与插入节点的key一致，确实发生冲突了~
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
// 将当前循环的元素的值赋值给oldVal
oldVal = e.val;
// 传入的putVal方法参数onlyIfAbsent：是否替换数据：
// false，当put数据时，遇到Map中有相同K,V的数据，则将其替换
// true，当put数据时，遇到Map中有相同K,V的数据，则不做替换，不插入
if (!onlyIfAbsent)
e.val = value;
break;
}
// --------------------------------------------------------
// CASE8：
// 当前元素与插入元素的key不一致时，会走下面程序:
// 1.更新循环遍历的节点为当前节点的下一个节点
// 2.判断下一个节点是否为null，如果是null，说明当前节点已经是队尾了，插入数据需要追加到队尾节点的后面。
Node<K,V> pred = e;
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
// ------------------------------------------------------------------
// CASE9：f instanceof TreeBin
// 如果条件成立，表示当前桶位是红黑树代理结点TreeBin
//（前置条件：该桶位一定不是链表）
else if (f instanceof TreeBin) {
// p表示红黑树中如果与你插入节点的key 有冲突节点的话，则putTreeVal方法会返回冲突节点的引用。
Node<K,V> p;
// 强制设置binCount为2，因为binCount <= 1 时有其它含义，所以这里设置为了2 （回头讲addCount的时候会详细介绍）
binCount = 2;
// 条件一成立，说明当前插入节点的key与红黑树中的某个节点的key一致，冲突了：
// putTreeVal向红黑树中插入结点，插入成功返回null，否则返回冲突结点对象
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
// 将冲突节点的值赋值给oldVal
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
// ------------------------------------------------------------------
// CASE10：binCount != 0
// 说明当前桶位不为null，可能是红黑树也可能是链表
if (binCount != 0) {
// 如果binCount>=8 表示处理的桶位一定是链表
if (binCount >= TREEIFY_THRESHOLD)
// 因为桶中链表长度大于了8，需要树化：
// 调用转化链表为红黑树的方法
treeifyBin(tab, i);
// 说明当前线程插入的数据key，与原有k-v发生冲突，需要将原数据v返回给调用者。
if (oldVal != null)
return oldVal;
break;
}
}
}
// Map中的元素数据量累加方法：有额外的以下功能
// 1.统计当前table一共有多少数据
// 2.并判断是否达到扩容阈值标准，触发扩容。
addCount(1L, binCount);
return null;
}
2、initTable方法源码分析
第一次放元素时，初始化桶数组：
/**
* table初始化
*/
private final Node<K,V>[] initTable() {
// tab: 引用map.table
// sc: sizeCtl的临时值
// sizeCtl：默认为0，用来控制table的状态、以及初始化和扩容操作:
// sizeCtl<0表示table的状态:
//（1）=-1，表示有其他线程正在进行初始化操作。（其他线程就不能再进行初始化，相当于一把锁）
//（2）=-(1 + nThreads)，表示有n个线程正在一起扩容。
// sizeCtl>=0表示table的初始化和扩容相关操作：
//（3）=0，默认值，后续在真正初始化table的时候使用，设置为默认容量DEFAULT_CAPACITY --> 16。
//（4）>0，将sizeCtl设置为table初始容量或扩容完成后的下一次扩容的门槛。
Node<K,V>[] tab; int sc;
// 附加条件的自旋： 条件是map.table尚未初始化...
while ((tab = table) == null || tab.length == 0) {
// -----------------------------------------------------------------------------
// CASE1: sizeCtl) < 0
// sizeCtl < 0可能是以下2种情况：
//（1）-1，表示有其他线程正在进行table初始化操作。
//（2）-(1 + nThreads)，表示有n个线程正在一起扩容。
if ((sc = sizeCtl) < 0)
// 这里sizeCtl大概率就是-1，表示其它线程正在进行创建table的过程，当前线程没有竞争到初始化table的锁，进而当前线程被迫等待...
Thread.yield();
// -----------------------------------------------------------------------------
// CASE2：sizeCtl) >= 0 且U.compareAndSwapInt(this, SIZECTL, sc, -1)结果为true
// U.compareAndSwapInt(this, SIZECTL, sc, -1)：以CAS的方式修改当前线程的sizeCtl为-1，
// sizeCtl如果成功被修改为-1，就返回true，否则返回false。
// 当返回true时，则该线程就可以进入下面的else if代码块中，这时候sizeCtl=-1相当于是一把锁，表示下面的else if代码块已经被该线程占用，其他线程不能再进入~
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
// ----------------------------------------------------------------------
// CASE3: 这里为什么又要判断呢？
// 为了防止其它线程已经初始化table完毕了，然后当前线程再次对其初始化..导致丢失数据。
// 如果条件成立，说明其它线程都没有进入过这个if块，当前线程就是具备初始化table权利了。
if ((tab = table) == null || tab.length == 0) {
// sc大于等于0的情况如下：
//（3）=0，默认值，后续在真正初始化table的时候使用，设置为默认容量DEFAULT_CAPACITY --> 16。
//（4）>0，将sizeCtl设置为table初始容量或扩容完成后的下一次扩容的门槛。
// 如果sc大于0,则创建table时使用sc为指定table初始容量大小，
// 否则使用16默认值DEFAULT_CAPACITY
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
// 创建新数组nt
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
// 将新数组nt赋值给table、tab
table = tab = nt;
// sc设置为下次散列表扩容的门槛：0.75n
sc = n - (n >>> 2);
}
} finally {
// 将sc赋值给sizeCtl，分为一下2种情况：
// 1、当前线程既通过了CASE2的判断，也通过了CASE3的判断：
// 则，当前线程是第一次创建map.table的线程，那么，sc就表示下一次扩容的阈值。
// 2、当线程通过了CASE2的判断，但是没有通过CASE3的判断：
// 则，当前线程并不是第一次创建map.table的线程，当前线程通过CASE2的判断时，将
// sizeCtl设置为了-1 ，那么在该线程结束上面的代码逻辑之后，需要将sc修改回进入CASE2之前的sc值。
sizeCtl = sc;
}
break;
}
}
return tab;
}
小节：
（1）使用CAS锁控制只有一个线程初始化桶数组；
（2）sizeCtl在初始化后存储的是扩容门槛；
（3）扩容门槛写死的是桶数组大小的0.75倍，桶数组大小即map的容量，也就是最多存储多少个元素。
3、addCount方法(难点)
在阅读addCount方法源码之前，最好再去熟悉下LongAdder源码：LongAdder源码解析。
addCount方法作用：每次添加元素后，元素数量加1，并判断是否达到扩容门槛，达到了则进行扩容或协助扩容。
/**
* Map中的元素数据量累加方法：有额外的以下功能
* 1.统计当前table一共有多少数据
* 2.并判断是否达到扩容阈值标准，触发扩容
*/
private final void addCount(long x, int check) {
// as 表示 LongAdder.cells
// b 表示LongAdder.base
// s 表示当前map.table中元素的数量
CounterCell[] as; long b, s;
// -------------------------统计当前table一共有多少数据----------------------------------
// CASE1:
// (as = counterCells) != null
// 条件一：true->表示cells已经初始化了，当前线程应该去使用hash寻址找到合适的cell 去累加数据
// false->表示当前线程应该直接将数据累加到base（没有线程竞争）
// !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)
// 条件二：false->表示写base成功，数据累加到base中了，当前竞争不激烈，不需要创建cells
// true->表示写base失败，与其他线程在base上发生了竞争，当前线程应该去尝试创建cells。
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
// 有几种情况进入到if块中？
// 条件一为true->表示cells已经初始化了，当前线程应该去使用hash寻址找到合适的cell去累加数据
// 条件二为true->表示写base失败，与其他线程在base上发生了竞争，当前线程应该去尝试创建cells。
// a 表示当前线程hash寻址命中的cell
CounterCell a;
// v 表示当前线程写cell时的期望值
long v;
// m 表示当前cells数组的长度
int m;
// true -> 未发生线程竞争 false->发生线程竞争
boolean uncontended = true;
// ---------------------------------------------------------------------------
// CASE2:
// 条件一：as == null || (m = as.length - 1) < 0
// true-> 表示当前线程是通过写base竞争失败（CASE1的条件二），然后进入的if块，就需要调用fullAddCount方法去扩容或者重试..
// （fullAddCount方法就类似于LongAdder.longAccumulate方法）
// 条件二：a = as[ThreadLocalRandom.getProbe() & m]) == null 前置条件：cells已经初始化了
// true->表示当前线程命中的cell表格是个空的，需要当前线程进入fullAddCount方法去初始化cell，放入当前位置.
// 条件三：!(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x)
// 前置条件：条件二中当前线程命中的cell表格不是空的~
// false->取反得到false，表示当前线程使用cas方式更新当前命中的cell成功
// true->取反得到true,表示当前线程使用cas方式更新当前命中的cell失败，需要进入fullAddCount进行重试或者扩容cells。
if (as == null || (m = as.length - 1) < 0 ||
(a = as[ThreadLocalRandom.getProbe() & m]) == null ||
!(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
) {
fullAddCount(x, uncontended);
// 考虑到fullAddCount里面的事情比较累，就让当前线程不参与到扩容相关的逻辑了，直接返回到调用点。
return;
}
if (check <= 1)
return;
// sumCount统计当前散列表元素个数sum = base + cells[0] + ... + cells[n]，这是一个期望值
s = sumCount();
}
// -------------------------判断是否达到扩容阈值标准，触发扩容----------------------------
// CASE3:
// check >= 0表示一定是一个put操作调用的addCount，check < 0是remove操作调用的addCount方法
if (check >= 0) {
// tab 表示 map.table
// nt 表示 map.nextTable
// n 表示map.table数组的长度
// sc 表示sizeCtl的临时值
Node<K,V>[] tab, nt; int n, sc;
/**
* sizeCtl < 0
* 1. -1 表示当前table正在初始化（有线程在创建table数组），当前线程需要自旋等待..
* 2.表示当前table数组正在进行扩容 ,高16位表示：扩容的标识戳 低16位表示：（1 + nThread） 当前参与并发扩容的线程数量
*
* sizeCtl = 0，表示创建table数组时 使用DEFAULT_CAPACITY为大小
*
* sizeCtl > 0
*
* 1. 如果table未初始化，表示初始化大小
* 2. 如果table已经初始化，表示下次扩容时的 触发条件（阈值）
*/
// 有条件自旋~
// 条件一：s >= (long)(sc = sizeCtl)
// true -> 1.当前sizeCtl为一个负数 表示正在扩容中..
// 2.当前sizeCtl是一个正数，表示扩容阈值，但是s已经达到扩容阈值（需要扩容）
// false -> 表示当前table尚未达到扩容阈值条件（不需要扩容）
// 条件二：(tab = table) != null 恒成立 true
// 条件三：(n = tab.length) < MAXIMUM_CAPACITY
// true -> 当前table长度小于最大值限制，则可以进行扩容。
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
// 获取扩容唯一标识戳
// eg: 16 -> 32 扩容标识戳为：1000 0000 0001 1011
// 什么意思呢？
// 即，所有将map容量由16扩容到32的线程，其拿到的扩容唯一标识戳都是1000 0000 0001 1011
int rs = resizeStamp(n);
// --------------------------------------------------------------------------
// CASE4:
// 条件成立：表示当前table正在扩容，则，当前线程理论上应该协助table完成扩容
if (sc < 0) {
// --------------------------------------------------------------
// CASE2: 条件1~4只要有个为true就会跳出循环，不会继续扩容~
// 条件一：(sc >>> RESIZE_STAMP_SHIFT) != rs
// true -> 说明当前线程获取到的扩容唯一标识戳 非 本批次扩容
// false -> 说明当前线程获取到的扩容唯一标识戳 是 本批次扩容
// 条件二：JDK1.8 中有bug,jira已经提出来了,其实想表达的是 sc == (rs << 16 ) + 1
// true -> 表示扩容完毕，当前线程不需要再参与进来了
// false -> 扩容还在进行中，当前线程可以参与
// 条件三：JDK1.8 中有bug,jira已经提出来了,其实想表达的是:
// sc == (rs << 16) + MAX_RESIZERS
// true-> 表示当前参与并发扩容的线程达到了最大值 65535 - 1
// false->表示当前线程可以参与进来
//条件四：(nt = nextTable) == null
// true -> 表示本次扩容结束
// false -> 扩容正在进行中
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
// 条件1~4只要有个为true就会跳出循环，不会继续扩容~
break;
// --------------------------------------------------------------
// CASE5:
// 前置条件：当前table正在执行扩容中（即，CASE4没有被通过）当前线程有机会参与进扩容。
// 条件成立：说明当前线程成功参与到扩容任务中，并且将sc低16位值加1，表示多了一个线程参与工作~
// 条件失败：
// 1.当前有很多线程都在此处尝试修改sizeCtl，有其它一个线程修改成功了
// 导致你的sc期望值与内存中的值不一致，CAS修改失败。(下次自选大概率不会在来到这里~)
// 2.transfer任务内部的线程也修改了sizeCtl。
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
// 扩容（迁移数据）：协助扩容线程，持有nextTable参数，进入该方法协助扩容~
transfer(tab, nt);
}
// --------------------------------------------------------------------------
// CASE6: 以CAS的方式去更新sc，更新成功返回true，否则返回false
// 条件成立，说明当前线程是触发扩容的**第一个**线程，在transfer方法需要做一些扩容准备工作:
// rs << RESIZE_STAMP_SHIFT：将扩容唯一标识戳左移16位 eg:
// 1000 0000 0001 1011 << 16 得到 1000 0000 0001 1011 0000 0000 0000 0000
// 紧接这 (rs << RESIZE_STAMP_SHIFT) + 2)操作：
// 1000 0000 0001 1011 0000 0000 0000 0000 + 2
// => 1000 0000 0001 1011 0000 0000 0000 0010，这个二进制数有如下含义:
// sizeCtl的高16位: 1000 0000 0001 1011 表示当前的扩容标识戳
// sizeCtl的低16位: 0000 0000 0000 0010 表示(1 + nThread)，即，目前有多少个线程正在参与扩容~
// 那么这里的n是怎么来的呢？
// eg: (rs << RESIZE_STAMP_SHIFT) + 2 这里的2，就是1 + 1，后面的1就是对1*Thread，即(1+1*Threads)
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
// 扩容（迁移数据）：触发扩容条件的线程 不持有nextTable
transfer(tab, null);
// 重新计算table中的元素个数
s = sumCount();
}
}
}
小节：
（1）元素个数的存储方式类似于LongAdder类，存储在不同的段上，减少不同线程同时更新size时的冲突；
（2）计算元素个数时把这些段的值及baseCount相加算出总的元素个数；
（3）正常情况下sizeCtl存储着扩容门槛，扩容门槛为容量的0.75倍；
（4）扩容时sizeCtl高位存储扩容邮戳(resizeStamp)，低位存储扩容线程数加1（1+nThreads）；
（5）其它线程添加元素后如果发现存在扩容，也会加入的扩容行列中来；
以上就是ConcurrentHashMap源码的put存入数据的方法以及相关方法，由于transfer 迁移数据这个方法比较复杂，我们放在下一篇文章中单独分析~
以上是关于ConcurrentHashMap源码解析_03 put方法源码分析的主要内容，如果未能解决你的问题，请参考以下文章
Java小白进阶系列——ConcurrentHashMap源码解析文章总目录
ConcurrentHashMap源码解析_01 成员属性内部类构造方法分析
ConcurrentHashMap源码解析_01 成员属性内部类构造方法分析