jdk1.8 LongAdder 难点解析

Posted 2022-06-07 kobebyrant

写这篇文章是因为jdk1.8 concurrentHashMap计算容量扩容前用到了这个类，所以之后就研究了一下这个类。

本文主要有几点内容：

1. 为什么需要LongAdder这个类
2. LongAdder实现原理解析

为什么需要LongAdder这个类

看这个东西前先看看AtomicLong（1.5）。这个类的主要功能是可以原子更新long类型的值。

实现也非常简单：

/**
 * Atomically increments by one the current value.
 *
 * @return the updated value
 */
public final long incrementAndGet() 
    for (;;) 
        long current = get();
        long next = current + 1;
        if (compareAndSet(current, next))
            return next;
    

以上为jdk1.7的实现，1.8的该方法实现被封装到了UNSAFE类里面的了。

以上实现就是很简单的自旋cas操作。

我们再看看官网对LongAdder的描述：

简单翻译一下核心内容

内部会有一个或者多个变量来共同维护一个初始化值为0的long变量。当这个long被更新的时候，如果线程之间发生竞争，那这个类里面的变量数目会根据竞争情况动态调整来减少竞争。sum方法好longValue方法返回的是当前变量的总和来维护这个sum。

当面临多线程更新一个共有的值的时候（比如收集统计信息），这个类的表现会比AtomicLong来的更好，这个类通常不用于细粒度的同步控制。在低更新竞争的时候，两者会有近似的性能。但是当高竞争压力的时候，这个LongAdder的吞吐量显然会比AtomicLong更高，同时也花费更高的空间消耗。

我们会看Atomic的实现，这个实现就是单纯地对一个变量进行自旋cas，那高并发的情境下，显然会很多变量cas失败，那并发吞吐量肯定是上不去的。（该情况下其实类似于锁争用了）

那我们思考一下吞吐量应该怎么上去？

LongAdder实现原理解析

先看看继承关系

可以隐约感到LongAdder的实现是比较依赖于其父类Striped64的

然后我们看核心方法的实现

/**
 * 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)) 
        boolean uncontended = true;
        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);
    

我们先看看方法中未定义的cells和base，点进去发现他们都是Striped64里面的成员变量：

/**
 * Table of cells. When non-null, size is a power of 2.
 */
transient volatile Cell[] cells;

/**
 * Base value, used mainly when there is no contention, but also as
 * a fallback during table initialization races. Updated via CAS.
 */
transient volatile long base;

/**
 * Spinlock (locked via CAS) used when resizing and/or creating Cells.
 */
transient volatile int cellsBusy;

这些解释都让人看起来比较费解。

cellsBusy比较好懂，就是用来当自旋锁的。

我们继续看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);
        
    

这个类的实现很简单，我们看到每个cell会维护一个long变量，然后只实现了cas方法，我们能看出Cell里面的value是保证了原子性的。

Striped64的类头的注释非常长，但是不得不看，我这里解释一下基本内容：

Striped64中维护了两个变量来进行原子更新：懒初始化的cells数组和base。

由于Cells数组占据空间比较大，当不需要他们的时候是不会创建的。如果是没有线程竞争的话，全部更新操作都是基于base变量进行。当第一次出现竞争的时候，cells数组会初始化为size=2，此后每次竞争都会*2.

再回看add方法：

/**
 * 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)) 
        boolean uncontended = true;
        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);
    

先看if，这种短路或的代码，都是先看怎么样能够不进入函数体内的。

如果cells为null，则直接尝试casBase来进行加法计算。

如果cells非空或者casBase失败进入if函数体：

这里有个uncontend变量，表征为在第一个if之后没有经过竞争。

然后看第二个if判定条件，这里我们看看什么时候能不进去第二个if体内。

先对cells判空，然后用一个利用当前线程计算出来的的hash值（getProbe）和以当前cells的长度的掩码相与得出要在第几个cell操作并赋值给a。然后如果里面有值的话，对他进行cas，只有成功地对某个cell进行cas才能结束。

不满足这四个条件的会进入longAccumulate，如果到了cas阶段失败的话wasUncontended会置为false表征经历过了竞争。

总结一下，只有对base或者某个cell进行cas加操作成功，才能不进入longAccumulate，那我们可以猜测longAccumulate这里面会有一些初始化cell，resize或者自旋的操作。

整个LongAdder的核心就是longAccumulate了。

这个类方法头也说了这个方法是为了处理初始化，resize和创建新的cells数组的。

这个方法很长，建议先看标注好的外层if的B和C处，再看A里面的H’处。细节流程都标注出来了

    /**
     * Handles cases of updates involving initialization, resizing,
     * creating new Cells, and/or contention. See above for
     * explanation. This method suffers the usual non-modularity
     * problems of optimistic retry code, relying on rechecked sets of
     * reads.
     *
     * @param x the value
     * @param fn the update function, or null for add (this convention
     * avoids the need for an extra field or function in LongAdder).
     * @param wasUncontended false if CAS failed before call
     */
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;
        //A 如果cells数组非空则进来这里，同时这个if会将cells初始化给as
        if ((as = cells) != null && (n = as.length) > 0) 
            //A' 第x个cell为空，则初始化cell
            if ((a = as[(n - 1) & h]) == null) 
                //下面流程跟B流程有点像，也和单例创建的双重检查很像
                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;
            
            //B' 这个是add方法cascell的失败的时候放置的，这里再给一次机会他换个hash去自旋尝试获取锁
            else if (!wasUncontended)       // CAS already known to fail
                wasUncontended = true;      // Continue after rehash
            //C' 这个方法是自旋尝试casCell，是整个方法的核心
            else if (a.cas(v = a.value, ((fn == null) ? v + x :
                                         fn.applyAsLong(v, x))))
                break;
            //D' 尝试失败的时候其实已经是冲突了，然后这里意思是如果当前表容量比cpu数量更大则不进入扩容流程
            else if (n >= NCPU || cells != as)
                collide = false;            // At max size or stale
            //E' 冲突位置为true
            else if (!collide)
                collide = true;
            //F' 经过两次冲突则来到扩容流程，依然是双重检测操作
            else if (cellsBusy == 0 && casCellsBusy()) 
                try 
                    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
            
            //G' 这里是很核心的地方！每一次尝试拿锁失败的时候都会换个hash值再去试
            h = advanceProbe(h);
        
        //前面说过cellBusy是用来当做一个锁用的
        //B：如果cells为空而且cellBusy未被其他线程抢走则尝试抢锁，cells == as为了判断引用的有效性（可能会被其他线程清空掉）
        else if (cellsBusy == 0 && cells == as && casCellsBusy()) 
            //抢锁成功会进来
            boolean init = false;
            try   
                // 初始化表，初始容量为2
                if (cells == as) 
                    Cell[] rs = new Cell[2];
                    rs[h & 1] = new Cell(x);
                    cells = rs;
                    init = true;
                
                //无论如何都要解锁
             finally 
                cellsBusy = 0;
            
            if (init)
                break;
        
        //C 初始化抢锁失败的时候会来到这里，再试试能不能用casBase来解决问题
        else if (casBase(v = base, ((fn == null) ? v + x :
                                    fn.applyAsLong(v, x))))
            break;                          // Fall back on using base
    

其实这个方法的核心也是自旋乐观操作，套路都是先给cas失败的一次机会，然后每次要改值的时候都是乐观cas加上双重或者多重检测来保证安全性。

然后这个方法里面的第一个if（A）里面的H‘处也是一个亮点，每次都换一个hash值去尝试cas到不同的cell，这样是将压力均摊到cells数组的每个cell的关键。然后这个collide位也是和之前的1.7concurrentHashMap的retries效果一样的。

通过两次collide来动态扩容，最大的容量比cpu核数量要小。

至此，Add方法解释完毕。

接下来看看另一个主要方法sum：

/**
 * Returns the current sum.  The returned value is <em>NOT</em> an
 * atomic snapshot; invocation in the absence of concurrent
 * updates returns an accurate result, but concurrent updates that
 * occur while the sum is being calculated might not be
 * incorporated.
 *
 * @return the sum
 */
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;

方法头的注释一定要看：返回值不符合原子性，只保证计算的时候没有并发更新发生的时候是准确的。所以这个longAdder不应该作为需要细粒度控制的量来使用。

因此我们可以总结一下可以得知：longAdder是最好用于计算统计量的，误差不大就好，不能完全准确，但是能保证吞吐量；如果要细粒度的控制值的准确性，应该用atomicLong。

其实我认为这个atomicLong和LongAdder的思想上的对比类似于hashtable和1.7concurrentHashMap。

concurrentHashMap里面每一个segment也维护了一个内部的哈希表。

补充说明

其中Cell加上了这个注解@sun.misc.Contended是为了减少缓存冲突。这个注解会引发一个操作叫做padding，会有一个效果就是让这些原子对象不规则地分布在内存中因此能做到互不干扰。如果不加上这个 @sun.misc.Contended注解，数组中的Cell们会被放置在相邻的位置，这样就大概率会共享缓存行，这样会对性能有很大的杀伤。

还有一个细节疑问:

            if (cellsBusy == 0)        // Try to attach new Cell
                Cell r = new Cell(x);   // Optimistically create

以上代码是longAccumulate中的，为什么不放在casCellBusy内层呢？

知道答案的大神们麻烦可以帮我解答一下，谢谢！