LongAdder的源码学习与理解

简介: LongAdder有了AtomicLong为什么还要LongAdderLongAdder中的主要方法
// 累加单元组,懒惰初始化
transient volatile Cell[] cells;
// 基础值,如果没有竞争,则用cas累加这个值
transient volatile long base;
//在cells创建或者扩容时,置为1,表示加锁
transient volatile int cellsBusy;

cellsBusy作用是当要修改cells数组时加锁,防止多线程同时修改cells数组,0为无锁,1为加锁,加锁的情况有三种

1、cells数组初始化的时候

2、cells数组扩容的时候

3、如果cells数组中某个元素为null,给这个位置创建新的Cell对象的时候


base有两个作用

在开始没有竞争的时候,将累加值到base

在cells初始化的过程中,cells不可用,这时候会尝试将值累加到base上

649e4b0ad8434ca3af2d19e411dde0e5.png

@sun.misc.Contended注解是为了防止缓存行伪共享

因为CPU与内存之间速度还是存在较大差距所以现在计算机在内存与CPU之间引入了三级缓存

3a5494726c3b44cfb0547566577babd7.png

L1与L2是每个CPU独享的,L3是所有CPU共享的

f55d37ad8ca644e2af6876f6af962baa.png

因为CPU与内存的速度差异很大,需要靠预读数据至缓存来提升效率

而缓存以缓存行为单位,每个缓存行对应着一块内存,一般是64byte

缓存的加入会造成数据副本的产生,即同一份数据会缓存在不同核心的缓存行中

CPU要保证数据的一致性,如果某个CPU核心更改了数据,其他CPU核心对应的整个缓存行必须失效。

e1264fdf42224c20a9038141c44fefb5.png

因为cell是数组形式,在内存中是连续存储的,一个Cell为24个字节(16字节的对象头和8字节的value),因此缓存行可以存下2个Cell对象,所以问题来了:

Core-0要修改Cell[0]

Core-1要修改Cell[1]

无论谁修改成功,都会导致对方Core的缓存行失效

@sun.misc.Contended注解就是用来解决这个问题,它的原理是在适用此注解的对象或字段的前后各增加128字节大小的padding,从而让CPU将对象预读至缓存时占用不同的缓存行,这样就不会造成对方的缓存行失效


有了AtomicLong为什么还要LongAdder


虽然AtomicLong使用CAS算法,但是CAS失败后还是通过无限循环的自旋锁不断的尝试,在高并发下CAS性能低下的原因所在

    public final int getAndAddInt(Object o, long offset, int delta) {
        int v;
        do {
            v = getIntVolatile(o, offset);
        } while (!compareAndSwapInt(o, offset, v, v + delta));
        return v;
    }

高并发下N多线程同时去操作一个变量会造成大量CAS失败,然后处于自选状态,导致严重浪费CPU资源,降低了并发性。既然AtomicLong性能问题是由于多线程同时去竞争同一个变量的更新而降低,那么把一个变量分解为多个变量,让同样多的线程去竞争多个资源

9de930f3e8fe4dc5bf5d5c5a08c2dd33.png386c35635d014e8da079f5f858c254ee.png

LongAdder在内部维护了一个Cells数组,每个Cell里面有一个初始值为0的long型变量,在同等并发量的情况下,争夺单个变量的线程会减少,这是变相减少了争夺共享资源的并发量,另外多个线程在争夺同一个原子变量时候,如果失败不是自选CAS重试而是尝试获取其他原子变量的锁,最后当获取当前值的时候把所有变量的值累加后再加上base的值返回

Cells占用内存相对比较大的所以一开始并不创建,而是在需要时候再创建,也就是惰性加载,当一开始没有空间的时候,所有的更新都是操作base变量


LongAdder中的主要方法


add方法

    public void add(long x) {
        Cell[] as; long b, v; int m; Cell a;
        /**
        * casBase 就是使用CAS来进行更改值的
        * 只有两种情况才会执行if内的语句
        * 1、cells数组不为空的时候(cells数组为空时候不存在竞争,所以直接操作caseBase,当不为空的时候就存在多个线程来竞争)
        * 2、cells为空,casBase执行失败的时候(casBase执行成功,则直接返回,如果casBase失败,说明第一次争用冲突产生,需要对cells数组初始化进入if)
        */
        if ((as = cells) != null || !casBase(b = base, b + x)) {
            boolean uncontended = true;
            /**
            * as == null :cells数组没有初始化,成立就进入if执行cell初始化
            * (m = as.length - 1) < 0 :cells数组的长度为0
            * 上面两个条件都代表cells数组没有被初始化成功
            * (a = as[getProbe() & m]) == null :说明当前线程获取的cells数组的这个位置的cell没有做过累加,所以需要创建一个cell对象
            * !(uncontended = a.cas(v = a.value, v + x)) :尝试对这个位置的cell进行累加并返回结果,如果累加失败就找另外一个cell
      *
      * 进入longAccumulate方法有三种情况
      * 1、cells没有初始化
      * 2、当前线程获取cell[i]的地方为空,需要创建一个cell对象
      * 3、当前对cell[i]cas加值失败
            */
            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);
        }
    }

longAccumulate方法

 final void longAccumulate(long x, LongBinaryOperator fn,boolean wasUncontended) {
        int h;
        /**
        * 
        */
        if ((h = getProbe()) == 0) {
            ThreadLocalRandom.current(); // force initialization
            h = getProbe();
            wasUncontended = true;
        }
        // cas冲突标志位
        boolean collide = false;                // True if last slot nonempty
        for (;;) {
            Cell[] as; Cell a; int n; long v;
            /**
            * 有三个分支
            * 主分支一:cells初始化好(处理add方法中的第3,4个条件)
            * 主分支二:cells数组没有初始化或者长度为0的情况(这个分支处理add方法的第1,2个条件)
            * 主分支三:cells数组正在被其他线程初始化则尝试将累加值通过cas累加到base上
            */
            if ((as = cells) != null && (n = as.length) > 0) {
              /**
              * 小分支一:如果当前cess[i]的位置为空处理的是第三个条件
              */
                if ((a = as[(n - 1) & h]) == null) {
                  // 代表没有其他线程修改
                    if (cellsBusy == 0) {       // Try to attach new Cell
                        // 创建一个cell对象
                        Cell r = new Cell(x);   // Optimistically create
                        // 如果没有其它线程修改通过cas将cellBusy设置为1
                        if (cellsBusy == 0 && casCellsBusy()) {
                          //标记create是否创建成功并放入cells数组被hash的位置上
                            boolean created = false;
                            try {               // Recheck under lock
                                Cell[] rs; int m, j;
                                //再次检查cells数组不为空并且长度大于0
                                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;
                            //失败说明被其它线程赋值了cells[i]位置
                            continue;           // Slot is now non-empty
                        }
                    }
                    collide = false;
                }
                /**
                * 小分支二:如果add方法中条件4通过cas加cell[i]失败则重新设置为true去找另一个cell
                */
                else if (!wasUncontended)       // CAS already known to fail
                    wasUncontended = true;      // Continue after rehash
                /**
                * 小分支三:给cell[i]加值如果成功则直接退出
                */
                else if (a.cas(v = a.value, ((fn == null) ? v + x :fn.applyAsLong(v, x))))
                    break;
                /**
                * 小分支四:如果cells发生了扩容或者当前cells数组长度大于了CPU的数量就存在冲突
                */
                else if (n >= NCPU || cells != as)
                    collide = false;            // At max size or stale
                /**
                * 小分支五:如果发生了冲突设置为true再次hash
                */
                else if (!collide)
                    collide = true;
        /**
        * 小分支六: 扩容
        */
                else if (cellsBusy == 0 && casCellsBusy()) {
                    try {
                      // 检查cells是否已经扩容
                        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
                }
                // 重新计算hash
                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;
            }
            /**
            * 主分支三:如果别人正在初始化就尝试CAS加base
            */
            else if (casBase(v = base, ((fn == null) ? v + x :fn.applyAsLong(v, x))))
                break;                          // Fall back on using base
        }
    }

advanceProbe

    static final int advanceProbe(int probe) {
        probe ^= probe << 13;   // xorshift
        probe ^= probe >>> 17;
        probe ^= probe << 5;
        UNSAFE.putInt(Thread.currentThread(), PROBE, probe);
        return probe;
    }

怎么确保Probe是当前线程的

在Thread类中定义了三个变量

    @sun.misc.Contended("tlr")
    long threadLocalRandomSeed;
    /** Probe hash value; nonzero if threadLocalRandomSeed initialized */
    @sun.misc.Contended("tlr")
    int threadLocalRandomProbe;
    /** Secondary seed isolated from public ThreadLocalRandom sequence */
    @sun.misc.Contended("tlr")
    int threadLocalRandomSecondarySeed;
相关文章
|
5月前
|
存储 并行计算 算法
深入解析Java并发库(JUC)中的Phaser:原理、应用与源码分析
深入解析Java并发库(JUC)中的Phaser:原理、应用与源码分析
|
5月前
|
算法 Java 容器
深入解析Java并发库(JUC)中的LongAdder
深入解析Java并发库(JUC)中的LongAdder
|
Java uml
JUC系列学习(六):ReentrantReadWriteLock的使用及源码解析
`ReentrantReadWriteLock`是一种读写锁,跟`ReentrantLock`一样也是实现了`Lock`,区别在于`ReentrantLock`是独占锁,同一时刻只能有一个线程持有锁,`ReentrantLock`在某些场景下可能会有并发性能的问题。而**ReentrantReadWriteLock是独占锁(写锁)、共享锁(读锁)可以同时存在的一种读写锁,在读操作远大于写操作的场景中,能实现更好的并发性**。当读锁存在时,其他线程仍然可以获取读锁并进行读操作,但是不能获得写锁进行写操作;当写锁存在时,其他线程的读锁、写锁都是不允许的。
|
Java 开发者
JUC系列学习(三):ReentrantLock的使用、源码解析及与Synchronized的异同
`ReentrantLock`同`Synchronized`一样可以实现线程锁的功能,同样具有可重入性,除此之外还可以实现公平锁&非公平锁,其底层是基于`AQS`框架实现的。
|
存储 安全 Java
java并发原理实战(7) -- 原子类使用和原理理解
java并发原理实战(7) -- 原子类使用和原理理解
126 0
java并发原理实战(7) -- 原子类使用和原理理解
|
存储 安全 Java
java并发原理实战(5)--线程安全性问题和synchronized原理理解
java并发原理实战(5)--线程安全性问题和synchronized原理理解
java并发原理实战(5)--线程安全性问题和synchronized原理理解
|
安全 Java C++
「死磕Java并发编程」说说Java Atomic 原子类的实现原理
「死磕Java并发编程」说说Java Atomic 原子类的实现原理
「死磕Java并发编程」说说Java Atomic 原子类的实现原理
|
安全 算法 Java
详解java并发原子类AtomicInteger(基于jdk1.8源码分析)
java并发包里面的类一直是学习和面试的重点,这篇文章主要是对java并发包的其中一个类AtomicInteger的讲解。从为什么要出现AtomicInteger再到其底层原理来一个分析。
334 0
详解java并发原子类AtomicInteger(基于jdk1.8源码分析)
Juc16_LongAdder引入、原理、Striped64、分散热点思想、深度解析LongAdder源码、LongAdder和AtomicLong区别(二)
③. LongAdder为什么这么快呢?(分散热点) ④. 源码解析 longAdder.increment( ) ①. add(1L)
166 0
Juc16_LongAdder引入、原理、Striped64、分散热点思想、深度解析LongAdder源码、LongAdder和AtomicLong区别(二)