LongAdder的源码学习与理解

// 累加单元组，懒惰初始化
transient volatile Cell[] cells;

// 基础值，如果没有竞争，则用cas累加这个值
transient volatile long base;

//在cells创建或者扩容时，置为1，表示加锁
transient volatile int cellsBusy;
1
2
3
4
5
6
7
8

cellsBusy作用是当要修改cells数组时加锁，防止多线程同时修改cells数组，0为无锁，1为加锁，加锁的情况有三种
1、cells数组初始化的时候
2、cells数组扩容的时候
3、如果cells数组中某个元素为null，给这个位置创建新的Cell对象的时候

base有两个作用
在开始没有竞争的时候，将累加值到base
在cells初始化的过程中，cells不可用，这时候会尝试将值累加到base上

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

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

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

因为CPU与内存的速度差异很大，需要靠预读数据至缓存来提升效率
而缓存以缓存行为单位，每个缓存行对应着一块内存，一般是64byte
缓存的加入会造成数据副本的产生，即同一份数据会缓存在不同核心的缓存行中
CPU要保证数据的一致性，如果某个CPU核心更改了数据，其他CPU核心对应的整个缓存行必须失效

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

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

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);
        }
    }
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

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
        }
    }

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

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

怎么确保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;
1
2
3
4
5
6
7
8
9
10