CAS与LongAdder比效率很低，LongAdder为何快
LongAdder 核心思想如下图
- 无竞争，CAS不断的操作变量base
- 有竞争，其余的操作全部打到具体的CELL[i]中进行各自CAS
- sum操作将base + CELL数组的值加起来即可，保证结果的最终正确性
LongAdder继承了Striped64，Striped64为抽象类
Striped64
/*** @sun.misc.Contended此注解为解决缓存行伪共享问题* Striped64中定义了静态内部类CELL，*/@sun.misc.Contended static final class Cell {// Cell对象中维护的变量valuevolatile long value;Cell(long x) { value = x; }// cell对象内部进行cas操作final boolean cas(long cmp, long val) {return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);}// Unsafe mechanicsprivate static final sun.misc.Unsafe UNSAFE;// 此Cell对象中的值的偏移量，当前Cell对象通过偏移量可以找到变量valueprivate static final long valueOffset;// static代码块，随类的加载仅执行一次，获取内存中的value的值static {try {UNSAFE = sun.misc.Unsafe.getUnsafe();Class<?> ak = Cell.class;valueOffset = UNSAFE.objectFieldOffset(ak.getDeclaredField("value"));} catch (Exception e) {throw new Error(e);}}}
// 获取CPU数量，影响CELL数组长度
static final int NCPU = Runtime.getRuntime().availableProcessors();
/**
* Cell数组，大小是2的次方，此规定利于hash的平均分布 hash&(size-1)
*/
transient volatile Cell[] cells;
// 无竞争时，对base变量进行操作 或者 当在CELL进行扩容，要先将CELL数据的值写入到base
transient volatile long base;
// 锁，在CAS下，创建cell 或者 调整数组大小时需要，0 无锁 1 有锁
transient volatile int cellsBusy;
// 通过CAS方式修改base的值
final boolean casBase(long cmp, long val) {
return UNSAFE.compareAndSwapLong(this, BASE, cmp, val);
}
// 获取锁
final boolean casCellsBusy() {
return UNSAFE.compareAndSwapInt(this, CELLSBUSY, 0, 1);
}
// 此方法简单理解为为从ThreadLocalRandom方法赋给当前线程一个HASH值
static final int getProbe() {
return UNSAFE.getInt(Thread.currentThread(), PROBE);
}
// 将当前线程的hash值进行重新计算赋值
static final int advanceProbe(int probe) {
probe ^= probe << 13; // xorshift
probe ^= probe >>> 17;
probe ^= probe << 5;
UNSAFE.putInt(Thread.currentThread(), PROBE, probe);
return probe;
}
LongAdder#add()方法
public void add(long x) {
// as cells对象的引用
// b 变量base的引用
// v cell中value的期望值的引用
// m cells长度
// a cell[]中某个具体的cell[i]的cell对象的引用
Cell[] as; long b, v; int m; Cell a;
/**
* true可进入if代码块的情况有2种：
* 1. cells已经初始化，表明已经发生了竞争
* 2. 对于base进行累加操作，但失败,取反为ture，表明竞争出现
*/
if ((as = cells) != null || !casBase(b = base, b + x)) {
// 当前线程操作CELL数组中的某个对象是否出现竞争 true 无，false 有
boolean uncontended = true;
// 1. cell数组为空，还未初始化，将进行初始化
// 2. 从长度判断cells未初始化
// 3. cells已经初始化了了，当前线程落在cells数组中的位置为空，表明此处未发生竞争，可以去新建个cell对象进行计算
// 4. 当前线程落在cells数组中，进行cas计算，计算失败表明当前位置存在竞争，进入if重新计算位置进行数值计算
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);
}
}
Striped64#longAccumulate()方法
final void longAccumulate(long x, LongBinaryOperator fn,
boolean wasUncontended) {
// h 用来保存当前线程的hash值
int h;
if ((h = getProbe()) == 0) {
// 线程hash值为0时，重新计算hash值，并赋值给h
ThreadLocalRandom.current(); // force initialization
h = getProbe();
// 不当作竞争态
wasUncontended = true;
}
// 是否扩容
boolean collide = false; // True if last slot nonempty
// 自旋
for (;;) {
/**
* as cells数组引用
* a cells中数组的下标对应的cell对象引用
* n celss数组长度
* v cell对象的value值的引用
*/
Cell[] as; Cell a; int n; long v;
// cells数组已经初始化
if ((as = cells) != null && (n = as.length) > 0) {
// 线程所在的cells的下标的cell对象为空，表示该下标处为填充cell对象，准备新建cell对象
if ((a = as[(n - 1) & h]) == null) {
// 锁未被抢占,方可新建cell对象
if (cellsBusy == 0) { // Try to attach new Cell
// 先把cell对象创建出来，然后先按兵不动，因为创建是在抢占锁成功后的动作
Cell r = new Cell(x); // Optimistically create
// 再次判断锁未被抢占，且该线程抢占成功才进行真正的创建cell对象
if (cellsBusy == 0 && casCellsBusy()) {
boolean created = false;
try { // Recheck under lock
Cell[] rs; int m, j;
// 再次判断！cells是初始化过的，数组长度是大于0的，
// cell对象要放入的数组槽位真的是空 才真正创建
if ((rs = cells) != null &&
(m = rs.length) > 0 &&
rs[j = (m - 1) & h] == null) {
// 创建对象！
rs[j] = r;
created = true;
}
} finally {
// 释放锁
cellsBusy = 0;
}
// 创建成功后 会改标识字段 created 为true，退出循环
if (created)
break;
//并发下，发现该槽位已经被占用了，继续自旋
continue; // Slot is now non-empty
}
}
//表明：只要cells数组槽位为空，就是无需扩容
collide = false;
}
/**
* cells已经初始化 且 当前线程的槽位不为空才可走到此处
* wasUncontended 为 false 时，仅是在该情况下：
* 1.cells已经初始化
* 2.当前线程线程命中的槽位不为null
* 3.当前线程存在写入竞争
* 而后 h = advanceProbe(h); 重置该线程的hash
*/
else if (!wasUncontended) // CAS already known to fail
wasUncontended = true; // Continue after rehash
/**
* 走此逻辑表明：
* 当前线程rehash过，定位的槽位处还是存在竞争，重试[一次]进行计算，成功则退出，失败则进入下一个else if
* 需注意：1.当重试一次后，为false 进入 步骤A 为false，继续到 步骤B
* 此时扩容标志取反的True，进入修改扩容标志为true，然后advanceProbe(h)重新rehash线程，
* 2.rehash后，线程可能会继续走到这里(槽位被占用)，
* 再次重试[一次]。如若又失败，此时扩容意向已经为TRUE，进行扩容
*/
else if (a.cas(v = a.value, ((fn == null) ? v + x :
fn.applyAsLong(v, x))))
break;
// cells数组长度大于等于了cpu数（一个cpu一次只能执行一个线程）
// 或者 cells不等同于其as引用，说明已经是扩容过的了
else if (n >= NCPU || cells != as) // -----步骤A
// 扩容标志改为flag
collide = false; // At max size or stale
// 对扩容标志取反，!false = true，修该扩容标志为true，仅作为标记，并不真正扩容数组
else if (!collide) // -----步骤B
collide = true;
// 真正的扩容逻辑 需要条件：无锁状态 且 获取锁成功
else if (cellsBusy == 0 && casCellsBusy()) {
try {
//再次判断 cells此时没有被扩容
if (cells == as) { // Expand table unless stale
// << 左移一位 == *2，扩容至原大小的2被
Cell[] rs = new Cell[n << 1];
for (int i = 0; i < n; ++i)
rs[i] = as[i];
cells = rs;
}
} finally {
// 释放锁
cellsBusy = 0;
}
// 无论该线程本身执行了扩容 或者 被其他线程执行过了，都需要改扩容标志为 false
collide = false;
continue; // Retry with expanded table
}
// rehash 线程值
h = advanceProbe(h);
}
/**
* 初始化cells逻辑
* 锁未被抢占 && cells未初始化 && 抢占锁成功
*/
else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
boolean init = false;
try { // Initialize table
/**
* 再次判断cells未初始化，进而进行初始化，防止并发下重复初始化导致值丢失、覆盖等问题
* 初始化cell数组长度为2，长度必须为2的平方
* h & (2-1 = 1),将当前hash定位到cells对应的下标处，并创建Cell对象
*/
if (cells == as) {
Cell[] rs = new Cell[2];
rs[h & 1] = new Cell(x);
cells = rs;
init = true;
}
} finally {
// 释放锁
cellsBusy = 0;
}
// 初始化逻辑结束
if (init)
break;
}
/**
* fn 为 操作方法函数，针对LongAdder，fn始终为null
* 当代码执行到此，表示cells数组没有初始化，而是处在'正在初始化'的状态，
* 直接自旋将值加到base上，然后退出循环
*/
else if (casBase(v = base, ((fn == null) ? v + x :
fn.applyAsLong(v, x))))
break; // Fall back on using base
}
}
longAdder # sum()
// base + cells数组的值。只保证最终一致性，如需强一致，仍需要AtomicLong
public long sum() {
Cell[] as = cells; Cell a;
long sum = base;
if (as != null) {
for (int i = 0; i < as.length; ++i) {
if ((a = as[i]) != null)
sum += a.value;
}
}
return sum;
}
若有收获，就点个赞吧
0 人点赞
