Java并发编程之LongAdder和LongAccumulator源码探究

Posted 2020-11-12 妮蔻

一.LongAdder原理

LongAdder类是JDK1.8新增的一个原子性操作类。AtomicLong通过CAS算法提供了非阻塞的原子性操作，相比受用阻塞算法的同步器来说性能已经很好了，但是JDK开发组并不满足于此，因为非常搞并发的请求下AtomicLong的性能是不能让人接受的。

如下AtomicLong 的incrementAndGet的代码，虽然AtomicLong使用CAS算法，但是CAS失败后还是通过无限循环的自旋锁不多的尝试，这就是高并发下CAS性能低下的原因所在。源码如下：

　　public final long incrementAndGet() {
        for (;;) {
            long current = get();
            long next = current + 1;
            if (compareAndSet(current, next))
                return next;
        }
    }

在高并发下N多线程同时去操作一个变量会造成大量线程CAS失败，然后处于自旋状态，这样导致大大浪费CPU资源，降低了并发性。

既然AtomicLong性能问题是由于过多线程同时去竞争同一个变量的更新而降低的，那么如果把一个变量分解为多个变量，让同样多的线程去竞争多个资源，那么性能问题不久迎刃而解了吗？

没错，因此，JDK8 提供的LongAdder就是这个思路。下面通过图形来标示两者的不同，如下图：

如上图 AtomicLong 是多个线程同时竞争同一个变量情景。

 

如上图所示，LongAdder则是内部维护多个Cell变量，每个Cell里面有一个初始值为0的long型变量，在同等并发量的情况下，争夺单个变量的线程会减少，这是变相的减少了争夺共享资源的并发量，另外多个线程在争夺同一个原子变量时候，

如果失败并不是自旋CAS重试，而是尝试获取其他原子变量的锁，最后当获取当前值时候是把所有变量的值累加后再加上base的值返回的。

 

LongAdder维护了要给延迟初始化的原子性更新数组和一个基值变量base数组的大小保持是2的N次方大小，数组表的下标使用每个线程的hashcode值的掩码表示，数组里面的变量实体是Cell类型。

Cell 类型是Atomic的一个改进，用来减少缓存的争用，对于大多数原子操作字节填充是浪费的，因为原子操作都是无规律的分散在内存中进行的，多个原子性操作彼此之间是没有接触的，但是原子性数组元素彼此相邻存放将能经常共享缓存行，也就是伪共享。所以这在性能上是一个提升。

另外由于Cells占用内存是相对比较大的，所以一开始并不创建，而是在需要时候再创建，也就是惰性加载，当一开始没有空间时候，所有的更新都是操作base变量。

 

接下来进行LongAdder代码简单分析

这里我只是简单的介绍一下代码的实现，详细实现，大家可以翻看代码去研究。为了降低高并发下多线程对一个变量CAS争夺失败后大量线程会自旋而造成降低并发性能问题，LongAdder内部通过根据并发请求量来维护多个Cell元素(一个动态的Cell数组)来分担对单个变量进行争夺资源。

首先我们先看LongAdder的构造类图，如下图：

 

 可以看到LongAdder继承自Striped64类，Striped64内部维护着三个变量，LongAdder的真实值其实就是base的值与Cell数组里面所有Cell元素值的累加，base是个基础值，默认是0，cellBusy用来实现自旋锁，当创建Cell元素或者扩容Cell数组时候用来进行线程间的同步。

 

接下来进去源码如看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 技术
        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的构造很简单，内部维护一个声明volatile的变量，这里声明为volatile是因为线程操作value变量时候没有使用锁，为了保证变量的内存可见性这里只有声明为volatile。另外这里就是先前文件所说的使用Unsafe类的方法来设置value的值

 

接下来进入LongAdder的源码里面去看几个重要的方法，如下：

　　1.long sum() 方法：返回当前的值，内部操作是累加所有 Cell 内部的 value 的值后累加 base，如下代码，由于计算总和时候没有对 Cell 数组进行加锁，所以在累加过程中可能有其它线程对 Cell 中的值进行了修改，也有可能数组进行了扩容，所以 sum 返回的值并不是非常精确的，

返回值并不是一个调用 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;
}

 

　　2.void reset() 方法：重置操作，如下代码把 base 置为 0，如果 Cell 数组有元素，则元素值重置为 0。源码如下：

　　

　　public void reset() {
        Cell[] as = cells; Cell a;
        base = 0L;
        if (as != null) {
            for (int i = 0; i < as.length; ++i) {
                if ((a = as[i]) != null)
                    a.value = 0L;
            }
        }
   }

 

　　3.long sumThenReset() 方法：是sum 的改造版本，如下代码，在计算 sum 累加对应的 cell 值后，把当前 cell 的值重置为 0，base 重置为 0。 当多线程调用该方法时候会有问题，比如考虑第一个调用线程会清空 Cell 的值，后一个线程调用时候累加时候累加的都是 0 值。

　　源码如下：

　　

　　public long sumThenReset() {
        Cell[] as = cells; Cell a;
        long sum = base;
        base = 0L;
        if (as != null) {
            for (int i = 0; i < as.length; ++i) {
                if ((a = as[i]) != null) {
                    sum += a.value;
                    a.value = 0L;
                }
            }
        }
        return sum;
    }

 

　　4.long longValue() 等价于 sum()，源码如下：

　　

    public long longValue() {
        return this.sum();
    }

 

　　5.void add(long x） 累加增量 x 到原子变量，这个过程是原子性的。源码如下：

　　

　　public void add(long x) {
        Cell[] as; long b, v; int m; Cell a;
        if ((as = cells) != null || !casBase(b = base, b + x)) {//(1)
            boolean uncontended = true;
            if (as == null || (m = as.length - 1) < 0 ||//(2)
                (a = as[getProbe() & m]) == null ||//(3)
                !(uncontended = a.cas(v = a.value, v + x)))//(4)
                longAccumulate(x, null, uncontended);//(5)
        }
    }

    final boolean casBase(long cmp, long val) {
        return UNSAFE.compareAndSwapLong(this, BASE, cmp, val);
    }

可以看到上面代码，当第一个线程A执行add时候，代码（1）会执行casBase方法，通过CAS设置base为 X， 如果成功则直接返回，这时候base的值为1。

假如多个线程同时执行add时候，同时执行到casBase则只有一个线程A成功返回，其他线程由于CAS失败执行代码（2），代码（2）是获取cells数组的长度，如果数组长度为0，则执行代码（5），否则cells长度不为0，说明cells数组有元素则执行代码（3），

代码（3）首先计算当前线程在数组中下标，然后获取当前线程对应的cell值，如果获取到则执行（4）进行CAS操作，CAS失败则执行代码（5）。

代码（5）里面是具体进行数组扩充和初始化，这个代码比较复杂，这里就不讲解了，有兴趣的可以进去看看。

 

二.LongAccumulator类源码分析

LongAdder类是LongAccumulator的一个特例，LongAccumulator提供了比LongAdder更强大的功能，如下构造函数，其中accumulatorFunction是一个双目运算器接口，根据输入的两个参数返回一个计算值，identity则是LongAccumulator累加器的初始值。

public LongAccumulator(LongBinaryOperator accumulatorFunction,long identity) {
        this.function = accumulatorFunction;
        base = this.identity = identity;
}
public interface LongBinaryOperator {

       //根据两个参数计算返回一个值
       long applyAsLong(long left, long right);
}

上面提到LongAdder 其实就是LongAccumulator 的一个特例，调用LongAdder 相当使用下面的方式调用 LongAccumulator。

　　 LongAdder adder = new LongAdder();
    LongAccumulator accumulator = new LongAccumulator(new LongBinaryOperator() {

        @Override
        public long applyAsLong(long left, long right) {
            return left + right;
        }
    }, 0);

LongAccumulator相比LongAdder 可以提供累加器初始非0值，后者只能默认为0，另外前者还可以指定累加规则，比如不是累加而相乘，只需要构造LongAccumulator 时候传入自定义双目运算器即可，后者则内置累加规则。

 

从下面代码知道LongAccumulator相比于LongAdde的不同在于casBase的时候，后者传递的是b+x,而前者则是调用了r=function.applyAsLong(b=base.x)来计算。

LongAdder类的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);
        }
    }

 

LongAccumulator的accumulate方法的源码如下：

　　public void accumulate(long x) {
        Cell[] as; long b, v, r; int m; Cell a;
        if ((as = cells) != null ||
            (r = function.applyAsLong(b = base, x)) != b && !casBase(b, r)) {
            boolean uncontended = true;
            if (as == null || (m = as.length - 1) < 0 ||
                (a = as[getProbe() & m]) == null ||
                !(uncontended =
                  (r = function.applyAsLong(v = a.value, x)) == v ||
                  a.cas(v, r)))
                longAccumulate(x, function, uncontended);
        }
    }

 

另外LongAccumulator调用longAccumulate时候传递的是function,而LongAdder是null,从下面代码可以知道当fn为null，时候就是使用v+x  加法运算，这时候就等价于LongAdder,fn不为null的时候则使用传递的fn函数计算，如果fn为加法则等价于LongAdder;

else if (casBase(v = base, ((fn == null) ? v + x :fn.applyAsLong(v, x))))
    　　　// Fall back on using base
　　　　　　break;