分析并发编程之LongAdder原理

public V put(K key, V value) {
    return putVal(key, value, false);
}

final V putVal(K key, V value, boolean onlyIfAbsent) {
    ...
    addCount(1L, binCount);
    return null;
}

private final void addCount(long x, int check) {
    CounterCell[] as; long b, s;
    if ((as = counterCells) != null ||
            !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
        CounterCell a; long v; int m;
        boolean uncontended = true;
        if (as == null || (m = as.length - 1) < 0 ||
                (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
                !(uncontended =
                        U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
            fullAddCount(x, uncontended);
            return;
        }
        if (check <= 1)
            return;
        s = sumCount();
    }
    ...
}

LongAdder adder = new LongAdder();
int num = 0;

@Test
public void test5() throws InterruptedException {
    Thread[] threads = new Thread[10];
    for (int i = 0; i < 10; i++) {
        threads[i] = new Thread(() -> {
            for (int j = 0; j < 10000; j++) {
                adder.add(1);
                num += 1;
            }
        });
        threads[i].start();
    }
    for (int i = 0; i < 10; i++) {
        threads[i].join();
    }
    System.out.println("adder:" + adder);
    System.out.println("num:" + num);
}

/**
 * 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;

public void add(long x) {
    Cell[] as; long b, v; int m; Cell a;
    /**
     * 这里优先判断了cell数组是否为空，之后才判断base字段的cas累加
     * 意味着如果线程不发生竞争，cell数组一直为空，那么所有的累加操作都会累加到base上
     * 而一旦发生过一次竞争导致cell数组不为空，那么所有的累加操作都会优先作用于数组中的对象上
     */
    if ((as = cells) != null || !casBase(b = base, b + x)) {
        /**
         * 这个字段是用来标识在对cell数组中的对象进行累加操作时是否发生了竞争
         * 如果发生了竞争，那么在longAccumulate方法中会多进行一次rehash的自旋
         * 这个在后面的方法中详细说明，这里先有个印象
         * true表示未发生竞争
         */
        boolean uncontended = true;
        /**
         * 如果cell数组为空或者长度为0则直接进入主逻辑方法
         */
        if (as == null || (m = as.length - 1) < 0 ||
                /**
                 * 这里的getProbe()方法可以认为就是获取线程的hash值
                 * hash值与(数组长度-1)进行位与操作后得到对应的数组下标
                 * 判断该元素是否为空，如果不为空那么就会尝试累加
                 * 否则进入主逻辑方法
                 */
                (a = as[getProbe() & m]) == null ||
                /**
                 * 对数组下标的元素进行cas累加，如果成功了，那么就可以直接返回
                 * 否则进入主逻辑方法
                 */
                !(uncontended = a.cas(v = a.value, v + x)))
            longAccumulate(x, null, uncontended);
    }
}

final void longAccumulate(long x, LongBinaryOperator fn,
                          boolean wasUncontended) {
  ...
}

int h;//线程的hash值，在后面的逻辑中会用到
if ((h = getProbe()) == 0) {
    ThreadLocalRandom.current(); // force initialization
    h = getProbe();
    wasUncontended = true;
}

/**
 * 用来标记某个线程在上一次循环中找到的数组下标是否已经有Cell对象了
 * 如果为true，则表示数组下标为空
 * 在主逻辑的循环中会用到
 */
boolean collide = false;
/**
 * 死循环，提供自旋操作
 */
for (; ; ) {
    Cell[] as;
    Cell a;
    int n;//cell数组长度
    long v;//需要被累积的值
    /**
     * 如果cells数组不为空，且已经被某个线程初始化成功，那么就会进入主逻辑，这个后面详细解释
     */
    if ((as = cells) != null && (n = as.length) > 0) {
        ...
        /**
         * 如果数组为空，那么就需要初始化一个Cell数组
         * cellsBusy用来标记cells数组是否能被操作，作用相当于一个锁
         * cells == as 判断是否有其他线程在当前线程进入这个判断之前已经初始化了一个数组
         * casCellsBusy 用一个cas操作给cellsBusy字段赋值为1，如果成功可以认为拿到了操作cells数组的锁
         */
    } else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
        /**
         * 这里就是初始化一个数组，不解释了
         */
        boolean init = false;
        try {                           
            if (cells == as) {
                Cell[] rs = new Cell[2];
                rs[h & 1] = new Cell(x);
                cells = rs;
                init = true;
            }
        } finally {
            cellsBusy = 0;
        }
        if (init)
            break;
        /**
         * 如果当前数组是空的，又没有竞争过其他线程
         * 那么就再次尝试去给base赋值
         * 如果又没竞争过（感觉有点可怜），那么就自旋
         * 另外提一下方法签名中的LongBinaryOperator对象就是用在这里的，不影响逻辑
         */
    } else if (casBase(v = base, ((fn == null) ? v + x :
            fn.applyAsLong(v, x))))
        break;                          // Fall back on using base
}

/**
 * 如果cells数组不为空，且已经被某个线程初始化成功，进入主逻辑
 */
if ((as = cells) != null && (n = as.length) > 0) {
    /**
     * 如果当前线程的hash值对应的数组元素为空
     */
    if ((a = as[(n - 1) & h]) == null) {
        /**
         * Cell数组并未被其他线程操作
         */
        if (cellsBusy == 0) {
            /**
             * 这里没有理解作者为什么会在这里初始化单个Cell
             * 作者这里的注释是Optimistically create，如果有理解的同学可以说一下
             */
            Cell r = new Cell(x);
            /**
             * 在此判断cell锁的状态，并尝试加锁
             */
            if (cellsBusy == 0 && casCellsBusy()) {
                boolean created = false;
                try {
                    /**
                     * 这里对数组是否为空等状态再次进行校验
                     * 如果校验通过，那么就将之前new的Cell对象放到Cell数组的该下标处
                     */
                    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;
                /**
                 * 走到这里说明在判空和拿到锁之间正好有其他线程在该下标处创建了一个Cell
                 * 因此直接continue，不rehash，下次就不会进入到该分支了
                 */
                continue;
            }
        }
        /**
         * 当执行到这里的时候，因为是在 if ((a = as[(n - 1) & h]) == null) 这个判断逻辑中
         * 就说明在第一个if判断的时候该下标处没有元素，所以赋值为false
         * collide的意义是：上一次循环中找到的数组下标是否已经有Cell对象了
         * True if last slot nonempty
         */
        collide = false;
    /**
     * 这个字段如果为false，说明之前已经和其他线程发过了竞争
     * 即使此时可以直接取尝试cas操作，但是在高并发场景下
     * 这2个线程之后依然可能发生竞争，而每次竞争都需要自旋的话会很浪费cpu资源
     * 因此在这里先直接增加自旋一次，在for的最后会做一次rehash
     * 使得线程尽快地找到自己独占的数组下标
     */
    } else if (!wasUncontended) 
        wasUncontended = true;
    /**
     * 尝试给hash对应的Cell累加，如果这一步成功了，那么就返回
     * 如果这一步依然失败了，说明此时整体的并发竞争非常激烈
     * 那就可能需要考虑扩容数组了
     * （因为数组初始化容量为2，如果此时有10个线程在并发运行，那就很难避免竞争的发生了）
     */
    else if (a.cas(v = a.value, ((fn == null) ? v + x :
            fn.applyAsLong(v, x))))
        break;
    /**
     * 这里判断下cpu的核数，因为即使有100个线程
     * 能同时并行运行的线程数等于cpu数
     * 因此如果数组的长度已经大于cpu数目了，那就不应当再扩容了
     */
    else if (n >= NCPU || cells != as)
        collide = false;
    /**
     * 走到这里，说明当前循环中根据线程hash值找到的数组下标已经有元素了
     * 如果此时collide为false，说明上一次循环中找到的下边是没有元素的
     * 那么就自旋一次并rehash
     * 如果再次运行到这里，并且collide为true，就说明明竞争非常激烈，应当扩容了
     */
    else if (!collide)
        collide = true;
    /**
     * 能运行到这里，说明需要扩容数组了
     * 判断锁状态并尝试获取锁
     */
    else if (cellsBusy == 0 && casCellsBusy()) {
        /**
         * 扩容数组的逻辑，这个扩容比较简单，就不解释了
         * 扩容大小为2倍
         */
        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，因为扩容过了，就先不rehash了
        */
        continue;               
    }
    /**
     * 做一个rehash，使得线程在下一个循环中可能找到独占的数组下标
     */
    h = advanceProbe(h);
}

public final int getAndIncrement() {
    return unsafe.getAndAddInt(this, valueOffset, 1);
}

public final int getAndAddInt(Object var1, long var2, int var4) {
    int var5;
    do {
        var5 = this.getIntVolatile(var1, var2);
    } while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));

    return var5;
}