以下代码为Kane老师整理总结
分析new CurrentHashMap时它在做什么(三个参数的暂时不讨论，大家可以自己去看)
//带参数构造器 ，不带参的啥都没干。
public ConcurrentHashMap(int initialCapacity) {
if (initialCapacity < 0)
throw new IllegalArgumentException();
int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
MAXIMUM_CAPACITY :
tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
//这个公式可以转换为sizeCtl = 【 (1.5 * initialCapacity + 1)，然 后向上取最近的 2 的 n 次方】 这样可以理解
this.sizeCtl = cap; //这个sizeCtl是什么不要着急.
}
分析put代码的过程
public V put(K key, V value) { return putVal(key, value, false); } //put方法里直接去调用了 putVal()
/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException(); //没有反正异常了
int hash = spread(key.hashCode()); //还是算hashCode
int binCount = 0; //局部变量 ,肯定有说法
for (Node<K,V>[] tab = table;;) { //注意 这里是个循环 因为后面是CAS操作，会需要大量的重试
Node<K,V> f; int n, i, fh;
if (tab == null || (n = tab.length) == 0) //如果没数组，创建
tab = initTable();
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { //找到hash值对应的数组下标，这里会得到第一个节点，也就是我们的元素头
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) //出现了，这个标记MOVED,可以去猜，这个东西肯定是扩容时要去做的事情
tab = helpTransfer(tab, f); //帮助数据迁移
else { //这里就是数组已经有元素了，这时候就该挂链表或者挂树了
V oldVal = null;
synchronized (f) { //获取头节点的监视器锁
if (tabAt(tab, i) == f) {
if (fh >= 0) { //头节点的hash值,大于0表示这下面有点东西
binCount = 1; //这个玩意是记录链表的长度的。 ---还是为了转树
for (Node<K,V> e = f;; ++binCount) { //for循环，表示遍历链表
K ek;
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) { //这段代码不解释了,覆盖重复的key
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
Node<K,V> pred = e;
if ((e = e.next) == null) { //到最后了没重复的key，就向后面挂
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
else if (f instanceof TreeBin) { //如果是个树
Node<K,V> p;
binCount = 2;
//插节点
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD) //判断链表的长度，然后转树.
//这里要注意一个地方！！！！！ --不是说像HashMap那样转树就没事了，这里涉及到一个核心思路，CurrentHashMap做了优化，这里如果数组长度小于64，它会先扩容，扩容代表什么含义？-- 原来的链表会被1分为2 分别散落在不同的节点上，这还是个数学公式，大家自己去证明,扩容代码，后面去说.
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}
- put方法的疑问
put的主流程结束了，当然会遗留问题！
初始化方法没看的，这个初始化和HashMap一样吗？
数组小于64会先扩容，从哪里体现的？
扩容的时候它是怎么去做的?
helpTransfer(tab, f) 这个方法为何会叫一个help方法？
初始化方法 initTable
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) { //是个空的，循环吧
if ((sc = sizeCtl) < 0) //注意这个变量sizeCtl 它是小于0的时候 说明了被占用了
Thread.yield(); // lost initialization race; just spin
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { //这里通过CAS操作去设置，告诉你，我拿到了锁了。
try {
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY; //默认初始容量16
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];//初始化数组，长度为16或者初始化提供的长度
table = tab = nt; //赋值给全局变量table，它是可见的
sc = n - (n >>> 2); //这个SC就是之前我们讨论的扩容阈值-->这个阈值还是 0.75*n
}
} finally {
sizeCtl = sc; //又去改了这个变量sizeCtl了，真的坑.
}
break;
}
}
return tab;
}
转红黑树方法 ：treeifyBin
private final void treeifyBin(Node<K,V>[] tab, int index) {
Node<K,V> b; int n, sc;
if (tab != null) {
//MIN_TREEIFY_CAPACITY为64
// 虽然进入到转树方法，如果数组长度小于64，那么先扩容
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
tryPresize(n << 1); //扩容方法后面再说
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)); //把数据放到红黑树中
}
}
}
}
}
扩容方法: tryPresize (注意，核心重点)
如果说ConcurrentHashMap的源码比较巧妙，就在扩容和迁移操作.
private final void tryPresize(int size) {
//c:size的1.5倍，在加1，再向上取最近的2的N次方
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;
//这个if分支和之前初始化数组是一样的，不看了。
if (tab == null || (n = tab.length) == 0) {
n = (sc > c) ? sc : c;
if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if (table == tab) {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = nt;
sc = n - (n >>> 2); //0.75*n
}
} finally {
sizeCtl = sc; //注意这个sizeCtl 它只有在cas操作后会变成-1，告诉我们有个线程在操作它。
}
}
}
else if (c <= sc || n >= MAXIMUM_CAPACITY) //如果数组已经到达最大长度了，就直接结束
break;
else if (tab == table) {
int rs = resizeStamp(n); //这个rs我不能确定干什么，但是影响不大。
if (sc < 0) { //刚开始扩容，我们的sc在上面已经被赋值了，于是这段代码不会执行.
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))
transfer(tab, nt);
}
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2)) //第一次扩容会执行这里的逻辑
transfer(tab, null);
}
}
}
数据迁移：transfer (难点)
//该方法通过全局的transferIndex来控制每个线程的迁移任务
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
//n为旧tab的长度，stride为步长(就是每个线程迁移的节点数)
int n = tab.length, stride;
//单核步长为1，多核为（n>>>3）/ NCPU,最小值为16
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
// 新的 table 尚未初始化
if (nextTab == null) { // initiating
try {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];// 扩容 2 倍
nextTab = nt;// 更新
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE; //扩容失败， sizeCtl 使用 int 最大值。
return;
}
//nextTable为全局属性
nextTable = nextTab;
transferIndex = n;// 更新转移下标，就是 老的 tab 的 length
}
int nextn = nextTab.length;// 新 tab 的 length
// 创建一个 fwd 节点，用于占位。当别的线程发现这个槽位中是 fwd 类型的节点，则跳过这个节点
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
// 首次推进为 true，如果等于 true，说明需要再次推进一个下标（i--），反之，如果是 false，那么就不能推进下标，需要将当前的下标处理完毕才能继续推进
boolean advance = true;
// 完成状态，如果是 true，就结束此方法。
boolean finishing = false; // to ensure sweep before committing nextTab
// 死循环,i 表示下标，bound 表示当前线程可以处理的当前桶区间最小下标
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
// 如果当前线程可以向后推进；这个循环就是控制 i 递减。同时，每个线程都会进入这里取得自己需要转移的桶的区间
while (advance) {
int nextIndex, nextBound;
// 对 i 减一，判断是否大于等于 bound （正常情况下，如果大于 bound 不成立，说明该线程上次领取的任务已经完成了。那么，需要在下面继续领取任务）
// 如果对 i 减一大于等于 bound（还需要继续做任务），或者完成了，修改推进状态为 false，不能推进了。任务成功后修改推进状态为 true。
// 通常，第一次进入循环，i-- 这个判断会无法通过，从而走下面的 nextIndex 赋值操作（获取最新的转移下标）。其余情况都是：如果可以推进，将 i 减一，然后修改成不可推进。如果 i 对应的桶处理成功了，改成可以推进。
if (--i >= bound || finishing)
advance = false;// 这里设置 false，是为了防止在没有成功处理一个桶的情况下却进行了推进 这里的目的是：1. 当一个线程进入时，会选取最新的转移下标。2. 当一个线程处理完自己的区间时，如果还有剩余区间的没有别的线程处理。再次获取区间。
else if ((nextIndex = transferIndex) <= 0) {
// 如果小于等于0，说明没有区间了 ，i 改成 -1，推进状态变成 false，不再推进，表示，扩容结束了，当前线程可以退出了
// 这个 -1 会在下面的 if 块里判断，从而进入完成状态判断
i = -1;
advance = false;// 这里设置 false，是为了防止在没有成功处理一个桶的情况下却进行了推进
}
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
bound = nextBound;// 这个值就是当前线程可以处理的最小当前区间最小下标
i = nextIndex - 1;// 初次对i 赋值，这个就是当前线程可以处理的当前区间的最大下标
advance = false;// 这里设置 false，是为了防止在没有成功处理一个桶的情况下却进行了推进，这样对导致漏掉某个桶。下面的 if (tabAt(tab, i) == f) 判断会出现这样的情况。
}
}
// 如果 i 小于0 （不在 tab 下标内，按照上面的判断，领取最后一段区间的线程扩容结束）
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {// 如果完成了扩容
nextTable = null;// 删除成员变量
table = nextTab;// 更新 table
sizeCtl = (n << 1) - (n >>> 1);// 更新阈值
return;
}
// 尝试将 sc -1. 表示这个线程结束帮助扩容了，将 sc 的低 16 位减一。
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
// 如果 sc - 2 不等于标识符左移 16 位。如果他们相等了，说明没有线程在帮助他们扩容了。也就是说，扩容结束了。
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;// 不相等，说明没结束，当前线程结束方法。
finishing = advance = true;// 如果相等，扩容结束了，更新 finising 变量
i = n; // recheck before commit// 再次循环检查一下整张表
}
}
// 获取老 tab i 下标位置的变量，如果是 null，就使用 fwd 占位。
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);// 如果成功写入 fwd 占位，再次推进一个下标
else if ((fh = f.hash) == MOVED)// 如果不是 null 且 hash 值是 MOVED。
advance = true; // already processed// 说明别的线程已经处理过了，再次推进一个下标
else {// 到这里，说明这个位置有实际值了，且不是占位符。对这个节点上锁。为什么上锁，防止 putVal 的时候向链表插入数据
synchronized (f) {
// 判断 i 下标处的桶节点是否和 f 相同
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;// low, height 高位桶，低位桶
if (fh >= 0) {
// 对老长度进行与运算（第一个操作数的的第n位于第二个操作数的第n位如果都是1，那么结果的第n为也为1，否则0）
// 由于 Map 的长度都是 2 的次方（000001000 这类的数字），那么取于 length 只有 2 种结果，一种是 0，一种是1
// 如果是结果是0 ，Doug Lea 将其放在低位，反之放在高位，目的是将链表重新 hash，放到对应的位置上，让新的取于算法能够击中他。
int runBit = fh & n;
Node<K,V> lastRun = f; // 尾节点，且和头节点的 hash 值取于不相等
// 遍历这个桶 接下来是常规的设置操作，我们先略过
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);
}
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;
}
}
// 如果树的节点数小于等于 6，那么转成链表，反之，创建一个新的树
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;
}
}
}
}
}
}
一个详细的总结可以查看以下文章
Java基础之ConcurrentHashMap
如果觉得有收获就点个赞吧，更多知识，请点击关注查看我的主页信息哦~
网友评论