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

深入理解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 mechanics
    private 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 mechanics
    private 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; //是否竞争标识: 控制多个线程是否竞争，同一个cell
            if (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 initialization
        h = getProbe();
        wasUncontended = true;
    }
    boolean collide = false;                // True if last slot nonempty
    for (;;) {
        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 Cell
                    Cell r = new Cell(x);   // Optimistically create
                    if (cellsBusy == 0 && casCellsBusy()) {
                        boolean created = false;
                        try {               // Recheck under lock
                            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;           // Slot is now non-empty
                    }
                }
                collide = false;
            }//没有发生竞争，设置成true
            else if (!wasUncontended)       // CAS already known to fail
                wasUncontended = 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 stale
            else 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 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;
        }
        else if (casBase(v = base, ((fn == null) ? v + x : //默认累加
                                    fn.applyAsLong(v, x))))
            //只允许一个线程对cell数组 初始化。使用casBase 来控制多个线程并发初始化，使用cas操作保证只有一个线程能够成功。所以只有一个线程能够创建cell数组，其他线程失败。不能够让其他线程做无谓的自旋，
            break;  
    }
}