当前位置：首页 > news >正文

深入理解并发之LongAdder、DoubleAdder的实现原理

news 2025/7/4 1:01:46

深入理解LongAdder、DoubleAdder的实现原理

本文主要通过LongAdder和DoubleAdder的源码，讲述一下其实现原理。通过LongAdder和DoubleAdder的源码可知。两者都是继承了Striped64的类。下面我们将通过源码的形式讲述一下这三个类都做了哪些事情。

1: Striped64

首先，我们看下Striped64这个类，做了哪些功能。

1.1 Cell类(内部类)

@sun.misc.Contended static final class Cell {volatile long value;Cell(long x) { value = x; }final boolean cas(long cmp, long val) {return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);}// Unsafe mechanicsprivate static final sun.misc.Unsafe UNSAFE;private static final long valueOffset;static {try {UNSAFE = sun.misc.Unsafe.getUnsafe();Class<?> ak = Cell.class;valueOffset = UNSAFE.objectFieldOffset(ak.getDeclaredField("value"));} catch (Exception e) {throw new Error(e);}}
}

1: @sun.misc.Contended 注解作用？

在前面的文章中，我们讲到缓存行锁能够代替总线锁的唯一条件:变量已经在缓存行中。

问题1: 我们思考一个问题，那一个数据可以分散在同一个缓存行中吗？

答案是否定的。如果一个数据分散在一个缓存行中，那么CPU加锁就没有作用了。那么 @sun.misc.Contended 注解就是将该变量数据强制刷到不同的缓存行中。

问题2: 如果能够做到强制刷到不同的缓存行中?

根据缓存行的长度，变量不足长度的数据，进行强制填充。将缓存行填充完整。

问题3: 如何填充？

2: Cell类最终是为了解决什么问题？

主要是为了 降低了CAS的范围。

如上图，线程如果CAS操作一直不成功的时候，就会导致CPU一直空转。那么我们是不是可以思考一下，将int数据进行打散，分成若干个值，分别计算，最好将值进行汇总。变成如下图形式:

这些，CPU的竞争降低了很多，降低了CAS的范围。采用的思想就是降低锁粒度，提供并发性能。

2: Striped64类

abstract class Striped64 extends Number {/*** cell数组。当非空时，size是2的幂*/transient volatile Cell[] cells;/***  基础值，当没有竞争的时候使用，使用cas更新*/transient volatile long base;/***  cas操作base属性值*/final boolean casBase(long cmp, long val) {return UNSAFE.compareAndSwapLong(this, BASE, cmp, val);}// Unsafe mechanicsprivate static final sun.misc.Unsafe UNSAFE;private static final long BASE;private static final long CELLSBUSY;private static final long PROBE;static {try {UNSAFE = sun.misc.Unsafe.getUnsafe();Class<?> sk = Striped64.class;BASE = UNSAFE.objectFieldOffset(sk.getDeclaredField("base"));CELLSBUSY = UNSAFE.objectFieldOffset(sk.getDeclaredField("cellsBusy"));Class<?> tk = Thread.class;PROBE = UNSAFE.objectFieldOffset(tk.getDeclaredField("threadLocalRandomProbe"));} catch (Exception e) {throw new Error(e);}}
}

2: LongAdder

public class LongAdder extends Striped64 implements Serializable {/*** Adds the given value.** @param x the value to add*/public void add(long x) {Cell[] as; long b, v; int m; Cell a;if ((as = cells) != null || !casBase(b = base, b + x)) {/*** 能够进入的两种情况* 1: 如果当前数组不为空,或者casBase不成功的时候，也就是说存在竞争的情况下* 2: 数据为空，casBas成功，不存在竞争，直接casBase*/boolean uncontended = true; //是否竞争标识: 控制多个线程是否竞争，同一个cellif (as == null || (m = as.length - 1) < 0 ||(a = as[getProbe() & m]) == null || !(uncontended = a.cas(v = a.value, v + x)))//如果内部cell没有初始化，退回到原来的cas // 如果cell没有初始化，则对cell进行初始化。longAccumulate(x, null, uncontended);}}/** * 所有cell值进行汇总求和，额外加上base*/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;}
}

final void longAccumulate(long x, LongBinaryOperator fn,boolean wasUncontended) {int h;if ((h = getProbe()) == 0) {ThreadLocalRandom.current(); // force initializationh = getProbe();wasUncontended = true;}boolean collide = false;                // True if last slot nonemptyfor (;;) {Cell[] as; Cell a; int n; long v;// 如果cells数组不为null，并且数组已经存在值if ((as = cells) != null && (n = as.length) > 0) {// 当线程访问的下标没有初始化,则进行初始化if ((a = as[(n - 1) & h]) == null) {if (cellsBusy == 0) {       // Try to attach new CellCell r = new Cell(x);   // Optimistically createif (cellsBusy == 0 && casCellsBusy()) {boolean created = false;try {               // Recheck under lockCell[] 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;           // Slot is now non-empty}}collide = false;}//没有发生竞争，设置成trueelse if (!wasUncontended)       // CAS already known to failwasUncontended = true;      // Continue after rehash// 再次尝试一下是否能够自旋成功    else if (a.cas(v = a.value, ((fn == null) ? v + x :fn.applyAsLong(v, x))))break;else if (n >= NCPU || cells != as)collide = false;            // At max size or staleelse if (!collide)collide = true;//继续找个cell，进行自旋else if (cellsBusy == 0 && casCellsBusy()) {try {if (cells == as) {      //进行扩容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}h = advanceProbe(h);} //casCellsBusy() 设置一个标志位,只允许一个线程进来else if (cellsBusy == 0 && cells == as && casCellsBusy()) {//cell数组为空，需要初始化boolean init = false;try {                           // Initialize tableif (cells == as) {Cell[] rs = new Cell[2];rs[h & 1] = new Cell(x);cells = rs;init = true;}} finally {cellsBusy = 0; //释放锁}if (init)//如果初始化了，则退出循环break;}else if (casBase(v = base, ((fn == null) ? v + x : //默认累加fn.applyAsLong(v, x))))//只允许一个线程对cell数组 初始化。使用casBase 来控制多个线程并发初始化，使用cas操作保证只有一个线程能够成功。所以只有一个线程能够创建cell数组，其他线程失败。不能够让其他线程做无谓的自旋，break;  }
}