介绍:
这个类是个抽象类，里面主要实现了两个方法longAccumulate，doubleAccumulate 分别long 和double的数据进行一个累计。主要提供给LongAdder ，LongAccumulator,DoubleAdder,DoubleAccumulator 进行一个调用。可以理解为用cells的方式来减少并发时产生的冲突。
详解：
final void longAccumulate(long x, LongBinaryOperator fn,
boolean wasUncontended) {
int h;
//获取probe 值，如果为0，进行初始化处理
if ((h = getProbe()) == 0) {
ThreadLocalRandom.current(); // force initialization
h = getProbe();
//标志未竞争为true
wasUncontended = true;
}
//设置没有cell 碰撞
boolean collide = false; // True if last slot nonempty
for (;;) {
Striped64.Cell[] as; Striped64.Cell a; int n; long v;
//cells中有值
if ((as = cells) != null && (n = as.length) > 0) {
//对应这个hash 值对应的cell中没有值
if ((a = as[(n - 1) & h]) == null) {
//创建cell的时候的锁
if (cellsBusy == 0) { // Try to attach new Cell
Striped64.Cell r = new Striped64.Cell(x); // Optimistically create
//加锁
if (cellsBusy == 0 && casCellsBusy()) {
boolean created = false;
try { // Recheck under lock
Striped64.Cell[] rs; int m, j;
if ((rs = cells) != null &&
(m = rs.length) > 0 &&
rs[j = (m - 1) & h] == null) {
rs[j] = r;
created = true;
}
} finally {
cellsBusy = 0;
}
if (created)
break;
//走到continue 表示这cell 已经创建了，重试吧，这种情况一般不会出现
continue; // Slot is now non-empty
}
}
collide = false;
}
//发生了cas竞争的失败
else if (!wasUncontended) // CAS already known to fail
//表示此cell肯定有值了，就是a，设置为true ，最后会重新计算 probe值，这个值的false，只会在当前线程probe 已经存在的情况下出现
wasUncontended = true; // Continue after rehash
// 知晓cell 就是a 了，对a 这个cell 里面的值进行一个操作
else if (a.cas(v = a.value, ((fn == null) ? v + x :
fn.applyAsLong(v, x))))
break;
//如果table长度已经是最大值，或as 已经不是原来的了,即竞争超时
else if (n >= NCPU || cells != as)
//设置碰撞为false，之后会重新计算probe值
collide = false; // At max size or stale
else if (!collide)
collide = true; //重新计算probe值时候还是碰撞，并且对cell更新还是一个失败
else if (cellsBusy == 0 && casCellsBusy()) { //竞争太多了，开始扩容了
try {
if (cells == as) { // Expand table unless stale
Striped64.Cell[] rs = new Striped64.Cell[n << 1];
for (int i = 0; i < n; ++i)
rs[i] = as[i];
cells = rs;
}
} finally {
cellsBusy = 0;
}
collide = false;
//扩容后的重试
continue; // Retry with expanded table
}
h = advanceProbe(h);//重新计算probe 值
}//表示为第一次遇到竞争，此时基本上是刚刚开始创建cells数组
else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
boolean init = false;
try { //初始化cells并写入第一个值 // Initialize table
if (cells == as) {
Striped64.Cell[] rs = new Striped64.Cell[2];
//初始化cell，并将值写入cell中
rs[h & 1] = new Striped64.Cell(x);
cells = rs;
init = true;
}
} finally {
cellsBusy = 0;
}
if (init)
break;
}// cellsBusy锁竞争失败 试试能不能修改base的值
else if (casBase(v = base, ((fn == null) ? v + x :
fn.applyAsLong(v, x))))
break; // Fall back on using base
}
}
cells的长度为2的幂次方
配合上LongAdder 类中的add方法来看,这个是实现是尽可能的减少并发时产生的冲突,实在不行了才进行下一个冲突。
- 更新base值，不行
2.如果没有初始化cells，初始化cells更新
3.如果对于hash的cell上没有值，写入，写入不成功，重新hash
4.如果之前在cell上有竞争的，重新hash
5.在对应cell值上对cas 进行更新。
6.如果cells的长度还没有到最大cpu数，设置碰撞为false，重新hash - 如果没有碰撞，设置为有碰撞，重新hash
8.如果上面都没有命中，那就开始扩容一倍，从第三步开始
public void add(long x) {
Cell[] as; long b, v; int m; Cell a;
// 如果cell为空，尝试对base值修改
if ((as = cells) != null || !casBase(b = base, b + x)) {
boolean uncontended = true;
// 如果cell上对应的值不为空，先试试更新cell中的值
if (as == null || (m = as.length - 1) < 0 ||
(a = as[getProbe() & m]) == null ||
!(uncontended = a.cas(v = a.value, v + x))
//不行的话 ，进行上不striped64 中的方式进行
longAccumulate(x, null, uncontended);
}
}
