C/C++教程
ConcurrentHashMap源码解析3.put()方法
本文主要是介绍ConcurrentHashMap源码解析3.put()方法，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！
1.putVal()方法
写数据大体流程
写前操作
1、ConcurrentHashMap 不允许key或value为NULL，会抛出异常
2、写数据前，会先对key的hash值进行一次加工spread()
写数据流程
整个写数据是一个 自旋(死循环) 的操作。
- 情况一: 当前的table还没有被初始化。调用initTable()去
尝试初始化table，然后继续自旋. - 情况二: 当前table已经被初始化，调用寻址算法
(hash & (table.length - 1))得到的桶位上的元素为NULL，尝试使用CAS操作将构造的节点写入桶位。成功退出循环，失败继续自旋。 - 情况三：当前table已经被初始化了，但是当前桶位的节点是FWD节点，说明此时table正在发生扩容操作，此时当前线程就去参与扩容。扩容后继续自旋。
- 情况四：说明当前桶位的形成了
链表或者红黑树，先使用synchronized加锁锁住当前桶位(锁对象就是当前桶位的头节点)，然后判断如果当前桶位形成了链表，就遍历链表中的节点去找是否存在和待插入的节点的key完全一致的节点，如果有的话就进行value替换，没有就将节点插入到链表末尾。 如果当前桶位已经形成了红黑树，就在树中找是否存在一个节点的key和待插入节点完全一致的节点，存在就将value替换。不存在就插入到树中。
写后操作
- ①如果当前桶位是链表的话，会检查是否达到了树化标准然后去树化。
②如果待插入节点跟桶位上的某一个节点冲突后进行替换了(无论是链表还是红黑树)，直接将冲突节点的value返回，不执行③的逻辑 (因为这是一次替换操作，没有新增节点)- ③没有发生冲突，说明这是一次添加操作，需要调用 addCount() 方法，内部先对table中的节点进行计数，然后判断是否达到扩容标准执行扩容逻辑
public V put(K key, V value) {
//调用了putVal方法。
return putVal(key, value, false);
}
/*
* @param key 元素的key
* @param value 元素value
* @param onlyIfAbsent 是否替换数据
* false --> 当put数据时，遇到Map中有相同k-v的数据，将其替换
* true --> 遇到Map中有相同k-v的数据，不替换，不进行插入
*/
final V putVal(K key, V value, boolean onlyIfAbsent) {
/* 注意 这里跟HashMap不同，
* HashMap允许key为null或者value为null， ConcurrentHashMap不允许key或者
* value为null。
*/
if (key == null || value == null) throw new NullPointerException();
//通过spread方法重新计算hash值(具体参考源码解析2 spread方法)
int hash = spread(key.hashCode());
/*
* binCount表示当前k-v 封装成node后插入到指定桶位后，在桶位中的所属链表的下标位置
* 0 当前桶位为null，node可以直接放
* 2 表示当前桶位可能是红黑树
*/
int binCount = 0;
/*
* 自旋
* tab 引用哈希表
*/
for (Node<K,V>[] tab = table;;) {
/*
* f 表示桶位的头节点
* n 表示散列表的长度
* i 表示key通过寻址计算后得到的桶位坐标
* fh 桶位头节点的哈希值
*/
Node<K,V> f; int n, i, fh;
/*
* CASE1 ：表示table尚未初始化
*
* 进入initTable()方法初始化 尝试初始化table。最终返回初始化完成的table。
*/
if (tab == null || (n = tab.length) == 0)
tab = initTable();
/*
* CASE2 ：table已经初始化，且通过寻址后发现当前桶位头结点为null
* 寻址算法： ((table.length - 1) & hash) 跟HashMap寻址算法一致
*
* 进入casTabAt尝试使用CAS添加Node。
*/
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//期望tab的第i个位置是NULL，是NULL的话就将创建的Node赋值上去
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
//设置成功，结束自旋。失败(有其他线程写成功了) 走其他逻辑继续自旋
break;
}
/*
* CASE3：table已经初始化，且通过寻址后发现当前桶位头结点为FWD节点，表示当前
* map正处于扩容(迁移)过程中。
* (MOVED == -1 表示当前节点是FWD节点)
*
* 进入helpTransfer有义务帮助当前map对象完成迁移数据的工作。
*/
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
/*
* CASE4：当前桶位(不为NULL)可能是链表 也可能是 红黑树代理类TreeBin
*/
else {
//oldVal 当插入节点的key存在时，会将旧值返回。
V oldVal = null;
/*
* synchronized加锁给当前桶位的头节点(理论上是头结点)。(只对当前桶位加
* 锁，不锁整个Map)
*/
synchronized (f) {
/*
* 为什么这里又要对比一下，看看当前桶位的头几点，是否是之前获取的头
* 结点？
* (tabAt()就是获取i位置上的节点)
* 为了避免其他线程已经将该桶位的头结点改掉了，导致当前线程加错锁，
* 就会出现问题，如果这里对比失败的话，就不需要在执行了。
*
* 如果条件成立，就可以进来造了!
*/
if (tabAt(tab, i) == f) {
//条件成立，说明当前桶位形成了链表。
if (fh >= 0) {
/*
* 1、
* 当前插入key与链表中的所有元素的key都不一致，当前的插入
* 操作就是将节点追加到链表的末尾，binCount表示链表长度。
* 2、
* 当前插入key与链表当中的某个元素的key一致时，当前插入操
* 作就是替换 binCount表示冲突位置(binCount-1)
*/
binCount = 1;
/*
* 遍历链表，判断是否有节点的key与当前要插入元素的key完全
* 一致，存在的话替换value，不存在插入到链表末尾。
*
*/
for (Node<K,V> 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<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) {
//p表示红黑树中如果与你插入节点的key有冲突时，会将这个冲突的节点赋给p
Node<K,V> p;
//强制设置binCount为2，因为binCount <= 1时有其他含义，
binCount = 2;
//p不为null，说明真的发生了冲突。
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value; //覆盖原key的值
}
}
}
}
//说明当前桶位不为null，可能是红黑树，也可能是链表
if (binCount != 0) {
//binCount >= 8，说明当前桶位一定是链表
if (binCount >= TREEIFY_THRESHOLD)
//可能树化(还得看table长度)
treeifyBin(tab, i);
if (oldVal != null)
//返回冲突节点的value。
return oldVal;
break;
}
}
}
/*
* 退出自旋之后，调用addCount
* 1、统计当前table一共有多少节点
* 2、判断是够达到扩容标准。
*/
addCount(1L, binCount);
return null;
}
2.initTable()方法
initTable方法流程。
当前线程尝试去初始化table，此时有两种情况
- sizeCtl的值为-1，表示当前有其他线程正在初始化table，当前线程先释放CPU执行权，然后再次尝试
直到table被初始化完成退出循环。 - sizeCtl的值大于等于0。当前线程使用
CAS的方式尝试去初始化table。初始化table后**(只有一个线程可以真正的初始化table)，将sizeCtl的值变为当前table长度的0.75，即扩容阈值**。
源码解析
private final Node<K,V>[] initTable() {
/*
* tab 表示table的引用
* sc 表示临时sizeCtl的值
*/
Node<K,V>[] tab; int sc;
/*
* 带条件(当前table尚未初始化)的自旋,当table被初始化后就退出。
*/
while ((tab = table) == null || tab.length == 0) {
/*
* 先判断sizeCtl的值，
* 1、sizeCtl < 0
* 1.1 sizeCtl = -1 表示有线程正在创建table数组，当前线程需要自旋等待
* 1.2 其他表示当前table数组正在扩容，高16位表示扩容的时间戳，低16位表示(1 + nThread)个线程正在并发扩容
* 2、sizeCtl = 0
* 表示创建table数组时，使用DEFAULT_CAPACITY(16)大小
*
* 3、sizeCtl > 0
* 两种情况
* 1.如果table未初始化，表示初始化大小
* 2.如果table已经初始化，表示下次扩容时的触发条件(阈值)
*/
if ((sc = sizeCtl) < 0)
//sizeCtl的值大概率为-1，即当前有其他线程正在创建table数组
//yield表示释放CPU执行权。
Thread.yield(); // lost initialization race; just spin
/*
* 有两种情况会走到这里
* 2、sizeCtl = 0
* 表示创建table数组时，使用DEFAULT_CAPACITY(16)大小
*
* 3、sizeCtl > 0
* 两种情况
* 1.如果table未初始化，表示初始化大小
* 2.如果table已经初始化，表示下次扩容时的触发条件(阈值)
*
* 通过调用UnSafe的CAS操作获取锁去修改sizeCtl的值
* 注意： 这里的CAS方法的SIZECTL常量就是sizeCtl变量的内存地址，即修改的就是sizeCtl变量的值。
*/
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
//仍然判断table是否为null，防止多线程情况下多次初始化。
if ((tab = table) == null || tab.length == 0) {
/*
* sc大于0，创建table时使用sc为指定大小，
* 否则使用默认值16初始化table
*/
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
//创建Node[].
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
//赋值给table
table = tab = nt;
// sc变为0.75n,即扩容阈值。
sc = n - (n >>> 2);
}
} finally {
/*
* 此时有两种情况
* 1、当前线程是第一次初始化table的线程，此时的sc就是扩容阈值
* 2、当前线程不是第一次初始化table的线程，所以不会进入初始化table
* 的逻辑，但是sizeCtl的值已经被改为-1了，最后需要将它的值改回来
*/
sizeCtl = sc;
}
break;
}
}
//最后返回tab。
return tab;
}
3.addCount()方法
addCount()方法的执行逻辑
1.先调用类似LongAdder的add逻辑，先尝试去写数据，
如果cells数组已经初始化了 或 尝试CAS写数据到base中失败了，就会继续判断cells数组的情况然后可能进入fullAddCount (就是LongAdder中的longAccumulate()方法)。
如果写成功，就会求一次节点个数总和。
注意：如果有线程真的调用了longAccumulate()，那么直接后return。既不求和也不参与table数组扩容的逻辑
2.进入扩容的逻辑，先计算出来一个扩容戳，然后自旋进行判断当前线程是否可以参与扩容的逻辑**(四个条件见源码解析)**，可以参与的话，使用CAS修改sizeCtl的值(此时sizeCtl为负数，高16位表示时间戳，低16位表示参与扩容的线程数)，CAS尝试将扩容线程+1,成功的话进入扩容逻辑，失败继续自旋。
private final void addCount(long x, int check) {
/*
* 这一部分代码就是LongAdder中的代码，先add()然后调用sum()去求和。
* as 表示LongAdder中的cells数组
* b 表示LongAdder中的base
* s 表示table中的节点个数
*/
CounterCell[] as; long b, s;
/*
* 进入if的两个条件(或的关系)
* 条件一： cells数组已经初始化了
* 条件二： cells数组未初始化，但是CAS尝试写base失败(发生了竞争)，
*/
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
/*
* a 表示当前线程寻址后命中的cell
* v 表示当前线程写cell时的期望值
* m 表示当前cells数组的长度
*/
CounterCell a; long v; int m;
//true表示未竞争，false表示发生了竞争
boolean uncontended = true;
/*
* fullAddCount()方法就是LongAdder中的longAccumulate()方法。
* 关于详细的longAccumulate()方法参考LongAdder源码解析
* 这里有三种情况会进入longAccumulate()方法
* 1、cells数组为NULL
* 2、cells数组不为NULL，但是当前线程寻址后命中的Cell为NULL。
* 3、cells数组不为NULL，并且命中的Cell也不为NULL，但是尝试CAS向Cell中写
* 数据失败
*/
if (as == null || (m = as.length - 1) < 0 ||
(a = as[ThreadLocalRandom.getProbe() & m]) == null ||
!(uncontended =
U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
//详细讲解参考 [https://blog.csdn.net/qq_46312987/article/details/121499330]
fullAddCount(x, uncontended);
/*考虑到fullAddCount()事情比较累，直接让当前线程不参与扩容的逻辑了。
直接返回了。*/
return;
}
if (check <= 1)
return;
//求和(就是LongAdder中的sum方法) 是一个期望值，(最终一致性)
s = sumCount();
}
//-------------------------------LongAdder Code ENd --------------------------
//----------------------------------扩容逻辑 Start-----------------------------
/*
* check >=0 表示一定是一个put操作调用的addCount
*/
if (check >= 0) {
/*
* tab 引用table
* nt 表示nextTable(扩容的table)
* n 表示table数组的长度
* sc 表示sizeCtl的临时值
*/
Node<K,V>[] tab, nt; int n, sc;
/*
* 先判断sizeCtl的值，走到这里sizeCtl可能的情况
* 1、sizeCtl < 0
* (不可能)1.1 sizeCtl = -1 表示有线程正在创建table数组，当前线程需要自旋等待
* 1.2 其他表示当前table数组正在扩容，高16位表示扩容的时间戳，低16位表示
* (1 + nThread)个线程正在并发扩容 (可能)
*
* 2、sizeCtl = 0
* 表示创建table数组时，使用DEFAULT_CAPACITY(16)大小 (不可能)
*
* 3、sizeCtl > 0
* 两种情况
* 1.如果table未初始化，表示初始化大小 (不可能)
* 2.如果table已经初始化，表示下次扩容时的触发条件(阈值) (可能)
*/
/*
* 自旋
* 条件一：s >= (long)(sc = sizeCtl)
* true -> 1.当前sizeCtl是一个负数，表示正在扩容
* false -> 2.当前sizeCtl是一个正数，表示扩容阈值，并且当前的元
* 素数量没有达到扩容阈值，无需扩容
*
* 条件二：(tab = table) != null 恒成立
*
* 条件三：(n = tab.length) < MAXIMUM_CAPACITY)
* true表示当前table长度小于最大值限制，可以进行扩容
*/
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
/*
* 扩容批次唯一表示戳
* 假设当前16 -> 32 计算出来的 rs = 1000 000 0001 1011
*/
int rs = resizeStamp(n);
/*
* sc小于0 表示当前table正在扩容
* 当前线程CAS尝试应该协助table完成扩容
*/
if (sc < 0) {
/*
* 当前线程是否可以进行扩容的条件。
*
* 条件一：
* (sc >>> RESIZE_STAMP_SHIFT) != rs
* true表示当前线程获得的扩容唯一表示戳 非 本批次扩容
* false表示当前线程获得的扩容唯一表示戳 是 本批次扩容
*
* 条件二：
* sc == rs + 1 (这里有BUG，后续JDK版本已经修改)
* 实际上想表达的是 sc == rs << 16 + 1
* true 表示扩容完毕，当前线程不需要参与
* false 表示扩容还在进行中，当前线程可以参与
*
* 条件三：
* sc == rs + MAX_RESIZERS (这里有BUG，后续JDK版本已修改)
* 实际想表达的是 sc == rs << 16 + MAX_RESIZERS
* true 表示当前参与的扩容线程数已经到了最大数 当前线程不需要参与
* false 表示当前当前参与的扩容线程数未到达最大数 当前线程可以参与
*
* 条件四：
* (nt = nextTable) == null
* true表示本次扩容结束(nextTable只有当在扩容时不为NULL，扩容结束就置为NULL)
* false表示本次扩容未结束
*/
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
/*
* 前置条件：当前table正在扩容中，当前线程有机会参与扩容
* 因为当前table正在扩容，说明sizeCtl的值一定为负数，低16位表示的
* 就是当前正在参与扩容的线程数,所以尝试+1，进入扩容的逻辑中。
*
* 尝试失败：
* 1.当前有很多线程都在此处尝试修改sizeCtl，有其他一个线程修改成
* 功，导致你的sc期望值与内存中的值不一致，修改失败
*
* 2.transfer任务内部的线程也修改了sizeCtl。
* 失败继续自旋。
*/
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
//协助扩容线程，持有nextTable参数
transfer(tab, nt);
}
/*
* CAS修改sizeCtl的值，成功的话将sizeCtl变为负数，说明当前现场是触发
* 扩容的第一个线程，在transfer()方法中需要做一些扩容的额外工作
*/
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
//触发扩容条件的线程，不持有nextTable。
transfer(tab, null);
//求和。
s = sumCount();
}
}
}
4.小结
-
（1）元素个数的存储方式类似于
LongAdder类，存储在不同的段上，减少不同线程同时更新size时的冲突； -
（2）计算元素个数时把这些段的值及baseCount相加算出总的元素个数；
-
（3）正常情况下sizeCtl存储着扩容门槛，扩容门槛为容量的 0.75倍；
-
（4）扩容时
sizeCtl(sizeCtl < 0)高位存储扩容戳(resizeStamp)，低位存储扩容线程数加1（1+nThreads） -
（5）其它线程添加元素后如果发现存在扩容，也会
尝试加入的扩容行列中来；
这篇关于ConcurrentHashMap源码解析3.put()方法的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！
您可能喜欢
-
【机器学习(二)】分类和回归任务-决策树(Decision Tree,DT)算法-Sentosa_DSML社区版11-25
-
增量更新怎么做？-icode9专业技术文章分享11-23
-
压缩包加密方案有哪些？-icode9专业技术文章分享11-23
-
用shell怎么写一个开机时自动同步远程仓库的代码？-icode9专业技术文章分享11-23
-
webman可以同步自己的仓库吗？-icode9专业技术文章分享11-23
-
在 Webman 中怎么判断是否有某命令进程正在运行？-icode9专业技术文章分享11-23
-
如何重置new Swiper?-icode9专业技术文章分享11-23
-
oss直传有什么好处？-icode9专业技术文章分享11-23
-
如何将oss直传封装成一个组件在其他页面调用时都可以使用？-icode9专业技术文章分享11-23
-
怎么使用laravel 11在代码里获取路由列表？-icode9专业技术文章分享11-23
-
怎么实现ansible playbook 备份代码中命名包含时间戳功能？-icode9专业技术文章分享11-22
-
ansible 的archive 参数是什么意思？-icode9专业技术文章分享11-22
-
ansible 中怎么只用archive 排除某个目录？-icode9专业技术文章分享11-22
-
exclude_path参数是什么作用？-icode9专业技术文章分享11-22
-
微信开放平台第三方平台什么时候调用数据预拉取和数据周期性更新接口？-icode9专业技术文章分享11-22
栏目导航
