Java中的并发计数器LongAdder

深碍√TFBOYSˉ_ 2022-09-02 01:10 315阅读 0赞

### 并发计数器LongAdder ###

*  基本概念
 *  Striped64
 *   *  Striped64重要成员变量
     *   *  cells
         *  cellsBusy
         *  base
     *  内部类Cell
     *   *  CPU缓存架构
         *  缓存行cache line
         *  伪共享
     *  Striped64重要方法
     *   *  longAccumulate
 *  LongAdder
 *   *  LongAdder重要方法
     *   *  add
 *  总结
 *   *  hash的值生成
     *  threadLocalRandomProbe初始化
     *  threadLocalRandomProbe重新生成

# 基本概念 #

*  位于**java.util.concurrent.atomic**包.是高并发下计数功能最好的数据结构,低并发下效率也非常高
 *  **LongAdder**和**DoubleAdder**数据结构类似

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pld2F2ZU94Zm9yZA_size_16_color_FFFFFF_t_70_pic_center]

*  **LongAdder**继承**Striped64**类,实现累加功能

# Striped64 #

*  **Striped64**是一个高并发累加的工具类
 *  **Striped64设计的核心思路:** 通过内部的分散计算来实现避免竞争
    
     *  **Striped64**中包含一个**base**和一个**cells**数组
     *  在没有竞争的情况下,累加的数通过**CAS**累加到**base**上
     *  在竞争的情况下,累加的数累加到数组**cells**中的某个**cell**元素上
     *  这样整个**Striped64**的值为:  s u m = b a s e + ∑ 0 n c e l l sum=base+\\sum\_0^ncell sum=base\+∑0ncell

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pld2F2ZU94Zm9yZA_size_16_color_FFFFFF_t_70_pic_center 1]

## Striped64重要成员变量 ##

*  **Striped64**中有三个重要的成员变量

/** * 存放元素cell的Hash表，大小为2的整数次幂 */
        transient volatile Cell[] cells;
    
        /** * 基础值: * - 在没有竞争的情况下，累加的数通过CAS累加到base上 * - 在数组cells初始化过程时，数组中的元素cell不可用。此时累加的数会尝试通过CAS累加到base上 */
        transient volatile long base;
    
        /** * 自旋锁，通过CAS加锁 * 用于创建和扩容数组cells的Hash表 */
        transient volatile int cellsBusy;

### cells ###

*  数组**cells**是**LongAdder**实现高性能的关键
 *  在**AtomicLong**中只有一个计数的值**value,** 所有线程的累加操作都要通过**CAS**锁竞争变量**value**来实现累加,处于高并发情况下,线程的竞争非常高
 *  **LongAdder**中有两个计数的值:
    
     *  **base:**
        
         *  在没有竞争或者初始化的情况下使用
         *  作用和**AtomicLong**中的**value**变量类似
     *  **cells:**
        
         *  在没有竞争的情况下,不会使用**cells**数组,值为**null**
         *  在竞争的情况下,就使用到**cells**数组:
            
             *  初始化时长度为**2**
             *  每次扩容长度都增加**1**倍,长度变为原来的**2**倍
             *  直到数组**cells**的长度大于等于当前服务器**CPU**的数量就不再扩容
             *  每个线程通过对**cells\[threadLocalRandomProbe%cells.length\]** 位置上的**cell**元素中的**value**值做累加,这样相当于将线程绑定到数组**cells**中的某个**cell**元素对象上

### cellsBusy ###

*  **cellsBusy**是自旋锁,通过**CAS**加锁.用于创建和扩展数组**cells**的**Hash**表
 *  **cellsBusy**的值包括无锁**0**和加锁**1,** 作用是修改数组**cells**时进行加锁,防止多线程同时修改**cells**数组
 *  **cellsBusy**加锁的情况有三种:
    
     *  数组**cells**初始化
     *  数组**cells**扩容
     *  数组**cells**中某个位置的为**null,** 在这个位置创建新的**Cell**对象

### base ###

*  **base**主要有两个作用:
    
     *  在没有竞争的情况下,将累加的数通过**CAS**累加到**base**上
     *  在数组**cells**初始化过程时,数组中的元素**cell**不可用.此时累加的数会尝试通过**CAS**累加到**base**上

## 内部类Cell ##

*  内部类**Cell**的类的注解是 **@jdk.internal.vm.annotation.Contended**
 *  内部类**Cell**是一个静态**final**类型的内部类,内部有一个**value**值,使用**CAS**来更新**value**的值

### CPU缓存架构 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pld2F2ZU94Zm9yZA_size_16_color_FFFFFF_t_70_pic_center 2]

*  缓存越靠近**CPU,** 缓存运行速度就越快,容量也越小:
    
     *  一级缓存**L1**紧靠着**CPU**内核,只能被一个单独的**CPU**内核使用. 尽管容量很小,但是运行速度很快
     *  二级缓存**L2**容量大一些运行速度也慢一些,只能被一个单独的**CPU**内核使用
     *  三级缓存**L3**容量更大一些运行速度也更慢一些,可以被单个插槽上的所有**CPU**内核共享
     *  主内存中保存着应用运行的所有数据,也是容量最大,运行速度最慢的,可以被全部插槽上的所有**CPU**内核共享
 *  **CPU**执行运算时,会首先从一级缓存**L1**中寻找数据,如果没有再去二级缓存**L2**中寻找数据,如果没有再去三级缓存**L3**中寻找数据.如果这些缓存中都没有数据,就去主内存中寻找数据
 *  因为**CPU**执行运算,寻找数据走得越远,执行耗费的时间就越长.所以对于一些需要频繁运算的数据,需要确保在一级缓存**L1**中

### 缓存行cache line ###

*  **CPU**的缓存系统是以缓存行**cache line**为单位存储的:
    
     *  缓存行**cache line**是**2**的整数幂次方个连续字节.通常情况下为**32~256**个字节
     *  常见的缓存行**cache line**的大小为**64**个字节
     *  缓存行**cache line**是缓存**cache**和内存**memory**之间数据传输的最小单元
 *  **Java**中一个**long**类型的长度是**8**个字节,通常一个缓存行中可以存放**8**个**long**类型的变量

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pld2F2ZU94Zm9yZA_size_16_color_FFFFFF_t_70_pic_center 3]

*  程序运行时,缓存每次更新都会从主内存中加载连续的**64**个字节.如果访问的是一个**long**类型的数组时,当数组中一个值被加载到缓存中时,同时另外**7**个元素的值也会被加载到缓存中
    
     *  如果数据结构中各项的值在内存中不是连续相邻的,比如链表.就无法获得这种同时加载数据到缓存中的这种好处
     *  这种同时加载存在缺点:
        
         *  如果一个**long**类型的变量**A,** 不是数组的一部分,而是一个单独的变量.并且有另外一个**long**类型的变量**B**紧挨着这个变量,那么当这个变量**A**被加载到缓存中时,另一个变量**B**也会被加载到缓存中
         *  此时如果一个**CPU**线程正在对这个变量**A**进行修改,另外一个**CPU**线程正在对另一个变量**B**进行读取
         *  对这个变量**A**修改的**CPU**线程修改变量**A**时,会将变量**A**和变量**B**同时加载到缓存行中,修改变量**A**后,因为其余缓存行中的变量**A**已经不是最新值,所以其余所有包含变量A的缓存行都将失效
         *  对另一个变量**B**读取的**CPU**线程读取变量**B**时,发现这个缓存行已经失效,就需要从主内存中重新加载数据到缓存行中
 *  **CPU**中的缓存都是以缓存行作为一个最小单位进行数据处理的,只要缓存行中有一个值失效,那么整个缓存行都会失效

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pld2F2ZU94Zm9yZA_size_16_color_FFFFFF_t_70_pic_center 4]

*  由此会导致伪共享问题: 变量**B**和变量**B**两个完全不相干的变量,每次变量**B**都要因为变量**A**的更新重主内存中重新读取变量**B,** 这样变量**B**就会因为变量**A**造成被缓存未命中而拖慢运行速度

### 伪共享 ###

*  **伪共享:** 当多线程中修改相互独立的变量时,如果这些变量共享同一个缓存行,就会无意中影响彼此的性能
    
     *  **CPU**中使用了**L1 cache**和**L2 cache:**
        
         *  **L1 cache**中数据处理方式是直写模式**write through**
            
             *  **write-through:** 直写模式. 数据更新时,同时写入缓存Cache和后端存储
                
                 *  **优点:** 操作简单
                 *  **缺点:** 数据修改后同时需要写入缓存和存储,写入的速度较慢
         *  **L2 cache**中数据处理方式是回写模式**write back**
            
             *  **write-back:** 回写模式. 数据更新时只写入缓存,只有在数据被替换出缓存时,修改的数据才会被写到后端存储中
                
                 *  **优点:** 因为不需要写存储,所以数据的写入速度很快
                 *  **缺点:** 如果更新后的数据没有被及时写入到存储中,此时系统出现异常,会导致修改后的数据无法找回
         *  **L1 cache**中的数据会及时写入缓存和存储中,而**L2 cache**中的数据只会写入缓存,而不会立即写回到内存存储中,这样就导致**cache**和**memory**中的数据不一致的情况
         *  对于多处理器**multiprocessors**的情况,由于**cache**是**CPU**内核私有的,不同的**CPU**内核的**cache**内容也会存在不一致的问题
         *  因此,在很多多处理器**multiprocessor**的计算机架构中,比如**ccnuma**和**smp**中,都实现了**cache coherence**机制,即不同的**CPU**内核的**cache**一致性机制
         *  **cache coherence:** 缓存一致性. 通过**cache-snooping**协议,每个**CPU**内核通过对总线**bus**的窥探**snoop**实现对**CPU**内核读写缓存**cache**的监控
            
             *  **某个CPU内核写缓存cache时:**
                
                 *  其余的**CPU**会检查自身的缓存**cahce**中的缓存行**cache line**
                 *  如果状态为**dirty(CPU**更新数据,写入缓存没有写入存储), 就将数据回写到存储中,并且将写缓存**cache**的**CPU**的相关**cache line**刷新
                 *  如果状态不是**dirty(CPU**更新数据,写入缓存没有写入存储),就将**cache line**使用**invalidate**使得缓存行失效
             *  **某个CPU内核读缓存cache时:**
                
                 *  其余的**CPU**会将自身的缓存**cache**中的缓存行**cache line**中标记为**dirty(CPU**更新数据,写入缓存没有写入存储)的部分的数据回写到存储中,并且将读缓存**cache**的**CPU**的相关**cache line**刷新
         *  提高**CPU**的缓存命中率**cache hit rate,** 减少**cache**和**memory**之间的数据传输,将会提高系统性能
     *  在程序和二进制对象的内存分配中保持缓存行对齐**cache line aligned**就十分重要. 否则会出现多个**CPU**内核中并行运行的进程或者线程同时读写同一个缓存行**cache line**的情况.这时候就会导致**CPU**内核的**cache**和**memory**之间反复出现**write back**和**refresh**的情况,这种情况称作**cache thrashing**
     *  **避免cache thrashing有以下两种途径:**
        
         *  对于**heap**的分配,在**malloc**调用中实现了强制的对齐**alignment**
         *  对于**stack**的分配,编译器提供了**stack aligned**的选项
            
             *  **注意:** 对于在编译器中指定**stack aligned,** 程序的会变得臃肿,会占用更多的内存,一定要做好取舍
 *  **伪共享的原理:** 一个缓存行通常使用**64**个字节,缓存每次更新时都会从主内存中加载连续的**64**个字节的数据,如果访问的是一个**8**个字节的**long**类型的数组时,当数组中一个值被加载到缓存中,同时另外**7**个相邻的**long**类型的数组也会被加载到缓存中,进而引发伪共享的问题
 *  **避免伪共享的几种方式:**
    
     *  **在两个原本相邻的不相干的long类型的变量之间添加7个额外的long类型的变量**
     *  **自定义long类型而不是使用Java自带的long类型**
        
         *  使得自定义的long类型数据可以占满64个字节
         *  通常也就是单独的long类型变量中添加7个额外的long类型的变量
     *  **使用@Contended注解.** 推荐使用这种方式
        
         *  如果使用这个注解无效的话,需要配置JVM启动参数后重新运行
        
            -XX:-RestrictContended
 *  **Java**中避免伪共享的示例:
    
     *  **ConcurrentHashMap**
        
         *  **ConcurrentHashMap**中的**size()** 方法是使用分段的思想来构造的
         *  每个段使用的类是**CounterCell**
         *  类**CounterCell**上就使用了 **@Contended**的注解
     *  **LongAdder**

## Striped64重要方法 ##

### longAccumulate ###

/** * 处理涉及初始化，扩容，创建新的Cell对象和竞争的更新情况 * * @param x 需要累加的值 * @param fn 更新函数，可以为null。用于累加操作 * @param wasUncontended 如果CAS操作在调用前执行失败则返回false。表明调用方法之前的add()方法是否没有发生争用 */
    final void longAccumulate(long x, LongBinaryOperator fn, boolean wasUncontended);

*  [longAccumulate()方法源码解析==>\[如果点进来的话, 顺手给个follow哈! 大佬的鼓励支持是小可持续更新的动力!!!\]][longAccumulate_ _follow_]

# LongAdder #

## LongAdder重要方法 ##

### add ###

/** * 对指定的值累加 * * @param x 需要累加的值 */
    public void add(long x);

*  [add()方法源码解析==>\[如果点进来的话, 顺手给个follow哈! 大佬的鼓励支持是小可持续更新的动力!!!\]][add_ _follow_]

# 总结 #

## hash的值生成 ##

*  **hash的值:** **LongAdder**中用来定位当前线程应该将值累加到数组**cells**的哪一个索引位置上
 *  **Thread**类中的成员变量**threadLocalRandomProbe:**

@jdk.internal.vm.annotation.Contended("tlr")
        int threadLocalRandomProbe;

*  **threadLocalRandomProbe**变量的值就是**LongAdder**用来确定**hash**的值来定位数组**cells**中的位置的 **.threadLocalRandomProbe**的初始值为**0**
 *  在**LongAdder**的父类**Striped64**中,通过**getProbe()** 方法获取当前线程的**threadLocalRandomProbe**的值:

/** * 返回当前线程的probe值 * 由于包装的限制，从ThreadLocalRandom中复制 */
    	static final int getProbe() { 
            return (int) THREAD_PROBE.get(Thread.currentThread());
        }

*  **probe**值是**threadLocalRandomProbe**变量在**Thread**类中的偏移量

## threadLocalRandomProbe初始化 ##

if ((h = getProbe()) == 0) { 
                // ThreadLocalRandom类强制初始化
                ThreadLocalRandom.current(); // force initialization
                // 设置h的值为0x9e3779b9
                h = getProbe();
                // 将没有争用的标识设置为true
                wasUncontended = true;
            }

*  当前线程使用**LongAdder**中的**add()** 方法执行累加操作,在没有进入**longAccumulate()** 方法之前 **,threadLocalRandomProbe**的值一直为**0**
 *  当执行累加操作发生争用后进入**longAccumulate()** 方法,第一次进入**longAccumulate()** 方法就判断**threadLocalRandomProbe**的值是否为**0,** 如果值为**0,** 则将**hash**的值设置为**0x9e3779b9**
 *  **ThreadLocalRandom.current()** 方法:

/** * 返回当前线程的ThreadLocalRandom */
    	public static ThreadLocalRandom current() { 
            if (UNSAFE.getInt(Thread.currentThread(), PROBE) == 0)
                localInit();
            return instance;
        }

*  在**current()** 方法中,判断**probe**的值是否为**0,** 如果**probe**的值为**0,** 就执行**localInit()** 方法,将当前的**probe**设置为非**0**的值:

/** * 为当前线程初始化线程字段 */
    	static final void localInit() { 
            int p = probeGenerator.addAndGet(PROBE_INCREMENT);
            // 跳过p为0的值
            int probe = (p == 0) ? 1 : p; // skip 0
            long seed = mix64(seeder.getAndAdd(SEEDER_INCREMENT));
            // 获取当前线程
            Thread t = Thread.currentThread();
            U.putLong(t, SEED, seed);
            // 将probe的值更新为probeGenerator的值
            U.putInt(t, PROBE, probe);
        }

*  **probeGenerator**是**static**类型的**AtomicInteger**类的对象,每执行一次**localInit()** 方法,都会将**probeGenerator**累加一次值**0x9e3779b9**
 *  **0x9e3779b9**是由**232** 除以黄金比例常数**1.6180339887**确定的
 *  如果**probeGenerator**值为**0,** 就将当前线程的**threadLocalRandomProbe**的值设置为**1.** 否则就将当前线程的**threadLocalRandomProbe**的值设置为**probeGenerator**的值

## threadLocalRandomProbe重新生成 ##

*  将**probe**的值进行**3**次左右移位和异或操作:

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

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pld2F2ZU94Zm9yZA_size_16_color_FFFFFF_t_70_pic_center]: /images/20220829/40635b72781c40289eb8d9530f4ea8e4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pld2F2ZU94Zm9yZA_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20220829/de6889a88543440f9afb2cd2aa58bcfa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pld2F2ZU94Zm9yZA_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20220829/43c512c93f99455ea047a44ea7440947.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pld2F2ZU94Zm9yZA_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20220829/3ce0c499d9ac4fb69b15f65a5ffc5e03.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pld2F2ZU94Zm9yZA_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20220829/03c2c4268a304997838a56eb37e73d45.png
[longAccumulate_ _follow_]: https://github.com/framework-projects/java/blob/master/src/main/java/java.base/java/util/concurrent/atomic/Striped64.java
[add_ _follow_]: https://github.com/framework-projects/java/blob/master/src/main/java/java.base/java/util/concurrent/atomic/LongAdder.java