37.并发编程之LongAdder介绍

Leefs 2022-11-09 PM 905℃ 0条

[TOC]

### 一、原子累加器对比

我们通过如下代码，比较`synchronized`、`AtomicInteger`、`AtomicLong`、`LongAdder`、`LongAccumulator`五种计数性能。

**示例**

> 需求：热点商品点赞计算器，点赞数进行统计，不要求实时精确。50个线程，每个线程100W次，统计总点赞数。

```java
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAccumulator;
import java.util.concurrent.atomic.LongAdder;

/**
 * @author lilinchao
 * @date 2022-11-09
 * @description 热点商品点赞数统计
 **/
class ClickNumber{
    int number = 0;
    public synchronized void add_synchronized(){
        number++;
    }

AtomicInteger atomicInteger = new AtomicInteger();
    public void add_AtomicInteger(){
        atomicInteger.incrementAndGet();
    }

AtomicLong atomicLong = new AtomicLong();
    public void add_AtomicLong(){
        atomicLong.incrementAndGet();
    }

LongAdder longAdder = new LongAdder();
    public void add_LongAdder(){
        longAdder.increment();
    }

LongAccumulator longAccumulator = new LongAccumulator((x, y)->x+y,0);
    public void add_longAccumulator(){
        longAccumulator.accumulate(1);
    }
}

public class LongAdderCalcDemo {
    public static final int SIZE_THREAD = 50;
    public static final int _1w = 10000;

public static void main(String[] args) throws InterruptedException {
        ClickNumber clickNumber = new ClickNumber();
        long startTime = System.currentTimeMillis();
        long endTime = System.currentTimeMillis();
        CountDownLatch latch_synchronized = new CountDownLatch(SIZE_THREAD);
        CountDownLatch latch_AtomicInteger= new CountDownLatch(SIZE_THREAD);
        CountDownLatch latch_AtomicLong = new CountDownLatch(SIZE_THREAD);
        CountDownLatch latch_LongAdder = new CountDownLatch(SIZE_THREAD);
        CountDownLatch latch_LongAccumulator = new CountDownLatch(SIZE_THREAD);

startTime = System.currentTimeMillis();
        for (int i = 1; i <= SIZE_THREAD; i++) {
            new Thread(()->{
                try {
                    for (int j = 1; j <= 100*_1w; j++) {
                        clickNumber.add_synchronized();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    latch_synchronized.countDown();
                }
            },String.valueOf(i)).start();
        }
        latch_synchronized.await();
        endTime = System.currentTimeMillis();
        System.out.println("synchronized花费时间："+ (endTime-startTime)+" 数值为："+clickNumber.number);

startTime = System.currentTimeMillis();
        for (int i = 1; i <= SIZE_THREAD; i++) {
            new Thread(()->{
                try {
                    for (int j = 1; j <= 100*_1w; j++) {
                        clickNumber.add_AtomicInteger();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    latch_AtomicInteger.countDown();
                }
            },String.valueOf(i)).start();
        }
        latch_AtomicInteger.await();
        endTime = System.currentTimeMillis();
        System.out.println("AtomicInteger花费时间："+ (endTime-startTime)+" 数值为："+clickNumber.atomicInteger.get());

startTime = System.currentTimeMillis();
        for (int i = 1; i <= SIZE_THREAD; i++) {
            new Thread(()->{
                try {
                    for (int j = 1; j <= 100*_1w; j++) {
                        clickNumber.add_AtomicLong();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    latch_AtomicLong.countDown();
                }
            },String.valueOf(i)).start();
        }
        latch_AtomicLong.await();
        endTime = System.currentTimeMillis();
        System.out.println("AtomicLong花费时间："+ (endTime-startTime)+" 数值为："+clickNumber.atomicLong.get());

startTime = System.currentTimeMillis();
        for (int i = 1; i <= SIZE_THREAD; i++) {
            new Thread(()->{
                try {
                    for (int j = 1; j <= 100*_1w; j++) {
                        clickNumber.add_LongAdder();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    latch_LongAdder.countDown();
                }
            },String.valueOf(i)).start();
        }
        latch_LongAdder.await();
        endTime = System.currentTimeMillis();
        System.out.println("LongAdder花费时间："+ (endTime-startTime)+" 数值为："+clickNumber.longAdder.longValue());

startTime = System.currentTimeMillis();
        for (int i = 1; i <= SIZE_THREAD; i++) {
            new Thread(()->{
                try {
                    for (int j = 1; j <= 100*_1w; j++) {
                        clickNumber.add_longAccumulator();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    latch_LongAccumulator.countDown();
                }

},String.valueOf(i)).start();
        }
        latch_LongAccumulator.await();
        endTime = System.currentTimeMillis();
        System.out.println("LongAccumulator花费时间："+ (endTime-startTime)+" 数值为："+clickNumber.longAccumulator.longValue());

}
}
```

**运行结果**

```
synchronized花费时间：900 数值为：50000000
AtomicInteger花费时间：585 数值为：50000000
AtomicLong花费时间：623 数值为：50000000
LongAdder花费时间：83 数值为：50000000
LongAccumulator花费时间：52 数值为：50000000
```

通过上面的代码运行我们可以清晰地看到，每种方式都有效的产完成了累加的效果，但是明显使用**LongAdder**的效率要更好，甚至要高出**AtomicLong**好几倍。

现在，我们可以简单地理解为，原子累加器就是JAVA在并发编程下提供的有保障的累加手段。

那么`LongAdder`是如何做的这一点的呢？接下来我们慢慢进行分析。

### 二、LongAdder概述

JDK1.8时，`java.util.concurrent.atomic`包中提供了一个新的原子类：`LongAdder`。

根据Oracle官方文档的介绍，`LongAdder`在高并发的场景下会比它的前辈`AtomicLong` 具有更好的性能，代价是消耗更多的内存空间。

+ **AtomicLong存在的问题**

在并发量较低的环境下，线程冲突的概率比较小，自旋的次数不会很多。但是，高并发环境下，N个线程同时进行自旋操作，会出现大量失败并不断自旋的情况，此时`AtomicLong`的自旋会成为瓶颈。

这就是`LongAdder`引入的初衷,解决高并发环境下`AtomicLong`的自旋瓶颈问题。

+ **使用场景**

在大数据处理过程，为了方便监控，需要统计数据，少不了原子计数器。为了尽量优化性能，需要采用高效的原子计数器。
		 在jdk8中，引入了`LongAdder`，非常适合多线程原子计数器。与`AtomicLong`做了一个测试，`LongAdder`在多线程环境中，原子自增长性能要好很多。它常用于状态采集、统计等场景。

### 三、LongAdder原理

#### 3.1 AtomicLong和LongAdder比较

+ **AtomicLong**

![37.并发编程之LongAdder介绍01.jpg](https://lilinchao.com/usr/uploads/2022/11/2268135403.jpg)

`AtomicLong`中有个内部变量value保存着实际的long值，所有的操作都是针对该变量进行。也就是说，高并发环境下，value变量其实是一个热点，也就是N个线程竞争一个热点。

+ **LongAdder**

![37.并发编程之LongAdder介绍02.jpg](https://lilinchao.com/usr/uploads/2022/11/1061699010.jpg)

`LongAdder`的基本思路就是分散热点，将value值分散到一个数组中，不同线程会命中到数组的不同槽中，各个线程只对自己槽中的那个值进行CAS操作，这样热点就被分散了，冲突的概率就小很多。如果要获取真正的long值，只要将各个槽中的变量值累加返回。

`AtomicLong`是多个线程针对单个热点值value进行原子操作。而`LongAdder`是每个线程拥有自己的槽，各个线程一般只对自己槽中的那个值进行CAS操作。

#### 3.2 原理

![37.并发编程之LongAdder介绍03.jpg](https://lilinchao.com/usr/uploads/2022/11/3416894313.jpg)

`LongAdder` 在无竞争条件下，跟 `AtomicInteger` 一样，对同一个 base 进行操作。

当出现了多线程竞争时采用化整为零的做法，用空间换时间，用一个数组`cells[]`，将一个 value 拆分进这个 `cells[]` 数组，多线程同时对各个 cell 中的 value 进行操作，可以对线程 id 进行 hash，再根据 hash 映射到这个数组的某个索引，然后对 cell 中的 value 进行 cas 操作，当所有线程都执行完毕，累加所有 cell 中的value 值 和 无竞争下的 base 即为最终结果。
总结一句公式： **LongAdder = base基础数据 + cell[] + CAS** 操作，用空间换时间所以快。

### 四、基类Striped64内部三个重要成员

`LongAdder` 继承于 `Stripped64` 类，`base` 值和 `cell` 数组都在 `Stripped64` 类中定义，它的内部三个重要的成员如下：

```java
/**
 * 成员一：存放Cell的哈希表，大小为2的幂
 */
transient volatile Cell[] cells;
/**
 * 成员二：基础值
 * 1.在没有竞争时会更新这个值
 * 2.在cells初始化时，cells不可用，也会尝试通过CAS操作值累加到base
 */
transient volatile long base;
/**
 * 自旋锁，通过CAS操作加锁，为0表示cells数组没有处于创建、扩容阶段
 * 为1表示正在创建或者扩展cells数组，不能进行新Cell元素的设置操作
 */
transient volatile int cellsBusy;
```

`Stripped64` 内部包含一个 `base` 和一个 `Cell[]` 类型的 `cells` 数组，`cells` 数组又叫哈希表，在没有竞争的情况下，要累加的数通过CAS累加到 `base` 上，如果有竞争的话，会将要累加的数累加到 `cells` 数组中的某个 `Cell` 元素里面。

`Stripped64` 的整体值 `value` 的获取函数如下：

```java
public long longValue() {
    return sum();
}

/**
 * 将多个cells数组中的值加起来的和
 */
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;
}
```

`Stripped64` 的设计核心思路是通过内部的分散计算来避免竞争，以空间换取时间。没有竞争时 `cells` 数组为 null，这时只使用 `base`，一旦发生竞争，`cells` 数组就上场了。

`cells` 数组第一次初始化长度为2，以后每次扩容都变为原来的两倍，一直到 `cells` 数组的长度大于等于当前服务器的CPU核数，同一时刻能持有CPU时间片去并发操作同一内存地址的最大线程数最多也就是CPU的核数。

在存在线程争用的时候，每个线程被映射到 `cells[threadLocalRandomProbe&cells.length]` 位置的 Cell 元素，该线程对 `value` 所做的累加操作就执行在对应的 `Cell` 元素的值上。

*附参考文章链接*

*https://www.jianshu.com/p/30dfa17a8521*

*https://www.cnblogs.com/54chensongxia/p/12191042.html*

标签: 并发编程

非特殊说明，本博所有文章均为博主原创。

如若转载，请注明出处：https://lilinchao.com/archives/2558.html

上一篇 36.并发编程之字段更新器

下一篇【转载】38.Java双刃剑之Unsafe类详解

37.并发编程之LongAdder介绍

评论已关闭

栏目分类

标签云

友情链接申请

37.并发编程之LongAdder介绍

评论已关闭

 栏目分类

标签云

友情链接申请

栏目分类

标签云

友情链接申请