LongAdder源码分析

矫情吗；* 2022-08-02 09:52 152阅读 0赞

# 1.LongAdder介绍 #

## 为什么要用LongAdder？ ##

我们知道，已经有了AtomicLong，为什么还会出现一个LongAdder。AutomicLong底层使用了CAS操作来控制并发的。在并发量级比较小的情况下，线程冲突的概率比较小，自旋次数少。但是，高并发的情况下，多个线程同时进行自旋操作，就会出现大量失败并一直自旋的情况，这个时候AutomicLong的性能就下降了。所以引入了LongAdder，解决高并发环境下AtomicLong自旋瓶颈的问题。

## LongAdder设计思想？ ##

LongAdder是一种以空间换时间的解决方案，其内部维护了一个base值和一个cell数组，当线程不存在竞争的时候，首先将值写入到base中，如果线程之间存在竞争，将其写入到cell数组的一个槽中。然后将base值和cell数组中的所有值求和就可以得到最终的LongAdder的值了。

![img][]

# 2.LongAdder分析 #

LongAdder内部只有一个空的构造方法

public class LongAdder extends Striped64 implements Serializable

然后提供了一个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);
        }
    }

其核心还是继承Striped64类

Striped64类的属性值介绍:

//表示当前计算机CPU数量，什么用？ 控制cells数组长度的一个关键条件
    static final int NCPU = Runtime.getRuntime().availableProcessors();
    
    //cell数组，大小为2的倍数
    transient volatile Cell[] cells;
    
    //没有发生过竞争时，数据会累加到 base上;当cells扩容时，需要将数据写到base中
    transient volatile long base;
    
    //初始化cells或者扩容cells都需要获取锁，0 表示无锁状态，1 表示其他线程已经持有锁了
    transient volatile int cellsBusy;

其内部定一个了一个静态内部类，就是Cell对象，对象有value值，并提供了一个cas方法，通过UNSAFE类的CAS操作设置该属性的值。

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

演示LongAdder的使用:

LongAdder longAdder = new LongAdder();
    longAdder.add(1);

public void add(long x) {
        //as为cell数组，b为base值，m为cell数组长度，a为数组的一个槽的值
        Cell[] as; long b, v; int m; Cell a;
        //cells为父类的属性值
        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);
        }
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI1NzI5NTU_size_16_color_FFFFFF_t_70][]

分析longAccumulate方法

//cells未初始化且CAS失败的情况会调用该方法
    final void longAccumulate(long x, LongBinaryOperator fn,
                              boolean wasUncontended) {
        //表示线程的hash值
        int h;
        //获取当前线程的hash值，获取不到，说明当前线程还未分配hash值，设置当前线程hash值
        if ((h = getProbe()) == 0) {
            ThreadLocalRandom.current(); // force initialization
            h = getProbe();
            wasUncontended = true;
        }
        //扩容意向，false一定不会扩容，true可能会扩容,当竞争很激烈的时候可能会触发扩容
        boolean collide = false;                // True if last slot nonempty
        //自旋
        for (;;) {
            //as表示cells的引用，a表示当前线程的cell，n表示cells数组长度，v表示期望值
            Cell[] as; Cell a; int n; long v;
            //CASE1:表示cells已经初始化，当前线程应该将数据写入到对应的cell中
            if ((as = cells) != null && (n = as.length) > 0) {
                //CASE1.1:表示当前数组下标位置的cell为空
                if ((a = as[(n - 1) & h]) == null) {
                    //锁未被占用的情况，创建一个新的cell
                    if (cellsBusy == 0) {       
                        Cell r = new Cell(x);  
                        //锁未被占用，且可以获取到锁，把cellsBusy设置为1
                        if (cellsBusy == 0 && casCellsBusy()) {
                            //是否创建成功标记
                            boolean created = false;
                            //以下是给cells数组赋值的情况，赋值成功释放锁
                            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;
                }
                //CASE1.2:wasUncontended全局没有地方被修改为false，感觉没啥用
                else if (!wasUncontended)       // CAS already known to fail
                    wasUncontended = true;      // Continue after rehash
                //CASE1.3:cells已经初始化了，当前线程竞争成功，通过CAS给cell重新设置值
                else if (a.cas(v = a.value, ((fn == null) ? v + x :
                                             fn.applyAsLong(v, x))))
                    break;
                //CASE1.4:
                else if (n >= NCPU || cells != as)
                    collide = false;            // At max size or stale
                else if (!collide)
                    collide = true;
                //扩不扩容还要看能不能拿到锁,当线程竞争很激烈的时候扩容
                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
                }
                h = advanceProbe(h);
            }
            //CASE2:cells未初始化
            //当前线程未加锁，cells == as的判断是因为其他线程可能在给as赋值后改变了cells，
            //casCellsBusy为true表示当前线程可以获取到锁，会把cellsBusy设置为1，false代表其他线程持有锁
            else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
                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;
            }
            //CASE3:cells正在初始化过程中
            //在case1和case2不满足的情况，说明当前线程获取不到锁且cells正在被其他线程初始化
            //当前线程将值累加到base，当cells被其他线程初始化后，fn在add方法中传入的是null
            else if (casBase(v = base, ((fn == null) ? v + x :
                                        fn.applyAsLong(v, x))))
                break;                          // Fall back on using base
        }
    }

//sum方法就是将base和每个cell的value叠加
    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;
    }

[img]: /images/20220731/8b2881f295bf4e5eb62d8623c8481b80.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI1NzI5NTU_size_16_color_FFFFFF_t_70]: /images/20220731/54818327110a4111882ee8c67d533d27.png