乐趣区

关于java:LongAdder源码分析

简述

LongAdder 是 JDK8 新增的一个用于高并发场景下进行统计的类。以前 AtomicInteger、AtomicLong 等通过应用 CAS 的形式来更新变量,比 synchronized 这些阻塞算法领有更好的性能。然而在高并发状况下,大量线程同时去更新一个变量,任意一个工夫点只有一个线程可能胜利,绝大部分的线程在尝试更新失败后,会通过自旋的形式再次进行尝试,重大占用了 CPU 的工夫片。导致 AtomicInteger、AtomicLong 在高并发场景下的性能重大升高,所以产生了 LongAdder 来满足高并发场景下的统计。

LongAdder 原理


没有竞争的时候,线程会对 base 外面的 value 进行批改
一旦呈现高并发场景下的多线程竞争,那么 LongAdder 会初始化一个 cell 数组,而后对每个线程获取对应的 hash 值,之后通过 hash & (size -1)[size 为 cell 数组的长度]将每个线程定位到对应的 cell 单元格,之后这个线程将值写入对应的 cell 单元格中的 value,之后将所有 cell 单元格的 value 和 base 中的 value 进行累加求和失去最终的值。
在整个过程中,波及到 cell 的初始化,线程定位到单元格,以及 cell 数组的扩容等一系列过程。接下来,咱们会对 LongAdder 的 add 办法源码进行剖析,来学习 LongAdder 在解决高并发场景下的思维

源码剖析

首先,LongAdder 继承 Striped64 这个类。在这个类外面定义了如下要害变量

 // 获取以后机器的 CPU 数量
   static final int NCPU = Runtime.getRuntime().availableProcessors();
// cell 数组 
    transient volatile Cell[] cells;
// 没有竞争时写入的 base 值
    transient volatile long base;
// cell 数组对应的锁 0 示意以后 cell 数组没有线程应用,1 示意以后数组曾经有线程占用
    transient volatile int cellsBusy;
        final boolean casBase(long cmp, long val) {return UNSAFE.compareAndSwapLong(this, BASE, cmp, val);
    }
// 通过 CAS 来获取以后 cell 数组的锁
    final boolean casCellsBusy() {return UNSAFE.compareAndSwapInt(this, CELLSBUSY, 0, 1);
    }
// 获取以后线程的 hash 值
    static final int getProbe() {return UNSAFE.getInt(Thread.currentThread(), PROBE);
    }

上面是 LongAdder 的 add 办法

/*
as 是以后 cell 数组的援用
b 代表 base 的值
m 代表 cell 数组的长度
a 代表以后线程 hash 之后定位到的 cell 单元格
v 代表期望值
uncontended=true 代表以后线程对应的 cell 单元格 CAS 胜利,uncontended=false 示意以后线程对应的 cell 单元格 CAS 写入失败,呈现竞争。*/
    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 办法的三种状态
                1,as == null || (m = as.length - 1) 以后 cell 数组为空,没有初始化
                2,a = as[getProbe() & m]) == null 获取以后线程的 hash 值而后和数组长度进行 & 运算失去对应的 cell 单元格为空
                3,uncontended = a.cas(v = a.value, v + x) 以后线程对应的 cell 单元格 CAS 失败,呈现竞争
                */
                longAccumulate(x, null, uncontended);
        }
    }

上面是 longAccumulate 办法

   */
    final void longAccumulate(long x, LongBinaryOperator fn,
                              boolean wasUncontended) {
        int h;
        if ((h = getProbe()) == 0) {ThreadLocalRandom.current(); // force initialization
            h = getProbe();
            wasUncontended = true;
        }
        // 是否扩容标记位,false 示意肯定不扩容,true 示意可能会扩容
        boolean collide = false;                
        for (;;) {Cell[] as; Cell a; int n; long v;
            // CASE-1: 以后 cell 数组曾经初始化
            if ((as = cells) != null && (n = as.length) > 0) {
            // CASE-1.1 以后线程对应的 cell 单元格为空
                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;
                }
                // CASE-1.2 以后线程对应的 cell 单元格 CAS 写入数据失败呈现竞争。else if (!wasUncontended)       
                    wasUncontended = true;      // Continue after rehash
                // CASE-1.3 以后线程对应的 cell 单元格 CAS 写入失败之后重试,如果胜利则跳出以后循环
                else if (a.cas(v = a.value, ((fn == null) ? v + x :
                                             fn.applyAsLong(v, x))))
                    break;
                // CASE-1.4 如果以后 cell 长度曾经超过以后机器的 CPU 数量,回绝扩容
                else if (n >= NCPU || cells != as)
                    collide = false;            
                else if (!collide)
                    collide = true;
                // 对 cell 数组进行扩容
                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);
            }
// CASE-2 cell 数组进行初始化
            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;
            }
// CASE-3 但 cell 数组为空,并且获取 cellBusy 失败再次进行重试 CAS 写入 base
            else if (casBase(v = base, ((fn == null) ? v + x :
                                        fn.applyAsLong(v, x))))
                break;                          // Fall back on using base
        }
    }

场景剖析

1,呈现多线程竞争,就是多个线程写入 base

就是 add 办法的判断条件 !casBase(b = base, b + x) 为 false,取反之后为 true,进入下一个 if

1.1 此时 cell 数组没有进行初始化

就是 as == null || (m = as.length - 1) < 0 为 true,进入 longAccumulate 办法的

else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
/*cellBusy 为 0 示意没有线程获取到以后 cell 的锁,通过 CAS 将 cellBusy 更新为 1 获取到锁开始初始化 cell 数组
init cell 数组是否初始化实现的标记位
*/
                boolean init = false;
                try {
/*
再次判断以后 cell 数组是否为后面的 as 援用。这里会呈现一种状况,就是线程 1 执行到 cellBusy= 0 之后因为系统调度的起因让出 CPU 工夫,这时线程 2 更新 cellBusy 胜利获取到锁并初始化 cell 数组胜利并开释锁,将 cellBusy 更新为 0。而后线程 1 失去 CPU 工夫开始更新 cellBusy 值获取锁胜利,但此时 cell 数组曾经初始化实现,此时线程 1 再次进行初始化会笼罩掉线程 2 曾经初始化的 cell 数组。其余中央的判断都是避免系统调度起因避免线程再次操作而笼罩。*/
                    if (cells == as) {
                    // 初始化一个长度为 2 的数组
                        Cell[] rs = new Cell[2];
                    // 以后线程的 hash 值和 1 进行 & 运算,后果只会是 0 和 1 所以这个线程只会定位到 cell[0]或者 cell[1]的单元格
                        rs[h & 1] = new Cell(x);
                        cells = rs;
                        // init 为 true 示意 cell 数组初始化实现
                        init = true;
                    }
                } finally {
                // 更新 cellBusy 为 0,开释锁
                    cellsBusy = 0;
                }
                // cell 数组初始化实现,跳出以后循环,开始下一轮循环
                if (init)
                    break;
            }

如果此时 cellBusy 为 0 然而通过 CAS 将 cellBusy 为 1 失败,阐明曾经有线程在初始化 cell 数组,那么就进入下一个判断

else if (casBase(v = base, ((fn == null) ? v + x :
                                        fn.applyAsLong(v, x))))
// 再次尝试写入 base,如果此时写入胜利,就跳出,否则进入下一个循环。
1.2 以后线程定位的 cell 单元格不存在为空(a = as[getProbe() & m]) == null

getProbe()办法获取的就是以后线程的 hash 值。这个判断条件满足一个前提就是 add 办法的 (as = cells) != null 为 true。此时进入 longAccumulate 的

/*
老规矩 二次判断
*/
            if ((as = cells) != null && (n = as.length) > 0) {
            // 二次判断
                if ((a = as[(n - 1) & h]) == null) {
                // 以后 cellBusy 为 0 没有线程批改 cell 数组
                    if (cellsBusy == 0) {
                    // 创立一个 cell 对象       
                        Cell r = new Cell(x);
                        // 获取以后 cell 数组的锁   
                        if (cellsBusy == 0 && casCellsBusy()) {
                        // cell 对象是否创立实现的标记位
                            boolean created = false;
                            try {
                            /*
                            再次判断,老样子,避免系统调度起因呈现线程的二次批改
                            */             
                                Cell[] rs; int m, j;
                                if ((rs = cells) != null &&
                                    (m = rs.length) > 0 &&
                                    rs[j = (m - 1) & h] == null) {rs[j] = r;
                                    // cell 单元格创立实现,更新 created 为 true
                                    created = true;
                                }
                            } finally {
                            // 开释锁
                                cellsBusy = 0;
                            }
                            // 跳出循环,走上面的 else if
                            if (created)
                                break;
                            continue;           
                        }
                    }
                    collide = false;
                }
                h = advanceProbe(h);
            }
1.3 以后线程对应的 cell 单元格呈现了竞争,多个线程通过 hash 之后都定位到同一个 cell 单元格,对应 add 办法的
!(uncontended = a.cas(v = a.value, v + x))

此时进入 longAccumulate 办法的

 else if (!wasUncontended)       
                    wasUncontended = true;   
                else if (a.cas(v = a.value, ((fn == null) ? v + x :
                                             fn.applyAsLong(v, x))))
                    break;
                 // 如果 cell 数组长度大于 CPU 数量,进行扩容
                else if (n >= NCPU || cells != as)
                    collide = false;            
                else if (!collide)
                    collide = true;
                // 获取到 cell 数组的锁,开始执行数组扩容
                else if (cellsBusy == 0 && casCellsBusy()) {
                    try {
                   // 老样子 二次判断
                        if (cells == as) {   
                        // cell 数组扩容为原来 2 倍  
                            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;                   
                }
1.4 LongAdder 的 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 数组的 value 并和 base 进行求和,最终失去 sum 值。

总结

通过下面的源码剖析咱们能够看到,LongAdder 通过每个线程对应本人的 cell 单元格来升高高并发下的竞争写问题。同时 cell 数组长度一旦大于 CPU 数量就进行扩容来最大水平应用并发数,因为一台机器真正并发的线程就等于 CPU 的数量,进而进步性能。
将对单个 base 的并发写散列开,每个线程对应一个 cell 单元格来升高并发。这个解决办法咱们在当前的高并发场景下能够学习借鉴。

退出移动版