JDK8 JUC阅读(1)

原子操作类

Posted by Jqy on January 9, 2020

LongAdder

解决高并发下多线程对一个变量不断进行CAS操作带来的性能上的损耗,LongAdder内部维护多个cells元素,分担了多线程对单个变量进行争夺的开销。

LongAdder的真实值为base+cells数组内的值,其中cells数组动态变化。

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内部维护一个volatile值,通过cas进行更新。由于cell通常呈多个出现,因此引入@sun.misc.Contended避免伪共享的影响。

LongAdder的取值如下

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

注意,这里在求值时并没有进行加锁,sum返回的值实际上并不精确。

add方法的实现如下

public void add(long x) {
        Cell[] as; long b, v; int m; Cell a;
    //检查cells是否为null,为null则直接在base上进行累加
        if ((as = cells) != null || !casBase(b = base, b + x)) {
            boolean uncontended = true;
            if (as == null || (m = as.length - 1) < 0 ||
                //确定当前线程访问哪一个cell
                (a = as[getProbe() & m]) == null ||
                //使用cas操作更新cell值
                !(uncontended = a.cas(v = a.value, v + x)))
                longAccumulate(x, null, uncontended);	//上述任一条件不满足执行
        }
    }

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

longAccumulate方法如下,该方法内完成了cells数组的扩容和初始化

final void longAccumulate(long x, LongBinaryOperator fn,
                              boolean wasUncontended) {
        //每个线程第一次执行时,会初始化当前线程的ThreadLocalRandomProbe,该变量用于计算当前线程分配到哪个cell
        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;
            if ((as = cells) != null && (n = as.length) > 0) {
                if ((a = as[(n - 1) & h]) == null) {
                    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;
                }
                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))))
                    break;
                else if (n >= NCPU || cells != as)
                    collide = false;            // At max size or stale
                else if (!collide)
                    collide = true;
                //前置条件不满足(当前cells元素个数小于CPU个数,当前多个线程访问了cells的同一元素)时
                //进行cells数组扩容
                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);
            }
            //初始化cells数组,cellsbusy为标识,0--当前cells没有被初始化或扩容  1--当前cells被使用
            //casCellsBusy用于cellsbusy的切换
            //数组初始化没有用cas,却是线程安全的,因为cellsBusy为volatile
            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;
            }
            else if (casBase(v = base, ((fn == null) ? v + x :
                                        fn.applyAsLong(v, x))))
                break;                          // Fall back on using base
        }
    }