longAdder源码解析

参考文章

• https://www.jianshu.com/p/d9d4be67aa56
• https://www.cnblogs.com/dwj-ngu/p/14623349.html
• https://www.cnblogs.com/tong-yuan/p/LongAdder.html

Striped类

//存放Cell的hash表，大小为2的幂。 
transient volatile Cell[] cells;
/** 
  * 基础值，没有竞争时会使用这个值，同时作为初始化table竞争失败的一种方案 
  * 也就是说没有竞争的时候会使用这个值，如果初始化table竞争失败也会使用这个值 
  */   
transient volatile long base;

//通过cas实现的锁，0无锁，1获得锁
transient volatile int cellsBusy;
//可用cpu数量
static final int NCPU=Runtime.getRuntime().availableProcessors();
1
2
3
4
5
6
7
8
9
10
11
12

前言

LongAdder 是并发大师 @author Doug Lea （大哥李）的作品，设计的非常精巧

LongAdder 类有几个关键域

// 累加单元数组, 懒惰初始化
transient volatile Cell[] cells;
// 基础值, 如果没有竞争, 则用 cas 累加这个域
transient volatile long base;
// 在 cells 创建或扩容时, 置为 1, 表示加锁
transient volatile int cellsBusy;
1
2
3
4
5
6

CAS锁

我们可以使用 CAS实现锁的功能

package cn.knightzz.atomic.lock;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author 王天赐
 * @title: CasLock
 * @projectName hm-juc-codes
 * @description: 使用CAS实现锁
 * @website http://knightzz.cn/
 * @github https://github.com/knightzz1998
 * @create: 2022-08-08 20:32
 */
@Slf4j(topic = "CasLock")
public class CasLock {

    // 0 无锁
    // 1 有锁
    private static AtomicInteger state = new AtomicInteger(0);


    public static void lock() {

        log.debug("lock ... ");

        while (true) {
            // 如果其他线程先拿到锁, state 就会被修改为 1
            // 然后就会被一直循环阻塞
            if (state.compareAndSet(0, 1)) {
                break;
            }
        }
    }

    public static void unlock() {

        log.debug("unlock ... ");
        state.set(0);
    }


}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

基本思路 :

• state.compareAndSet(0, 1) 通过设置 AtomicInteger 类型的state 来实现加锁和解锁
• 如果其他线程修该了 state 的值, 那么 state.compareAndSet(0, 1) 就会返回false, 然后一直无限循环

原理之伪共享

Cell代码

@sun.misc.Contended 
static final class Cell {
        volatile long value;
        Cell(long x) { value = x; }
        final boolean cas(long cmp, long val) {
            return UNSAFE.compareAndSwapLong(this, valueOffset, cmp, val);
        }

        // Unsafe mechanics
        private static final sun.misc.Unsafe UNSAFE;
        private static final long valueOffset;
        static {
            try {
                UNSAFE = sun.misc.Unsafe.getUnsafe();
                Class<?> ak = Cell.class;
                valueOffset = UNSAFE.objectFieldOffset
                    (ak.getDeclaredField("value"));
            } catch (Exception e) {
                throw new Error(e);
            }
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

其中 Cell 即为累加单元

缓存与内存

从上面表格可以看到 CPU从寄存器读取数据的速度和从内存中读取数据的速度差异很大, 所以为了提高效率, 在CPU和

内存中间增加了缓存 , 所以在某些情况下就会出现缓存和内存数据不一致的情况出现!

• 而缓存以缓存行为单位，每个缓存行对应着一块内存，一般是 64 byte（8 个 long）
• 缓存的加入会造成数据副本的产生，即同一份数据会缓存在不同核心的缓存行中
• CPU 要保证数据的一致性，如果某个 CPU 核心更改了数据，其它 CPU 核心对应的整个缓存行必须失效

因为 Cell 是数组形式，在内存中是连续存储的，一个 Cell 为 24 字节（16 字节的对象头和 8 字节的 value），因

此缓存行可以存下 2 个的 Cell 对象。这样问题来了：

• Core-0 要修改 Cell[0]

• Core-1 要修改 Cell[1]

无论谁修改成功，都会导致对方 Core 的缓存行失效，比如 Core-0 中 Cell[0]=6000, Cell[1]=8000 要累加

Cell[0]=6001, Cell[1]=8000 ，这时会让 Core-1 的缓存行失效 , 这俩核心的缓存是放在同一个缓存行里面

@sun.misc.Contended 用来解决这个问题，它的原理是在使用此注解的对象或字段的前后各增加 128 字节大小的

padding，从而让 CPU 将对象预读至缓存时占用不同的缓存行，这样，不会造成对方缓存行的失效

累加方法

累加主要调用下面的方法

public void add(long x) {
    // as是Striped64中的cells属性
    // b是Striped64中的base属性
    // v是当前线程hash到的Cell中存储的值
    // m是cells的长度减1，hash时作为掩码使用
    // a是当前线程hash到的Cell
    Cell[] as; long b, v; int m; Cell a;
    // 条件1：cells不为空，说明出现过竞争，cells已经创建
    // 条件2：cas操作base失败，说明其它线程先一步修改了base，正在出现竞争
    if ((as = cells) != null || !casBase(b = base, b + x)) {
        // true表示当前竞争还不激烈
        // false表示竞争激烈，多个线程hash到同一个Cell，可能要扩容
        boolean uncontended = true;
        // 条件1：cells为空，说明正在出现竞争，上面是从条件2过来的
        // 条件2：应该不会出现
        // 条件3：当前线程所在的Cell为空，说明当前线程还没有更新过Cell，应初始化一个Cell
        // 条件4：更新当前线程所在的Cell失败，说明现在竞争很激烈，多个线程hash到了同一个Cell，应扩容
        if (as == null || (m = as.length - 1) < 0 ||
            // getProbe()方法返回的是线程中的threadLocalRandomProbe字段
            // 它是通过随机数生成的一个值，对于一个确定的线程这个值是固定的
            // 除非刻意修改它
            (a = as[getProbe() & m]) == null ||
            !(uncontended = a.cas(v = a.value, v + x)))
            // 调用Striped64中的方法处理
            longAccumulate(x, null, uncontended);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

逐行分析上面的代码 :

Cell[] as; long b, v; int m; Cell a;
1

• as 是 累加单元数组
• b 是基础值, x 是累加值

if ((as = cells) != null || !casBase(b = base, b + x)) {
}
1
2

transient volatile Cell[] cells;
1

回顾原子累加器性能提升的原因 :

性能提升的原因很简单，就是在有竞争时，设置多个累加单元，Therad-0 累加 Cell[0]，而 Thread-1 累加

Cell[1]… 最后将结果汇总。这样它们在累加时操作的不同的 Cell 变量，因此减少了 CAS 重试失败，从而提高性

能。

有竞争 : 多个线程调用累加器

然后我们再回到下面的代码 :

if ((as = cells) != null || !casBase(b = base, b + x)) {
}
1
2

很显然, 如果有竞争的情况下, 也就是说 cells != null , 就进入 if 的代码块

亦或是没有竞争的情况下, 直接调用 casBase(b = base, b + x) , b 相当于 expectValue , 用于与内存中的值进行对比, b + x 是累加后的值, 如果累加失败, 也会直接进入代码块

然后就是 if 代码块内的代码

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);
        }
1
2
3
4
5
6
7

• uncontended 表示 cell 没有竞争

• as == null || (m = as.length - 1) < 0 表示累加单元数组尚未创建

• (a = as[getProbe() & m]) == null || 当前线程的 cell 数组尚未创建, 注意啊 getProbe() & m 这个是获取当前线程对应的cell 在 cells 的index , 每个线程对应一个 Cell 对象 (a) , 存放在 Cells 数组中

• !(uncontended = a.cas(v = a.value, v + x))) cas 给当前线程的 cell 累加失败 uncontended=false ( a 为当前线程的 cell )

• longAccumulate(x, null, uncontended); 进入 cell 数组创建、cell 创建的流程

基本流程图如下 :

LongAdder源码总结

基本策略

LongAdder 的基本策略是 :

• 无竞争状态 : 多线程顺序调用累加方法时, 使用一个基础的 base 值来进行累加
• 有竞争状态 : 多个线程同时调用累加方法时, 让不同的线程更新不同的值cell, 然后把所有的值进行累加

无竞争状态 :

有竞争状态 :

Add方法

add源码 :

public void add(long x) {
    // as是Striped64中的cells属性
    // b是Striped64中的base属性
    // v是当前线程hash到的Cell中存储的值
    // m是cells的长度减1，hash时作为掩码使用
    // a是当前线程hash到的Cell
    Cell[] as; long b, v; int m; Cell a;
    // 条件1：cells不为空，说明出现过竞争，cells已经创建
    // 条件2：cas操作base失败，说明其它线程先一步修改了base，正在出现竞争
    if ((as = cells) != null || !casBase(b = base, b + x)) {
        // true表示当前竞争还不激烈
        // false表示竞争激烈，多个线程hash到同一个Cell，可能要扩容
        boolean uncontended = true;
        // 条件1：cells为空，说明正在出现竞争，上面是从条件2过来的
        // 条件2：应该不会出现
        // 条件3：当前线程所在的Cell为空，说明当前线程还没有更新过Cell，应初始化一个Cell
        // 条件4：更新当前线程所在的Cell失败，说明现在竞争很激烈，多个线程hash到了同一个Cell，应扩容
        if (as == null || (m = as.length - 1) < 0 ||
            // getProbe()方法返回的是线程中的threadLocalRandomProbe字段
            // 它是通过随机数生成的一个值，对于一个确定的线程这个值是固定的
            // 除非刻意修改它
            (a = as[getProbe() & m]) == null ||
            !(uncontended = a.cas(v = a.value, v + x)))
            // 调用Striped64中的方法处理
            longAccumulate(x, null, uncontended);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

基本思路 :

第一个 if 判断 : if ((as = cells) != null || !casBase(b = base, b + x))

1. 先判断cells 是否为空 , 如果为空, 说明多个线程还没有开始竞争, 那就直接在 base 上面累加(执行casBase)

   如果 cells 不为空, 说明已经出现竞争, cell数组已经创建了

2. 执行 casBase() 方法, 如果执行 casBase方法失败, 那就说明有多个线程同时执行了casBase() 方法

   此时说明有竞争 (有竞争就可以创建 cells数组, 然后每个线程创建一个 Cell 对象用于累加)。

以上两种 : cells 不为空, 或者执行 casBase失败都会进入第二个if判断

第二个if判断 : as == null , (m = as.length - 1) < 0 , (a = as[getProbe() & m]) == null , !(uncontended = a.cas(v = a.value, v + x) :

1. 先判断 cell 数组是否创建【条件1】, 或者创建了, 但是cells数组中没有Cell对象【条件2】
2. 通过hash的方法获取当前线程Cell对象【条件3】 , 如果当前对象为null, 说明还未创建Cell , 此时就初始化一个Cell对象
3. 更新当前的Cell对象的值(进行累加)【条件4】，如果更新失败 uncontended=false , 表明竞争激烈, 需要扩容Cells数组

然后执行累加方法 : longAccumulate(x, null, uncontended) , 可能涉及到新建Cell对象或者扩容

简单总结 :

（1）最初无竞争时只更新base；

（2）直到更新base失败时，创建cells数组；

（3）当多个线程竞争同一个Cell比较激烈时，可能要扩容；

longAccumulate

源码注释

final void longAccumulate(long x, LongBinaryOperator fn,
                              boolean wasUncontended) {
    // 存储线程的probe值
    int h;
    // 如果getProbe()方法返回0，说明随机数未初始化
    if ((h = getProbe()) == 0) {
        // 强制初始化
        ThreadLocalRandom.current(); // force initialization
        // 重新获取probe值
        h = getProbe();
        // 都未初始化，肯定还不存在竞争激烈
        wasUncontended = true;
    }
    // 是否发生碰撞
    boolean collide = false;                // True if last slot nonempty
    for (;;) {
        Cell[] as; Cell a; int n; long v;
        // cells已经初始化过
        if ((as = cells) != null && (n = as.length) > 0) {
            // 当前线程所在的Cell未初始化
            if ((a = as[(n - 1) & h]) == null) {
                // 当前无其它线程在创建或扩容cells，也没有线程在创建Cell
                if (cellsBusy == 0) {       // Try to attach new Cell
                    // 新建一个Cell，值为当前需要增加的值
                    Cell r = new Cell(x);   // Optimistically create
                    // 再次检测cellsBusy，并尝试更新它为1
                    // 相当于当前线程加锁
                    if (cellsBusy == 0 && casCellsBusy()) {
                        // 是否创建成功
                        boolean created = false;
                        try {               // Recheck under lock
                            Cell[] rs; int m, j;
                            // 重新获取cells，并找到当前线程hash到cells数组中的位置
                            // 这里一定要重新获取cells，因为as并不在锁定范围内
                            // 有可能已经扩容了，这里要重新获取
                            if ((rs = cells) != null &&
                                (m = rs.length) > 0 &&
                                rs[j = (m - 1) & h] == null) {
                                // 把上面新建的Cell放在cells的j位置处
                                rs[j] = r;
                                // 创建成功
                                created = true;
                            }
                        } finally {
                            // 相当于释放锁
                            cellsBusy = 0;
                        }
                        // 创建成功了就返回
                        // 值已经放在新建的Cell里面了
                        if (created)
                            break;
                        continue;           // Slot is now non-empty
                    }
                }
                // 标记当前未出现冲突
                collide = false;
            }
            // 当前线程所在的Cell不为空，且更新失败了
            // 这里简单地设为true，相当于简单地自旋一次
            // 通过下面的语句修改线程的probe再重新尝试
            else if (!wasUncontended)       // CAS already known to fail
                wasUncontended = true;      // Continue after rehash
            // 再次尝试CAS更新当前线程所在Cell的值，如果成功了就返回
            else if (a.cas(v = a.value, ((fn == null) ? v + x :
                                         fn.applyAsLong(v, x))))
                break;
            // 如果cells数组的长度达到了CPU核心数，或者cells扩容了
            // 设置collide为false并通过下面的语句修改线程的probe再重新尝试
            else if (n >= NCPU || cells != as)
                collide = false;            // At max size or stale
            // 上上个elseif都更新失败了，且上个条件不成立，说明出现冲突了
            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为新数组
                        cells = rs;
                    }
                } finally {
                    // 释放锁
                    cellsBusy = 0;
                }
                // 已解决冲突
                collide = false;
                // 使用扩容后的新数组重新尝试
                continue;                   // Retry with expanded table
            }
            // 更新失败或者达到了CPU核心数，重新生成probe，并重试
            h = advanceProbe(h);
        }
        // 未初始化过cells数组，尝试占有锁并初始化cells数组
        else if (cellsBusy == 0 && cells == as && casCellsBusy()) {
            // 是否初始化成功
            boolean init = false;
            try {                           // Initialize table
                // 检测是否有其它线程初始化过
                if (cells == as) {
                    // 新建一个大小为2的Cell数组
                    Cell[] rs = new Cell[2];
                    // 找到当前线程hash到数组中的位置并创建其对应的Cell
                    rs[h & 1] = new Cell(x);
                    // 赋值给cells数组
                    cells = rs;
                    // 初始化成功
                    init = true;
                }
            } finally {
                // 释放锁
                cellsBusy = 0;
            }
            // 初始化成功直接返回
            // 因为增加的值已经同时创建到Cell中了
            if (init)
                break;
        }
        // 如果有其它线程在初始化cells数组中，就尝试更新base
        // 如果成功了就返回
        else if (casBase(v = base, ((fn == null) ? v + x :
                                    fn.applyAsLong(v, x))))
            break;                          // Fall back on using base
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

简单总结

• 有cell数组, 新建cell对象, 然后存入cell数组
• 有cell数组, 但是出现冲突, cell数组需要扩容
• 没有cell数组, 初始化cell数组
• 以上过程都需要锁, 但是当前线程获取不到锁, 直接更新base值

详细解释

先看传参 :

• long x : 累加的值
• boolean wasUncontended , false 表示竞争激烈 需要扩容

然后是获取当前线程的prob值, 每个线程都会有一个随机的值, 这个值参与hash, 因为要把当前线程对应的Cell对象存入Cells数组, 使用的是hash的方式去获取应该存放的数组下标的位置

	// 存储线程的probe值
    int h;
    // 如果getProbe()方法返回0，说明随机数未初始化
    if ((h = getProbe()) == 0) {
        // 强制初始化
        ThreadLocalRandom.current(); // force initialization
        // 重新获取probe值
        h = getProbe();
        // 都未初始化，肯定还不存在竞争激烈
        wasUncontended = true;
    }
1
2
3
4
5
6
7
8
9
10
11

初始化一个值 boolean collide = false; 表示是否出现了 hash 碰撞

然后进入for循环内 :

第一个 if 条件 : 已经初始化过cell数组了

1. 先判断cells数组是否初始化(创建)
2. 数组中是否已经有Cell对象了

第二个else if 判断 : 未初始化过cells数组，尝试占有锁并初始化cells数组

1. 判断 cellsBusy == 0 即判断 cellBusy 是否存在锁, 0 表示无锁, 1 表示有锁
2. 判断是否创建了多个cell数组, 如果as和cells 不相等, 说明出现了多个 cells 数组
3. 尝试获取锁

第三个 else if 判断 : 其它线程在初始化cells数组中

1. 如果有其它线程在初始化cells数组中，就尝试更新base , 尝试更新base值,

以上三种情况每种只要执行完为 true , 最后都会 break, 结束循环, 不会同时执行

然后我们回到第一个if 判断 :

判断cell数组是否初始化, 如果已经初始化过了, 那就为 true , 进入代码块 :

简单描述 :

1. 当前线程对应的Cell对象没创建, 就创建cell对象, 然后存入cell数组,

• 但是可能会存在多个线程并发问题, 所以要先获取锁, 然后在创建cell对象并存入cells数组
• 存入cell数组的时候, 需要重新获取cells数组, 防止其他线程扩容cell数组
• 执行完成, 释放锁即可 然后break结束

2. 当前线程所在的Cell不为空, 说明Cell对象已经创建并且Cell值累加失败

• 注意啊 在 add() 方法里 (uncontended = a.cas(v = a.value, v + x) a 是Cell对象, 累加失败才会进入 longAccumulate
• 设置 wasUncontended = true;

3. 再次尝试CAS更新当前线程所在Cell的值，如果成功了就break
4. 如果cells数组的长度达到了CPU核心数，或者cells扩容了 collide = false

• 设置collide为false并通过下面的语句修改线程的probe再重新尝试

5. 上上个elseif都更新失败了，且上个条件不成立，说明出现冲突了 collide = true;

我们到第二个 else if 判断 : 明确出现冲突了，尝试占有锁，并扩容

简单描述

1. 检查是否有其它线程已经扩容过了 cells == as

• 数组大小扩容一倍
• 把旧数组元素拷贝到新数组中
• 重新赋值cells为新数组
• 释放锁

2. collide = false; 已解决冲突
3. 跳过当前循环, 然后重试

第三个 else if 判断 : 未初始化过cells数组，尝试占有锁并初始化cells数组

• 获取锁

• 检测是否有其它线程初始化过 cells == as

• 创建Cell数组

  1. 新建一个大小为2的Cell数组
  2. 找到当前线程hash到数组中的位置并创建其对应的Cell
  3. 赋值给cells数组
  4. 初始化成功 init = true;
  5. 释放锁

• 初始化成功直接返回 break

• 为增加的值已经同时创建到Cell中了