JDK8系列之LongAdder解析

r囧r小猫 2021-06-10 20:38 306阅读 0赞

前言

最近公司的项目在陆续升级jdk8，正好想起之前有人问我的jdk8中新增的`LongAdder`类和`AtomicLong`的区别，就忍不住想探究一番。

源码解析

首先看一下类的定义

public class LongAdder extends Striped64 implements Serializable

然后来看一下类的属性和方法

![format_png][]

这里让我困惑的一个问题是`LongAdder`中没有类似于`AtomicLong`中`getAndIncrement()`或者`incrementAndGet()`这样的原子操作，所以只能通过`increment()`方法和`longValue()`方法的组合来实现更新和获取的动作，然而这样不能保证这个组合操作的原子性，猜想也许`LongAdder`就是不具备这样的机制吧。那么就主要看一下`increment()`和`longValue()`方法。

longValue方法

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

首先看一下获取值的`longValue()`方法，这里调用了`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;
        }

这里出现了一个类`Cell`，这个类是`Striped64`类的静态内部类，因此当`Striped64`对象初始化时并不会连带将`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);
                }
            }
        }

这里有一个`@sun.misc.Contended`标识非常奇怪，查了资料发现是用于解决[伪共享][Link 1]（false sharing）问题的，这里就不深入讲解了。这个类只有一个`cas(long cmp, long val)`方法，做的事情也就是简单的CAS操作，接着来看一下`cells`和`base`的定义，这两个属性都是`Striped64`中的

transient volatile Cell[] cells;
    
        transient volatile long base;

可以看到全部都是volatile类型的，因此都具有可见性，并且可以推测会遇到并发的操作。  
回到`sum()`方法就很好理解了，这个方法返回的是`base`和`cells`数组中所有元素的和，这里的`base`像是一个初始值的作用，我们继续往下分析。

increment方法

public void increment() {
            add(1L);
        }

接着分析`increment()`方法，可以看到该方法就是对`add(long x)`的封装，那么具体来分析一下这个方法

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);
            }
        }

这里重点是`casBase(b = base, b + x)`，来看一下它做了什么

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

由于此处的x为1，那么该方法就是对`Striped64`的`BASE`的值进行累加并返回是否成功，顺带一提这里的`BASE`是`base`值所对应的内存偏移量，所以`casBase(b = base, b + x)`就是对`base`进行CAS操作，执行成功的话操作就结束了，那么什么时候会不成功呢，当然就是并发量大的时候，结合之前分析的`longValue()`方法，这里就可以得出一个结论——当并发不大的时候只对`base`进行更新，获取值得时候当然也只从`base`获取即可，这个时候其实和`AtomicLong`的原理几乎一模一样，看来区别就在于后面的分支，继续往下看。  
首先置标志位`uncontended`为true，从字面意思也可以看出来uncontended意思是没有竞争的。当`casBase`不成功则将进入`add(long x)`的if分支中，分支由四部分组成中间用||连接，必须全部满足才会结束方法执行，否则进入`longAccumulate(x, null, uncontended)`方法，这四部分分别为

1.  as == null，数组为空
2.  (m = as.length - 1) < 0 ，或者数组长度小于1
3.  (a = as\[getProbe() & m\]) == null，或者位置上没有Cell对象,即getProbe()&m其实相当于hashMap里面的tab\[i = (n - 1) & hash\]
4.  !(uncontended = a.cas(v = a.value, v + x))，或者修改cell的值失败

这里的四个条件其实并不是并列的，而是递进式的，1和2判断`cells`数组是否为空，3取`cells`数组中的任意一个元素a判断是否为空，4是对a进行cas操作并将执行结果赋值标志位`uncontended`。从这里可以给出第二个结论，当竞争激烈到一定程度无法对`base`进行累加操作时，会对`cells`数组中某个元素进行更新。

小结

*  如果Cells表为空，尝试用CAS更新base字段，成功则退出；
 *  如果Cells表为空，CAS更新base字段失败，出现竞争，uncontended为true，调用longAccumulate；
 *  如果Cells表非空，但当前线程映射的槽为空，uncontended为true，调用longAccumulate；
 *  如果Cells表非空，且前线程映射的槽非空，CAS更新Cell的值，成功则返回，否则，uncontended设为false，调用longAccumulate。

看到这里大概应该知道为什么LongAdder会比AtomicLong更高效了，没错，唯一会制约AtomicLong高效的原因是高并发，高并发意味着CAS的失败几率更高， 重试次数更多，越多线程重试，CAS失败几率又越高，变成恶性循环，AtomicLong效率降低。 那怎么解决？ LongAdder给了我们一个非常容易想到的解决方案：减少并发，将单一value的更新压力分担到多个value中去，降低单个value的 “热度”，分段更新！！！

这样，线程数再多也会分担到多个value上去更新，只需要增加value就可以降低 value的 “热度” AtomicLong中的 恶性循环不就解决了吗？ cells 就是这个 “段” cell中的value 就是存放更新值的， 这样，当我需要总数时，把cells 中的value都累加一下不就可以了么！！

在看看add方法中的代码，casBase方法可不可以不要，直接分段更新,上来就计算 索引位置，然后更新value？

不是不行，而是有所考虑的,因为，casBase操作等价于AtomicLong中的CAS操作，要知道，LongAdder这样的处理方式是有坏处的，分段操作必然带来空间上的浪费，可以空间换时间，但是，能不换就不换，空间时间都节约.

casBase操作保证了在低并发时，不会立即进入分支做分段更新操作，因为低并发时，casBase操作基本都会成功，只有并发高到一定程度了，才会进入分支

最后来看一下当上述条件无法全部满足时调用的`longAccumulate(x, null, uncontended)`方法

final void longAccumulate(long x, LongBinaryOperator fn,
                                  boolean wasUncontended) {
            int h;
            if ((h = getProbe()) == 0) {
                ThreadLocalRandom.current(); // force initialization
                h = getProbe();  //返回当前线程的threadLocalRandomProbe值
                wasUncontended = true;
            }
            boolean collide = false;                // True if last slot nonempty
            for (;;) {
                Cell[] as; Cell a; int n; long v;
                if ((as = cells) != null && (n = as.length) > 0) {
                    if ((a = as[(n - 1) & h]) == null) {
                        if (cellsBusy == 0) {       // cells数组中对应位置没有数据则插入新对象
                            Cell r = new Cell(x);   
                            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;
                    }
                    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))))         //对该位置的cell元素进行累加
                        break;
                    else if (n >= NCPU || cells != as)
                        collide = false;            // At max size or stale   //判断数组大小是否大于核数
                    else if (!collide)
                        collide = true;
                    else if (cellsBusy == 0 && casCellsBusy()) {         //对cells数组进行扩容，直接扩容为2倍
                        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
                    }
                    h = advanceProbe(h);
                }
                else if (cellsBusy == 0 && cells == as && casCellsBusy()) {  //cellsBusy这里是做为一个自旋锁来使用的
                    boolean init = false;
                    try {                           // 初始化cells数组大小为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;
                }
                else if (casBase(v = base, ((fn == null) ? v + x :
                                            fn.applyAsLong(v, x))))      //对base进行CAS操作
                    break;                          // Fall back on using base
            }
        }

![format_png 1][]

longAccumulate.png

*  如果Cells表为空，尝试获取锁之后初始化表（初始大小为2）；
 *  如果Cells表非空，对应的Cell为空，自旋锁未被占用，尝试获取锁，添加新的Cell；
 *  如果Cells表非空，找到线程对应的Cell，尝试通过CAS更新该值；
 *  如果Cells表非空，线程对应的Cell CAS更新失败，说明存在竞争，尝试获取自旋锁之后扩容,将cells数组扩大，降低每个cell的并发量后再试

这个方法比较长，大致对几个关键点做了注释，该方法主要是用一个死循环对`cells`数组中的元素进行操作，当要更新的位置的元素为空时插入新的cell元素，否则在该位置进行CAS的累加操作，如果CAS操作失败并且数组大小没有超过核数就扩容`cells`数组。

总结

`LongAdder`类与`AtomicLong`类的区别在于高并发时前者将对单一变量的CAS操作分散为对数组`cells`中多个元素的CAS操作，取值时进行求和；而在并发较低时仅对`base`变量进行CAS操作，与`AtomicLong`类原理相同。不得不说这种分布式的设计还是很巧妙的。

[format_png]: /images/20210527/41803f8d531f427e83111f67ef5b7ad4.png
[Link 1]: http://ifeve.com/falsesharing/
[format_png 1]: /images/20210527/471f451fc34d4a748f39536ee646949d.png