深入理解ConcurrentHashMap原理
首先提醒读者，这期难度较大，请大家吃好晕车药💊，如果在读后有任何问题都可以在评论区留言，本人会一一回复😁
什么是ConcurrentHashMap
上一期我们讲解了HashMap的原理，但是HashMap是线程不安全的类，那么ConcurrentHashMap就帮我们解决了并发的问题，接下来将带大家来进一步分析ConcurrentHashMap
ConcurrentHashMap底层数据结构
这里向大家介绍的是jdk8的ConcurrentHashMap，因为jdk8对HashMap进行了进一步的优化，使得锁的粒度更细，这里也会涉及更多的知识点，所以面试会很常问
HashMap采用的是数组+链表+红黑树的结构，其中并发问题是通过CAS+synchronized解决的，接下来就带大家从源码入手分析作者的思路
ConcurrentHashMap源码分析
我们可以思考一下我们应该从哪里入手了解ConcurrentHashMap呢？
首先从ConcurrentHashMap的类头入手
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
implements ConcurrentMap<K,V>, Serializable {
private static final long serialVersionUID = 7249069246763182397L;
// 最大的数组容量
private static final int MAXIMUM_CAPACITY = 1 << 30;
// 默认的数组容量数
private static final int DEFAULT_CAPACITY = 16;
// 能够将Map转化为数组的最大长度
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
// 并发级别，兼容1.8以前版本
private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
// 负载因子
private static final float LOAD_FACTOR = 0.75f;
// 链表转化为红黑树的阀值
static final int TREEIFY_THRESHOLD = 8;
// 红黑树退化为链表的阀值
static final int UNTREEIFY_THRESHOLD = 6;
// 最小链表转化为红黑树的数组容量
static final int MIN_TREEIFY_CAPACITY = 64;
// 如果下面扩容相关的常量不了解可以先有个印象下面会解释扩容的整个过程
// 扩容时转移数据的最小转移容量
private static final int MIN_TRANSFER_STRIDE = 16;
// 用于生成扩容戳
private static int RESIZE_STAMP_BITS = 16;
// 帮助扩容的最大线程数
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
// 扩容时移动指定位数可以得到扩容戳的前几位版本号
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
// 状态标记位
// 正在扩容移动数据
static final int MOVED = -1; // hash for forwarding nodes
// 节点是红黑树节点
static final int TREEBIN = -2; // hash for roots of trees
// 保留节点，只有在computerIfAbsent和computer时出现
static final int RESERVED = -3; // hash for transient reservations
// 用于计算key的hash值时把hash值控制到integer的最大值以内
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
// 获取当前运行系统的CPU核心数
static final int NCPU = Runtime.getRuntime().availableProcessors();
// 因为要在多线程中可见所以都加上了volatile
//（volatile不懂的可以给我评论或留言，之后会出一篇volatile使用和底层原理的文章）
// 数组
transient volatile Node<K,V>[] table;
// 下一个要使用的数组，只有在数组扩容时使用
private transient volatile Node<K,V>[] nextTable;
// 用于记录基本的键值对，具体计算size逻辑在后面展开
private transient volatile long baseCount;
// 表示初始化或者扩容的标记位，如果为负，则表示正在初始化或者扩容
private transient volatile int sizeCtl;
// 在进行扩容时的索引位
private transient volatile int transferIndex;
// 计数单元格是否频繁访问标记位
private transient volatile int cellsBusy;
// 用于计算size的计数单元格
private transient volatile CounterCell[] counterCells;
// views
private transient KeySetView<K,V> keySet;
private transient ValuesView<K,V> values;
private transient EntrySetView<K,V> entrySet;
复制代码其中扩容相关的常量值和扩容所创建的变量大家可能看着很晕，不知道什么意思，在进行分析源码时，这些值会一一为大家展开
接下来我们分析HashMap的构造函数
// ConcurrentHashMap有很多重载的构造函数，但是最全最基本的就是这个
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
// 并发级别，ConcurrentHashMap在1.8以后默认的并发级别是数组的容量，所以当初始容量小于并发级别时，把并发级别赋给初始容量
if (initialCapacity < concurrencyLevel) // Use at least as many bins
initialCapacity = concurrencyLevel; // as estimated threads
// 获取扩容容量
long size = (long)(1.0 + (long)initialCapacity / loadFactor);
// 将扩容容量转化为2的倍数
int cap = (size >= (long)MAXIMUM_CAPACITY) ?
MAXIMUM_CAPACITY : tableSizeFor((int)size);
this.sizeCtl = cap;
}
复制代码put()方法
接下来就开始揭开put()方法真实的面纱啦！！
// 常用的put方法
public V put(K key, V value) {
return putVal(key, value, false);
}
// 这是具体实现put操作的方法
final V putVal(K key, V value, boolean onlyIfAbsent) {
// hashMap是可以存在null值，但是ConcurrentHashMap不允许存在null，否则会抛出NPE
if (key == null || value == null) throw new NullPointerException();
// 根据key获取hash值
int hash = spread(key.hashCode());
int binCount = 0;
// 自旋
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
// 如果tab为null，初始化数组
if (tab == null || (n = tab.length) == 0)
// 初始化数组
tab = initTable();
// 通过key的hash值判断数组中的位置是否为null
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
}
// 如果位置的节点不是null，而且节点的hash为-1，则代表数组正在扩容，则去尝试帮助扩容数组
else if ((fh = f.hash) == MOVED)
// 帮助数组扩容
tab = helpTransfer(tab, f);
// 进行新增数据
else {
V oldVal = null;
// 锁住头节点，synchronized在jdk6被优化过了，性能已经很好
synchronized (f) {
// 再次判断头节点是否发生变化
if (tabAt(tab, i) == f) {
// 代表当前节点是链表节点，上面声明的常量都是负数，反之不是负数就是链表节点
if (fh >= 0) {
binCount = 1;
// 从头节点向后遍历
for (Node<K,V> e = f;; ++binCount) {
K ek;
// 节点的key和待插入的key相同，则覆盖value值
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;
// 直到遍历到最后都没发现key相同的情况，把新数据插入到链表尾部（jdk8采用的是尾插法）
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;
// 返回key相同的节点或者新增红黑树节点
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
// binCount默认是0，在进行添加节点时发生变化
if (binCount != 0) {
// 判断链表长度是否大于8
if (binCount >= TREEIFY_THRESHOLD)
// 在满足数组容量的条件时链表转红黑树
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
// 增加键值对数量
addCount(1L, binCount);
return null;
}
复制代码计算hash值
// 跟HashMap基本一致，只是增加了&HASH_BITS，防止最终的结果超过int最大值
static final int spread(int h) {
return (h ^ (h >>> 16)) & HASH_BITS;
}
复制代码初始化数组
// 初始化数组
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
// 因为存在并发，循环判断数组是否为null，其中tab是
while ((tab = table) == null || tab.length == 0) {
// 如果sc小于0代表数组已经在进行扩容或者已经正在初始化了，所以当前线程让出cpu执行权
if ((sc = sizeCtl) < 0)
Thread.yield(); // lost initialization race; just spin
// 还未进行扩容，CAS方式修改SIZECTL为-1
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
// 如果数组还是null
if ((tab = table) == null || tab.length == 0) {
// 如果传入了容量数进行扩容，否则取默认值进行初始化数组
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
// 初始化数组
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
// 可以转化为 n - n/4
sc = n - (n >>> 2);
}
} finally {
// 扩容完毕把sizeCtl更新为下一次待扩容数
sizeCtl = sc;
}
break;
}
}
// 最后初始化完毕，返回数组
return tab;
}
复制代码接下来看下帮助数组扩容的过程
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
Node<K,V>[] nextTab; int sc;
// 判断当前数组是否为null，当前节点是否是扩容节点，迁移所需的目标数组是否为null
if (tab != null && (f instanceof ForwardingNode) &&
(nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
// 根据当前数组长度，生成一个扩容戳， 扩容戳的高位代表扩容的版本号，低位是0
int rs = resizeStamp(tab.length);
// 自旋判断当前数组是否还在扩容
while (nextTab == nextTable && table == tab &&
(sc = sizeCtl) < 0) {
//(sc>>>RESIZE_STAMP_SHIFT)!=rs代表生成的扩容戳版本是否与正在扩容的扩容戳版本相同，不相同则不帮助扩容
// sc == rs + 1 代表扩容已经结束
// sc == rs + MAX_RESIZERS 代表帮助扩容的线程数已经达到最大值
// transferIndex <= 0 表示所有的扩容任务有已经结束
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || transferIndex <= 0)
break;
// 通过CAS增加扩容戳的值，是每有一个线程帮助扩容扩容戳都会增加1
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
// 扩容的具体实现逻辑
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}
复制代码满足数组的容量的条件时链表转化为红黑树
// 如果数组小于64扩容数组，否则链表转化为红黑树
private final void treeifyBin(Node<K,V>[] tab, int index) {
Node<K,V> b; int n, sc;
if (tab != null) {
// 数组容量是否小于64，满足条件进行数组扩容
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
// 尝试进行数组扩容
tryPresize(n << 1);
// 判断当前数组位的头节点非null并且hash >=0 代表当前节点是链表节点
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));
}
}
}
}
}
复制代码增加键值对数
首先介绍一下作者思路，我觉得这里作者设计的非常精妙，如果只有一个size变量，当每次put一个新键值对时都通过CAS修改size值这样虽然能保证并发安全，但是如果并发量特别大这里容易出现很多线程循环CAS的情况，所以作者为了优化采用了分片的设计思想，作者通过创建数组，之后可以对数组不同位置进行CAS操作，当调用size()方法时只需要把数组中所有的位置的数相加就能得到数据
// 增加键值对数
private final void addCount(long x, int check) {
CounterCell[] as; long b, s;
// 1.判断counterCells计数表是否null，如果null则直接对baseCount进行CAS累加
// 2.如果失败则证明存在竞争，这时就不能通过baseCount进行计数，只能通过CounterCell计数表进行记录
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
CounterCell a; long v; int m;
boolean uncontended = true;
// 1.CounterCell计数表为null直接调用fullAddCount
// 2.如果从计数表中随机取出一个位置的元素为null，直接调用fullAddCount
// 3.通过CAS修改随机位置的元素，如果修改失败则代表出现了冲突，则调用fullAddCount
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);
return;
}
// 链表长度小于等于1，无需考虑扩容
if (check <= 1)
return;
// 统计键值对个数
s = sumCount();
}
// 链表长度大于0，检查扩容
if (check >= 0) {
Node<K,V>[] tab, nt; int n, sc;
// 集合大小大于等于扩容阀值并且数组长度小于最大容量
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
// 生成唯一的扩容戳
int rs = resizeStamp(n);
// sc < 0代表已经存在别的线程正在进行扩容
if (sc < 0) {
// 1. 比较当前生成的扩容戳的版本号和正在扩容线程所生成的扩容版本号是否一致
// 2. sc == rs + 1表示扩容已经结束
// 3. sc == rs + MAX_RESIZERS表示帮助扩容的线程已到最大值
// 4. (nt = nextTable) == null表示用于扩容的数组为null，说明扩容已结束
// 5. transferIndex <= 0代表所有的扩容任务均被领完了
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
// 说明当前可以参加扩容任务，通过CAS把扩容戳+1，代表参与扩容
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
// 进行扩容
transfer(tab, nt);
}
// 当前不存在其他线程正在扩容，通过CAS新建扩容任务
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
// 开始进行扩容
transfer(tab, null);
// 统计键值对个数
s = sumCount();
}
}
}
复制代码先讲解fullAddCount的作用
进行初始化或者扩容计数表
// 初始化或者扩容计数表
private final void fullAddCount(long x, boolean wasUncontended) {
int h;
// 如果当前线程生成的随机数是0，则重新生成随机数
if ((h = ThreadLocalRandom.getProbe()) == 0) {
// 强制初始化
ThreadLocalRandom.localInit(); // force initialization
// 获取随机数
h = ThreadLocalRandom.getProbe();
// 由于重新生成了随机数，未冲突位设置成true
wasUncontended = true;
}
boolean collide = false; // True if last slot nonempty
// 自旋
for (;;) {
CounterCell[] as; CounterCell a; int n; long v;
// as != null代表已经进行了初始化
if ((as = counterCells) != null && (n = as.length) > 0) {
// 通过该值与当前线程生成的随机数做&，获取节点下标
if ((a = as[(n - 1) & h]) == null) {
// 代表计数表不在进行初始化和扩容
if (cellsBusy == 0) { // Try to attach new Cell
// 构造一个CounterCell，传入元素个数
CounterCell r = new CounterCell(x); // Optimistic create
// 修改cellsBusy，防止其他线程对计数表进行并发操作
if (cellsBusy == 0 &&
U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
boolean created = false;
try { // Recheck under lock
CounterCell[] rs; int m, j;
// 将初始化的r对象放入到指定位置中
if ((rs = counterCells) != null &&
(m = rs.length) > 0 &&
rs[j = (m - 1) & h] == null) {
rs[j] = r;
created = true;
}
} finally {
// 恢复标志位
cellsBusy = 0;
}
// 创建成功，退出循环
if (created)
break;
// 说明cells下标元素不是null，进入下一次循环
continue; // Slot is now non-empty
}
}
collide = false;
}
// 说明在addCount方法中CAS失败了，并且获取probe的值不为null
else if (!wasUncontended) // CAS already known to fail
// 设置为未冲突，进入下一次自旋
wasUncontended = true; // Continue after rehash
// 因为指定位置的值不为null，所以通过CAS进行累加，如果成功直接退出
else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
break;
// 1. 如果已经有其他线程新建了counterCells
// 2. counterCells的容量大于cpu核心数，这里其实很巧妙，因为线程并发数不会超过cpu核心数
else if (counterCells != as || n >= NCPU)
// 设置循环失败，不继续进行扩容
collide = false; // At max size or stale
// 恢复状态，表示下一次循环进行扩容
else if (!collide)
collide = true;
// 如果线程能执行到这，说明当前线程竞争严重，所以进行counterCells的扩容
else if (cellsBusy == 0 &&
U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
try {
if (counterCells == as) {// Expand table unless stale
// 长度左移1位==容量*2
CounterCell[] rs = new CounterCell[n << 1];
for (int i = 0; i < n; ++i)
rs[i] = as[i];
counterCells = rs;
}
} finally {
// 恢复标志位
cellsBusy = 0;
}
collide = false;
// 继续进行下一次循环
continue; // Retry with expanded table
}
h = ThreadLocalRandom.advanceProbe(h);
}
// 能走到这说明counterCells是null的
// 1.cellsBusy == 0表示没有线程操作cellsBusy
// 2.通过CAS修改标识位，准备进行counterCells的初始化
else if (cellsBusy == 0 && counterCells == as &&
U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
boolean init = false;
try { // Initialize table
if (counterCells == as) {
// 初始化容量是2
CounterCell[] rs = new CounterCell[2];
rs[h & 1] = new CounterCell(x);
// 赋值
counterCells = rs;
// 设置初始化完成标记
init = true;
}
} finally {
//恢复标志位
cellsBusy = 0;
}
// 初始化成功，退出循环
if (init)
break;
}
// 如果还是很激烈，就采用最后的保底方法，直接累加在baseCount上
else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))
break; // Fall back on using base
}
}
复制代码大家是否已经晕了呢，接下来带大家分析具体的扩容过程，大家小心晕车！
扩容是ConcurrentHash的精华之一，扩容的核心就在于数据迁移，在单线程的场景下，转移很简单就是把旧数组中的数据迁移到新数组中，但是在多线程的场景下，可能在扩容的时候也会存在线程在添加元素，可以通过加互斥锁的方式防止并发问题，但是这样它的吞吐量就会极度下降，所以ConcurrentHashMap并没有采用加锁的方式，而是采用了无锁的CAS策略，其实最精华的部分就是它可以利用多线程来进行扩容，每个线程迁移一部分的数据，这样就实现了多线程扩容
// 前面都是帮助扩容的逻辑，这里才是真正扩容的具体实现方法
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
// 这里进行桶的划分，将（n >>> 3相当于n/8）然后除以CPU核心数，如果得到小于16则就是用16
// 这样做的目的就是让每个CPU处理的桶一样多，防止出现分配不均的情况，如果桶比较少，就默认一个CPU处理扩容任务
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
// 代表用来扩容的数组为null，初始化数组
if (nextTab == null) { // initiating
try {
@SuppressWarnings("unchecked")
// 新建一个2倍当前容量的数组
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
// 扩容失败，sizeCtl使用int最大值
sizeCtl = Integer.MAX_VALUE;
return;
}
// 更新成员标量
nextTable = nextTab;
transferIndex = n;
}
// 新数组的容量
int nextn = nextTab.length;
// 创建一个ForwardingNode节点，这个表示这个节点是处于正在迁移的过程，并且它的hash值是-1.这个是不是似曾相识，对的就是上面put方法中判断的是否正在扩容的条件
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;
if (--i >= bound || finishing)
advance = false;
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
bound = nextBound;
i = nextIndex - 1;
advance = false;
}
}
// i < 0 说明当前线程已经遍历完了旧数组
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
// 如果完成了扩容
if (finishing) {
// 删除扩容时出现的成员变量
nextTable = null;
table = nextTab;
// 更新下次扩容阀值
sizeCtl = (n << 1) - (n >>> 1);
return;
}
// 还记得每次帮助扩容都会把扩容戳+1吗，这里就是完成扩容后通过CAS把扩容戳-1
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
// 只有第一个扩容线程扩容时是把扩容戳+2的，其他帮助扩容的都是+1，大家如果有问题的可以去上边看看，就理解了
// 之后如果最后把扩容戳-2之后和另一值相等了，说明没有线程在帮助扩容了，也就是扩容结束了
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
// 如果相等，扩容结束
finishing = advance = true;
i = n; // recheck before commit
}
}
// 如果该位置没有节点则放入fwd节点
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) {
Node<K,V> ln, hn;
// hash值是正数，上面出现过，这里代表当前当前节点是链表节点
if (fh >= 0) {
// 下面的逻辑和hashMap相似，都是构建高低链，因为数组长度进行了扩容，可能出现某些key重新hash后在当前位置的情况
int runBit = fh & n;
Node<K,V> lastRun = f;
for (Node<K,V> 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;
}
// 循环链表，构建高低链
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);
}
// 通过CAS方式，把高低链放到目标数组中
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
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;
}
}
// 分别放到目标数组的不同位置
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);
advance = true;
}
}
}
}
}
}
复制代码总结
到这里ConcurrentHashMap中精华部分都已经结束，能看到这的我觉得应该给自己点掌声，因为并发安全的ConcurrentHashMap确实比HashMap底层要复杂的多，有很多为了解决并发问题而进行的处理。
我们可以看到作者在其中很多巧妙的设计，比如在计算键值对个数而采用的计数表，这里采用了分片的思想来解决并发问题、通过多线程扩容的方式降低扩容所需的时间，这些都是我们应该进行思考的，如果我们是ConcurrentHashMap的设计者，我们是否能想到用这种方式降低线程竞争，提高吞吐量，我们能否把这些思路利用到我们的日常开发中，这些都是我们需要考虑的，看完源码之后的思考我觉得才是提高一个人能力的关键。
其实我知道看源码会很枯燥，很乏味，但是我觉得还是应该坚持下去，这样能做到知其然，知其所以然，这些看源码的能力会潜移默化的提升你的能力，当你做开发时会不由自主的想到作者的一些思路，从而提升你的代码能力，我希望每个开发者都能坚持下来，一起加油！
最后我想用一句话结束这篇文章，努力加油吧，多年以后，你一定会感谢曾经那么努力的自己！
我是爱写代码的何同学，我们下期再见！
作者：爱写代码的何同学
链接：https://juejin.cn/post/6945762887566639135
来源：稀土掘金
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。
京公网安备 11010502036488号