当前位置 博文首页 > NicholasGUB:LongAdder源码阅读笔记

    NicholasGUB:LongAdder源码阅读笔记

    作者:NicholasGUB 时间:2021-06-12 18:22

    功能描述

    LongAdder通过创建多个副本对象,解决了多线程使用CAS更新同一个对象造成的CPU阻塞,加快了对线程处理的速度。当多个线程同一时刻更新一个AtomicLong类型的变量时,只有一个线程能够更新成功,其他线程则更新失败,继续尝试更新。

    当使用LongAdder类型的变量时,由于副本数组的存在,线程不一定直接更新变量的本身而是更新副本数组,这样多线程请求的对象变多了,从而减少了更新时间,当需要使用变量值时,返回的值是基础变量的值加上数组内每一个副本的值的和。

    源码解析

    LongAdder继承自Striped64并实现了Serializable接口,而在Striped64类中有一个Cell类
    image

    add方法分析

    首先从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);
        }
    }
    

    从上面的代码可以看到LongAdder的实现主要依靠的是cells数组,如果cells数组为空的话,则尝试使用cas更新基础变量base,如果成功了,则add成功,方法结束,如果cas更新base失败了,则证明此时有其他线程参与base变量的更新,此后的处理与cells不为空一致(如果cells不为空,则在此次方法执行前就已经有多线程参与了更新)。
    当cells数组不为空或者更新base变量失败后,则转而更新cells数组中的副本,此时先判断cells数组是否为空或长度为0,如果为空或长度为0则说明这是第一次操作cells数组,应先初始化cells数组,因此调用方法longAccumulate(x, null, true);
    如果cells数组不为空,则尝试直接访问数组中的副本,getProbe方法代码:

    static final int getProbe() {
        return UNSAFE.getInt(Thread.currentThread(), PROBE);
    }
    

    从上面代码可以看到getProbe方法获取了当前线程内的PROBE变量,而PROBE定义在Striped64类中

        private static final long PROBE;
        static {
            try {
                UNSAFE = sun.misc.Unsafe.getUnsafe();
                Class<?> sk = Striped64.class;
                BASE = UNSAFE.objectFieldOffset
                    (sk.getDeclaredField("base"));
                CELLSBUSY = UNSAFE.objectFieldOffset
                    (sk.getDeclaredField("cellsBusy"));
                Class<?> tk = Thread.class;
                PROBE = UNSAFE.objectFieldOffset
                    (tk.getDeclaredField("threadLocalRandomProbe"));
            } catch (Exception e) {
                throw new Error(e);
            }
        }
    

    从上面的代码可以看出PROBE来自于Thread类的threadLocalRandomProbe变量,是一个线程级变量。getProbe() & m则是获取当前要更新的cell,如果cell为空的话则调用longAccumulate(x, null, true);方法设置cell的值,如果cell不为空的话则使用cas直接更新cell的值,并将更新结果保存在uncontended中,如果uncontended的值为false(即cas更新失败了,此时应该有多个线程同时访问了一个cell),那么继续调用longAccumulate(x, null, false);方法。

    longAccumulate方法分析

    从上面的add方法可以看到,getProbe()取得了Thread类中的threadLocalRandomProbe变量,而threadLocalRandomProbe变量的初始值为0,因为getProbe()方法参与了多线程访问哪一个cell的定位,因此getProbe()的值不可能为0,那么threadLocalRandomProbe变量是在哪里赋值的呢?
    在add方法中观察到,没当方法进行不下去时(base变量更新失败,cells为空,cell更新失败),都会调用longAccumulate方法,因此longAccumulate一定是涉及了cells数组的初始化和扩容,观察longAccumulate方法代码:

    final void longAccumulate(long x, LongBinaryOperator fn,
                                  boolean wasUncontended) {
            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;
                    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);
                }
                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
            }
        }
    
    bk