LongAdder的源码学习与理解
👨💻个人主页： 才疏学浅的木子 🙇♂️ 本人也在学习阶段如若发现问题，请告知非常感谢 🙇♂️ 📒 本文来自专栏： Java基础
LongAdder
// 累加单元组，懒惰初始化
transient volatile Cell[] cells;
// 基础值，如果没有竞争，则用cas累加这个值
transient volatile long base;
//在cells创建或者扩容时，置为1，表示加锁
transient volatile int cellsBusy;cellsBusy作用是当要修改cells数组时加锁，防止多线程同时修改cells数组，0为无锁，1为加锁，加锁的情况有三种 1、cells数组初始化的时候 2、cells数组扩容的时候 3、如果cells数组中某个元素为null，给这个位置创建新的Cell对象的时候
base有两个作用 在开始没有竞争的时候，将累加值到base 在cells初始化的过程中，cells不可用，这时候会尝试将值累加到base上
@sun.misc.Contended注解是为了防止缓存行伪共享
因为CPU与内存之间速度还是存在较大差距所以现在计算机在内存与CPU之间引入了三级缓存
L1与L2是每个CPU独享的，L3是所有CPU共享的
因为CPU与内存的速度差异很大，需要靠预读数据至缓存来提升效率 而缓存以缓存行为单位，每个缓存行对应着一块内存，一般是64byte 缓存的加入会造成数据副本的产生，即同一份数据会缓存在不同核心的缓存行中 CPU要保证数据的一致性，如果某个CPU核心更改了数据，其他CPU核心对应的整个缓存行必须失效
因为cell是数组形式，在内存中是连续存储的，一个Cell为24个字节(16字节的对象头和8字节的value)，因此缓存行可以存下2个Cell对象，所以问题来了： Core-0要修改Cell[0] Core-1要修改Cell[1] 无论谁修改成功，都会导致对方Core的缓存行失效 @sun.misc.Contended注解就是用来解决这个问题，它的原理是在适用此注解的对象或字段的前后各增加128字节大小的padding，从而让CPU将对象预读至缓存时占用不同的缓存行，这样就不会造成对方的缓存行失效
有了AtomicLong为什么还要LongAdder
虽然AtomicLong使用CAS算法，但是CAS失败后还是通过无限循环的自旋锁不断的尝试，在高并发下CAS性能低下的原因所在
public final int getAndAddInt(Object o, long offset, int delta) {
int v;
do {
v = getIntVolatile(o, offset);
} while (!compareAndSwapInt(o, offset, v, v + delta));
return v;
}高并发下N多线程同时去操作一个变量会造成大量CAS失败，然后处于自选状态，导致严重浪费CPU资源，降低了并发性。既然AtomicLong性能问题是由于多线程同时去竞争同一个变量的更新而降低，那么把一个变量分解为多个变量，让同样多的线程去竞争多个资源
LongAdder在内部维护了一个Cells数组，每个Cell里面有一个初始值为0的long型变量，在同等并发量的情况下，争夺单个变量的线程会减少，这是变相减少了争夺共享资源的并发量，另外多个线程在争夺同一个原子变量时候，如果失败不是自选CAS重试而是尝试获取其他原子变量的锁，最后当获取当前值的时候把所有变量的值累加后再加上base的值返回 Cells占用内存相对比较大的所以一开始并不创建，而是在需要时候再创建，也就是惰性加载，当一开始没有空间的时候，所有的更新都是操作base变量
LongAdder中的主要方法
add方法
public void add(long x) {
Cell[] as; long b, v; int m; Cell a;
/**
* casBase 就是使用CAS来进行更改值的
* 只有两种情况才会执行if内的语句
* 1、cells数组不为空的时候(cells数组为空时候不存在竞争，所以直接操作caseBase，当不为空的时候就存在多个线程来竞争)
* 2、cells为空，casBase执行失败的时候(casBase执行成功，则直接返回，如果casBase失败，说明第一次争用冲突产生，需要对cells数组初始化进入if)
*/
if ((as = cells) != null || !casBase(b = base, b + x)) {
boolean uncontended = true;
/**
* as == null ：cells数组没有初始化，成立就进入if执行cell初始化
* (m = as.length - 1) < 0 ：cells数组的长度为0
* 上面两个条件都代表cells数组没有被初始化成功
* (a = as[getProbe() & m]) == null ：说明当前线程获取的cells数组的这个位置的cell没有做过累加，所以需要创建一个cell对象
* !(uncontended = a.cas(v = a.value, v + x)) ：尝试对这个位置的cell进行累加并返回结果，如果累加失败就找另外一个cell
*
* 进入longAccumulate方法有三种情况
* 1、cells没有初始化
* 2、当前线程获取cell[i]的地方为空，需要创建一个cell对象
* 3、当前对cell[i]cas加值失败
*/
if (as == null || (m = as.length - 1) < 0 ||
(a = as[getProbe() & m]) == null ||
!(uncontended = a.cas(v = a.value, v + x)))
longAccumulate(x, null, uncontended);
}
}longAccumulate方法
final void longAccumulate(long x, LongBinaryOperator fn,boolean wasUncontended) {
int h;
/**
*
*/
if ((h = getProbe()) == 0) {
ThreadLocalRandom.current(); // force initialization
h = getProbe();
wasUncontended = true;
}
// cas冲突标志位
boolean collide = false; // True if last slot nonempty
for (;;) {
Cell[] as; Cell a; int n; long v;
/**
* 有三个分支
* 主分支一：cells初始化好(处理add方法中的第3，4个条件)
* 主分支二：cells数组没有初始化或者长度为0的情况(这个分支处理add方法的第1，2个条件)
* 主分支三：cells数组正在被其他线程初始化则尝试将累加值通过cas累加到base上
*/
if ((as = cells) != null && (n = as.length) > 0) {
/**
* 小分支一：如果当前cess[i]的位置为空处理的是第三个条件
*/
if ((a = as[(n - 1) & h]) == null) {
// 代表没有其他线程修改
if (cellsBusy == 0) { // Try to attach new Cell
// 创建一个cell对象
Cell r = new Cell(x); // Optimistically create
// 如果没有其它线程修改通过cas将cellBusy设置为1
if (cellsBusy == 0 && casCellsBusy()) {
//标记create是否创建成功并放入cells数组被hash的位置上
boolean created = false;
try { // Recheck under lock
Cell[] rs; int m, j;
//再次检查cells数组不为空并且长度大于0
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;
//失败说明被其它线程赋值了cells[i]位置
continue; // Slot is now non-empty
}
}
collide = false;
}
/**
* 小分支二：如果add方法中条件4通过cas加cell[i]失败则重新设置为true去找另一个cell
*/
else if (!wasUncontended) // CAS already known to fail
wasUncontended = true; // Continue after rehash
/**
* 小分支三：给cell[i]加值如果成功则直接退出
*/
else if (a.cas(v = a.value, ((fn == null) ? v + x :fn.applyAsLong(v, x))))
break;
/**
* 小分支四：如果cells发生了扩容或者当前cells数组长度大于了CPU的数量就存在冲突
*/
else if (n >= NCPU || cells != as)
collide = false; // At max size or stale
/**
* 小分支五：如果发生了冲突设置为true再次hash
*/
else if (!collide)
collide = true;
/**
* 小分支六： 扩容
*/
else if (cellsBusy == 0 && casCellsBusy()) {
try {
// 检查cells是否已经扩容
if (cells == as) { // Expand table unless stale
Cell[] rs = new 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
}
// 重新计算hash
h = advanceProbe(h);
}
/**
* 主分支二：初始化
*/
else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
boolean init = false;
try { // Initialize table
if (cells == as) {
Cell[] rs = new Cell[2];
rs[h & 1] = new Cell(x);
cells = rs;
init = true;
}
} finally {
cellsBusy = 0;
}
if (init)
break;
}
/**
* 主分支三：如果别人正在初始化就尝试CAS加base
*/
else if (casBase(v = base, ((fn == null) ? v + x :fn.applyAsLong(v, x))))
break; // Fall back on using base
}
}advanceProbe
static final int advanceProbe(int probe) {
probe ^= probe << 13; // xorshift
probe ^= probe >>> 17;
probe ^= probe << 5;
UNSAFE.putInt(Thread.currentThread(), PROBE, probe);
return probe;
}怎么确保Probe是当前线程的 在Thread类中定义了三个变量
@sun.misc.Contended("tlr")
long threadLocalRandomSeed;
/** Probe hash value; nonzero if threadLocalRandomSeed initialized */
@sun.misc.Contended("tlr")
int threadLocalRandomProbe;
/** Secondary seed isolated from public ThreadLocalRandom sequence */
@sun.misc.Contended("tlr")
int threadLocalRandomSecondarySeed;腾讯云开发者
