jdk 8 ConcurrentHashMap put 源码分析

65 阅读 0 评论 43 点赞
我是靠谱客的博主结实云朵，最近开发中收集的这篇文章主要介绍jdk 8 ConcurrentHashMap put 源码分析，觉得挺不错的，现在分享给大家，希望可以做个参考。
概述

 //------------------------------常用常量------------------
// 最大容量
private static final int MAXIMUM_CAPACITY = 1 << 30;
// 默认初始容量
private static final int DEFAULT_CAPACITY = 16;
// 数组的最大容量,防止抛出OOM
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
// 最大并行度，仅用于兼容JDK1.7以前版本
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;
// 链表转红黑树要求 数组的最小长度为 64
static final int MIN_TREEIFY_CAPACITY = 64;
// 扩容搬运时，批量搬运的最小槽位数（步长）
private static final int MIN_TRANSFER_STRIDE = 16;
// 当前待扩容table的邮戳位,通常是高16位
private static int RESIZE_STAMP_BITS = 16;
// 同时搬运的线程数自增的最大值
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
// 搬运线程数的标识位，通常是低16位
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
// 当前哈希表
transient volatile Node<K, V>[] table;
// 扩容的下一个哈希表
private transient volatile Node<K, V>[] nextTable;
// 计数的基准值
private transient volatile long baseCount;
//控制变量
private transient volatile int sizeCtl;
// 并发搬运过程中CAS获取区段的下限值
private transient volatile int transferIndex;
// 计数cell初始化或者扩容时基于此字段使用自旋锁
private transient volatile int cellsBusy;
// 加速多核CPU计数的cell数组
private transient volatile CounterCell[] counterCells;
//------------------------------public 方法------------------
public V put(K key, V value) {
return putVal(key, value, false);
}
final V putVal(K key, V value, boolean onlyIfAbsent) {
// key 或 value 有一个为空，就抛出异常
if (key == null || value == null) throw new NullPointerException();
// 获取 key 的 hash(高16位 和 低16位做异或运算后，和 0x7fffffff 做与运算，保证高低位同时参与运算，且最高位为正数)
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K, V>[] tab = table; ; ) {
Node<K, V> f;
int n, i, fh;
if (tab == null || (n = tab.length) == 0)
// 如果 table 还没有初始化，先初始化
tab = initTable();
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
// 在 table hash 和 数组长度 -1 做与运算的位置是 Null，就在这个位置通过 CAS 添加 新 node
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)
// 如果 table 的该槽位是在移动到新表的过程中，加入扩容工作
tab = helpTransfer(tab, f);
else {
// 添加新的数据
V oldVal = null;
// 对头节点加锁
synchronized (f) {
// 再次校验槽位的值没有变
if (tabAt(tab, i) == f) {
// 如果 原来的 node 的 hash 大于 0，说明是 链表，红黑树的 treeBin 是 -2
if (fh >= 0) {
binCount = 1;
for (Node<K, V> e = f; ; ++binCount) {
K ek;
// 如果 链表当前的 node 的 hash 和 key 都和要添加的值相同，那就是同一个 key
if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) {
// 获取旧的 value
oldVal = e.val;
// 根据 onlyIfAbsent 决定是否需要覆盖旧的 value
if (!onlyIfAbsent)
e.val = value;
break;
}
// 如果当前的 node 和要添加的值不相同，继续往下一个 Node 查找，下一个为 null 就直接添加，否则循环处理
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) {
// 这里 fh = -2，是红黑树结构。
Node<K, V> p;
binCount = 2;
// 通过 CAS 将要添加的值添加到树，并且会真正的将树平衡化
if ((p = ((TreeBin<K, V>) f).putTreeVal(hash, key, value)) != null) {
// 获取旧的 value
oldVal = p.val;
// 根据 onlyIfAbsent 决定是否需要覆盖旧的 value
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
// 如果 binCount 大于 树化的阈值
if (binCount >= TREEIFY_THRESHOLD)
// 进行树化
treeifyBin(tab, i);
if (oldVal != null)
// 如果是添加的值在以前的 map 中，说明没有新增加值，直接返回旧的 value
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}
private static final int tableSizeFor(int c) {
int n = c - 1;
n |= n >>> 1;
n |= n >>> 2;
n |= n >>> 4;
n |= n >>> 8;
n |= n >>> 16;
return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
}
private final Node<K, V>[] initTable() {
Node<K, V>[] tab;
int sc;
// 这里 通过 while 和 cas 保证只有一个线程初始化 table
while ((tab = table) == null || tab.length == 0) {
// 如果 sizeCtl 小于 0
// 初始化就为 0，这里第一个线程就直接通过 cas 设置
SIZECTL 为 -1，注意 SIZECTL = sizeCtl
if ((sc = sizeCtl) < 0)
Thread.yield(); // lost initialization race; just spin
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
// 如果成功将 SIZECTL 设置为 -1
try {
if ((tab = table) == null || tab.length == 0) {
// 默认 数组容量 为 16
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
// 初始化 node 数组
Node<K, V>[] nt = (Node<K, V>[]) new Node<?, ?>[n];
// 将 node 数组 赋值给 全局变量 table
table = tab = nt;
// sc = 16 - 16 >>> 2 = 16 - 16/4 = 12
sc = n - (n >>> 2);
}
} finally {
// 数组容量扩容的阈值在第一次初始化就是 12 了
sizeCtl = sc;
}
break;
}
}
return tab;
}
final Node<K, V>[] helpTransfer(Node<K, V>[] tab, Node<K, V> f) {
Node<K, V>[] nextTab;
int sc;
// 原表 tab 不为 null ，原表的 槽位是在数据搬运中。并且搬运的新表也已经创建好了
if (tab != null && (f instanceof ForwardingNode) && (nextTab = ((ForwardingNode<K, V>) f).nextTable) != null) {
int rs = resizeStamp(tab.length);
while (nextTab == nextTable && table == tab && (sc = sizeCtl) < 0) {
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || transferIndex <= 0)
break;
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
// 通过 cas 将 SIZECTL +1
// 数据转移到新表 nextTab
transfer(tab, nextTab);
break;
}
}
// 返回新的 table
return nextTab;
}
// 返回旧的 table
return table;
}
static final int resizeStamp(int n) {
// 获取 n 从高位开始 0 到 1 之间的个数 a，| 1 向左移 15 位得到 b，这里就是 将 a+b
return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));
}
/**
* 将每个节点移动到新表
*/
private final void transfer(Node<K, V>[] tab, Node<K, V>[] nextTab) {
int n = tab.length, stride;
// 如果是多核 cpu，table 的长度 / 8 / cpu 数量，如果是单核 cpu，步长就是 table 的长度
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
// 如果 步长 小于 默认值 16，就使用 16 作为步长
stride = MIN_TRANSFER_STRIDE; // subdivide range
if (nextTab == null) {
// initiating
// 如果新表是 null
try {
@SuppressWarnings("unchecked")
// 先创建 新的 table，容量是当前 table 的 2 倍数
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;
}
// 将新的 table 赋值给全局的新 table
nextTable = nextTab;
// 设置 并发搬运过程中CAS获取区段的下限值 为 原来 table 长度
transferIndex = n;
}
// 新 table 的长度
int nextn = nextTab.length;
// 新建 搬运标识，构造方法设置 hash 为 MOVED
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))) {
// 从原来 table 的右边开始搬运，nextBound 就是区间的左侧
bound = nextBound;
i = nextIndex - 1; // 区间的右侧 - 1，搬运完了该槽位，advance又会置为 true
advance = false;
}
}
if (i < 0 || i >= n || i + n >= nextn) {
// 搬运完成了
int sc;
if (finishing) {
// 完成搬运标识为 true，就将新 table 赋值给 table，新 table 设为 null
nextTable = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1); // 扩容的阈值由原来的 table 长度 *2 - 原来的 table 长度 /2 ，如原来的 table 长度为 16，16 * 2 - 16 / 2 = 24
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) {
Node<K, V> ln, hn;
if (fh >= 0) {
// 链表的搬运
// 注意这里是 n ，table 的长度，只有 1 位是 1，其他位都是 0，所以 runBit 只能是 0 或者 1
int runBit = fh & n;
Node<K, V> lastRun = f;
// 找到最后一个 和 链表头 在 table 长度位不一致的 节点
for (Node<K, V> p = f.next; p != null; p = p.next) {
int b = p.hash & n;
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
// 如果 runBit 为 0 ，就放到低位，否则放在高位
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)
// 注意第4个值是 next。
ln = new Node<K, V>(ph, pk, pv, ln);
else
hn = new Node<K, V>(ph, pk, pv, hn);
}
// 添加低位的链表
setTabAt(nextTab, i, ln);
// 添加高位的链表
setTabAt(nextTab, i + n, hn);
// 设置 第 i 个槽位为搬运 节点
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;
}
}
}
}
}
}
/**
* 注意方法是 TreeBin 中的方法
*/
final TreeNode<K, V> putTreeVal(int h, K k, V v) {
Class<?> kc = null;
boolean searched = false;
// 从 根节点开始循环查找
for (TreeNode<K, V> p = root; ; ) {
int dir, ph;
K pk;
if (p == null) {
// 根节点为 null ，新建一个根节点，跳出循环
first = root = new TreeNode<K, V>(h, k, v, null, null);
break;
} else if ((ph = p.hash) > h) // 如果 当前节点（下面可能替换为子节点）的 hash 大于 要添加的 hash
dir = -1;
else if (ph < h) // 如果 当前节点的 hash 小于 要添加的 hash
dir = 1;
else if ((pk = p.key) == k || (pk != null && k.equals(pk)))
// 如果 当前节点的 hash 等于 要添加的 hash ，并且 key 也是一样，就是同一个 node
return p;
else if ((kc == null && (kc = comparableClassFor(k)) == null) || (dir = compareComparables(kc, k, pk)) == 0) {
// 这里 当前节点的 hash 等于 要添加的 hash，但是 key 不一致。先通过 当前节点的左右节点找到 给定 k 的节点，没有找到 就对比 身份 hash
// 如果 要添加的 key 没有实现 Comparable，或者 获取到的 泛型 （Comparable<?>）和 当前节点 的key 的类型一致
if (!searched) {
TreeNode<K, V> q, ch;
searched = true;
// 取出 当前节点 的左或者右 节点当作根节点，尝试获取到 给定 k 的节点
if (((ch = p.left) != null && (q = ch.findTreeNode(h, k, kc)) != null) ||
((ch = p.right) != null && (q = ch.findTreeNode(h, k, kc)) != null))
// 找到就结束
return q;
}
dir = tieBreakOrder(k, pk);
}
TreeNode<K, V> xp = p;
// 如果 dir 为 -1 ，取左子节点，否则子右节点
if ((p = (dir <= 0) ? p.left : p.right) == null) {
// 如果取出的子节点为 null
TreeNode<K, V> x, f = first;
// 新建一个 treeNode , 当前节点（p）的父节点（xp），原来的 first 节点作为 next，重置 first 节点为新建节点
first = x = new TreeNode<K, V>(h, k, v, f, xp);
if (f != null)
// 如果 原来的 first 节点不为 null，将 原来的 first 节点的前节点为 新建节点
f.prev = x;
if (dir <= 0)
// 如果 dir = -1，将 新建节点 放置在父节点（xp）左边
xp.left = x;
else
// 如果 dir = 1，将 新建节点 放置在父节点（xp）右边
xp.right = x;
if (!xp.red)
// 如果父节点（xp）是黑色，现在变色为红色
x.red = true;
else {
lockRoot();
// 加锁
try {
// 将树进行平衡
root = balanceInsertion(root, x);
} finally {
// 解锁
unlockRoot();
}
}
break;
}
}
assert checkInvariants(root);
return null;
}
private final void treeifyBin(Node<K, V>[] tab, int index) {
Node<K, V> b;
int n, sc;
if (tab != null) {
// 如果 table 的长度小于 64，那就还是先扩容 table，不进行树化
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
tryPresize(n << 1);
else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
// 取出 table 在 index 槽位的 node，并且该 node 是链表的 node，treeBin 是 -2
synchronized (b) { // 给 槽位的 第一个 对象 加锁，
// 再次校验 该 node 没有变化
if (tabAt(tab, index) == b) {
TreeNode<K, V> hd = null, tl = null;
// 根据 槽位的 第一个 node，遍历该 链表，全部转换为 TreeNode
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) // 会设置每一个 node 的 prev
// 设置 树 的 头 node
hd = p;
else
// 1，2，3，4，5，6
tl.next = p;// 会设置每一个 node 的 next
tl = p;
}
// 转换完成后 ，头 node 使用 treeBin 包装。
// 注意：这里 TreeBin 的构造方法会去维护树结构
setTabAt(tab, index, new TreeBin<K, V>(hd));
}
}
}
}
}
// 尝试将 table 的大小扩容到给定的 size
private final void tryPresize(int size) {
// 如果 size 大于最大值就取最大值，否则取大于 size 的最近的 2 的次幂数
int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY : tableSizeFor(size + (size >>> 1) + 1);
int sc;
while ((sc = sizeCtl) >= 0) {
Node<K, V>[] tab = table;
int n;
// 如果 table 还是 Null，还是先初始化
if (tab == null || (n = tab.length) == 0) {
// 确定 table 的长度
n = (sc > c) ? sc : c;
if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
// 通过 CAS 保证只有一个线程初始化 table
try {
if (table == tab) {
// 再次检查 table 没有变化
@SuppressWarnings("unchecked")
Node<K, V>[] nt = (Node<K, V>[]) new Node<?, ?>[n];
// 将 初始化 的 数组 赋值给 table
table = nt;
// sc = 16 - 16 >>> 2 = 16 - 16/4 = 12
sc = n - (n >>> 2);
}
} finally {
// 数组容量扩容的阈值在第一次初始化就是 12 了
sizeCtl = sc;
}
}
} else if (c <= sc || n >= MAXIMUM_CAPACITY)
// 如果都已经最大了，就不扩容
break;
else if (tab == table) {
// 开始扩容
int rs = resizeStamp(n);
// 在上面初始化 table 时，table 刚初始化好，其它线程进入，sc 还是 -1，此时会进入
if (sc < 0) {
Node<K, V>[] 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))
// 通过 cas 将 SIZECTL +1
// 数据转移到新表 nt
transfer(tab, nt);
} else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2))
// 通过 cas 将 SIZECTL 设置为： rs 左移 16 位（扩容标记）+低位 + 2（表示扩容线程）
// 数据转移到新表
transfer(tab, null);
}
}
}
private final void addCount(long x, int check) {
CounterCell[] as;
long b, s;
// 借助 counterCells 数组来统计数量
// 如果 counterCells 数组 不为 null（有 counterCells 数组 就借助这个数组去增加数量），或者 没有成功把 map 中 的 BASECOUNT 值 + x
if ((as = counterCells) != null || !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
CounterCell a;
long v;
int m;
boolean uncontended = true;
// 如果 counterCells 数组为空 或者 随机数 & 数组长度 的槽位 为空 或者没有成功向数组的 CELLVALUE 位的值 + x
if (as == null || (m = as.length - 1) < 0 || (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
!(uncontended = U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
// 添加 x
fullAddCount(x, uncontended);
return;
}
if (check <= 1)
return;
// 统计 map 中的 node 个数： baseCount + counterCells 数组中的 数量
s = sumCount();
}
if (check >= 0) {
Node<K, V>[] tab, nt;
int n, sc;
// 如果 map 中的个数超过了 扩容的阈值，就要进行扩容
while (s >= (long) (sc = sizeCtl) && (tab = table) != null && (n = tab.length) < MAXIMUM_CAPACITY) {
int rs = resizeStamp(n);
if (sc < 0) {
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))
// 通过 cas 将 SIZECTL +1
// 数据转移到新表 nt
transfer(tab, nt);
} else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2))
// 通过 cas 将 SIZECTL 设置为： rs 左移 16 位（扩容标记）+低位 + 2（表示扩容线程）
// 数据转移到新表
transfer(tab, null);
// 统计 map 中的 node 个数，回到 while（可能这个时候又添加了大量的 node）
s = sumCount();
}
}
}