原子操作类AtomicLong、LongAdder、LongAccumulator

1. 原子操作类

1.1 原子操作类概述

多线程环境不使用原子类保证线程安全i（基本数据类型)

public class AtomicIntegerTest {

    volatile int num=0;

    public int getNumber(){
        return num;
    }

    public synchronized void addNumber(){
         num++;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12

多线程环境使用原子类保证线程安全i++（基本数据类型）

public class AtomicIntegerTest {

    AtomicInteger atomicInteger = new AtomicInteger(0);

    public int getNum(){
        return atomicInteger.get();
    }

    public void addNum(){
        atomicInteger.getAndIncrement();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12

阿里Java开发手册
volatile解决多线程内存不可见问题对于一写多读，是可以解决变量同步问题，但是如果多写，同样无法解决线程安全问题。
如果是count++操作，使用如下类实现:
AtomicInteger count = new AtomicInteger();
count.addAndGet(1);
如果是JDK8，推荐使用LongAdder对象，比AtomicLong性能更好（减少乐观锁的重试次数)。

1.2 分类

原子操作类位于java.util.concurrent.atomic包下
按类型分类

1.2.1 基本类型原子类

AtomicInteger常用API
AtomicInteger 常用API操作

public class AtomicIntegerTest2 {
    public static void main(String[] args) {
        AtomicInteger atomicInteger = new AtomicInteger(0);
        int size=50;
        CountDownLatch latch = new CountDownLatch(size);
        for(int i=0;i<size;i++){
            new Thread(()->{
                try {
                    for(int j=0;j<10000;j++){
                        atomicInteger.getAndIncrement();
                    }
                }finally {
                    latch.countDown();
                }

            },""+i).start();
        }
        try {
            latch.await(1, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            System.err.println("===============执行中断=============");
            return;
        }
        if(latch.getCount()>0){
            System.err.println("===============执行超时=============");
            return;
        }
        System.out.println(atomicInteger.get());
    }
}
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

CountDownLatch 最佳实践：使用带有超时时间的await的方法，完了之后判断count是否大于0，来判断是否执行完毕还是代码超时。比起不带超时的await方法，优点在于等待时间可控，不会因意外一直等待。

1.2.2 数组类型原子类

AtomicIntegerArray初始化必须指定数组，方法API也十分简单

public class AtomicIntegerArrayTest {
    public static void main(String[] args) {
        AtomicIntegerArray atomicIntegerArray = new AtomicIntegerArray(new int[5]);
        for (int i=0;i<atomicIntegerArray.length();i++){
            System.out.println(atomicIntegerArray.get(i));
        }
        AtomicIntegerArray array = new AtomicIntegerArray(new int[]{1,2,3,4,5});
        array.getAndAdd(0, 999);
        System.out.println(array.get(0));
        array.getAndIncrement(4);
        System.out.println(array.get(4));
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13

1.2.3 引用类型原子类

AtomicStampedReference

携带版本号的引用类型原子类，可以解决ABA问题。解决修改过几次。状态戳原子引用

AtomicMarkableReference

原子更新带有标记位的引用类型对象。它的定义就是将状态戳简化为true/false。解决是否修改过，类似一次性筷子。状态戳(true/false)原子引用

代码演示

public class AtomicMarkableReferenceTest {
    public static void main(String[] args) {
        AtomicMarkableReference<Integer> reference = new AtomicMarkableReference<>(1,false);
        new Thread(()->{
            boolean marked = reference.isMarked();
            Integer integer = reference.getReference();
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            reference.compareAndSet(integer,99,marked,!marked);
        },"A").start();
        new Thread(()->{
            boolean marked = reference.isMarked();
            Integer integer = reference.getReference();
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("B线程修改结果："+reference.compareAndSet(integer, 66, marked, !marked));
            System.out.println("最终标志位："+reference.isMarked()+"，最终结果:"+reference.getReference());
        },"B").start();
    }
}
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

1.2.4 对象的属性修改原子类

以一种线程安全的方式操作非线程安全对象内的某些字段
AtomiclntegerFieldUpdater：原子更新对象中int类型字段的值，基于反射的实用程序，可对指定类的指定volatile int字段进行原子更新。
AtomicLongFieldUpdater：原子更新对象中Long类型字段的值，基于反射的实用程序，可以对指定类的指定volatile long字段进行原子更新。
AtomicReferenceFieldUpdater：原子更新引用类型字段的值，基于反射的实用程序，可以对指定类的指定volatile引用字段进行原子更新。

使用要求：
更新的对象属性必须使用volatile修饰符。
因为对象的属性修改类型原子类都是抽象类，所以每次使用都必须使用静态方法newUpdater()创建一个更新器，并且需要设置想要更新的类和属性。

AtomiclntegerFieldUpdater代码

public class AtomicIntegerFieldUpdaterTest {

    public static void main(String[] args) {
        int size = 50;
        CountDownLatch latch = new CountDownLatch(size);
        Dog dog = new Dog();
        for (int i = 0; i < 50; i++) {
            new Thread(() -> {
                try {
                    for (int j = 0; j < 10000; j++) {
                        dog.add();
                    }
                } finally {
                    latch.countDown();
                }
            }, String.valueOf(i)).start();
        }
        try {
            latch.await(1, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        if (latch.getCount() > 0) {
            System.out.println("============执行超时============");
            return;
        }
        System.out.println(dog.age);
    }

    static class Dog {
        private volatile int age;
        private String name;

        AtomicIntegerFieldUpdater<Dog> updater = AtomicIntegerFieldUpdater.newUpdater(Dog.class, "age");

        public void add() {
            updater.getAndIncrement(this);
        }
    }
}
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

AtomicReferenceFieldUpdater代码

public class AtomicReferenceFieldUpdaterTest {
    public static void main(String[] args) {
        Fish fish = new Fish();
        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                fish.init();
            }).start();
        }
    }

    static class Fish {
        volatile Boolean isLive = Boolean.FALSE;
        AtomicReferenceFieldUpdater<Fish, Boolean> updater = AtomicReferenceFieldUpdater.newUpdater(Fish.class, Boolean.class, "isLive");

        public void init() {
            if (updater.compareAndSet(this, Boolean.FALSE, Boolean.TRUE)) {
                System.out.println("初始化的线程：" + Thread.currentThread().getName());
            } else {
                System.out.println("==========已被其他线程初始化=============");
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

1.2.5 原子操作增强类原理深度解析

热点商品点赞计算器，点赞数加加统计，不要求实时精确
一个很大的List，里面都是int类型，如何实现加加，说说思路

LongAdder

当多个线程更新用于收集统计信息但不用于细粒度同步控制的目的的公共和时，此类通常优于AtomicLong。在低更新争用下，这两个类具有相似的特征。但在高争用的情况下，这一类的预期吞吐量明显更高，但代价是空间消耗更高。

LongAdder常用API
LongAdder只能用来计算加减法，且从零开始计算
LongAccumulator提供了自定义的函数操作

代码

public class LongAdderTest {
    public static void main(String[] args) {
        LongAdder longAdder = new LongAdder();
        longAdder.add(3L);
        longAdder.add(5L);
        longAdder.increment();
        System.out.println(longAdder.sum());
        longAdder.decrement();
        System.out.println(longAdder.sum());

        LongAccumulator longAccumulator = new LongAccumulator((x,y)->x+y,0);
        longAccumulator.accumulate(6);
        longAccumulator.accumulate(2);
        System.out.println(longAccumulator.get());
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

开100个线程，每个线程累加100万次，比较synchronized、AtomicInteger、LongAdder、LongAccumulator的执行时间

public class LogKpiCountTest {
    public static void main(String[] args) {
        LogKpiCount count = new LogKpiCount();
        int size = 100;
        CountDownLatch latch1 = new CountDownLatch(size);
        CountDownLatch latch2 = new CountDownLatch(size);
        CountDownLatch latch3 = new CountDownLatch(size);
        CountDownLatch latch4 = new CountDownLatch(size);
        Long startTime=null;
        startTime=System.currentTimeMillis();
        for (int i=0;i<size;i++){
            new Thread(()->{
                try {
                    for (int j = 0; j < 1000000; j++) {
                        count.synchronizedAdd();
                    }
                }finally {
                    latch1.countDown();
                }

            }).start();
        }
        try {
            latch1.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("synchronizedAdd执行耗时："+(System.currentTimeMillis()-startTime)+",执行结果:"+count.num);

        startTime=System.currentTimeMillis();
        for (int i=0;i<size;i++){
            new Thread(()->{
                try {
                    for (int j = 0; j < 1000000; j++) {
                        count.atomicIntegerAdd();
                    }
                }finally {
                    latch2.countDown();
                }

            }).start();
        }
        try {
            latch2.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("atomicIntegerAdd执行耗时："+(System.currentTimeMillis()-startTime)+",执行结果:"+count.atomicInteger.get());

        startTime=System.currentTimeMillis();
        for (int i=0;i<size;i++){
            new Thread(()->{
                try {
                    for (int j = 0; j < 1000000; j++) {
                        count.longAdderAdd();
                    }
                }finally {
                    latch3.countDown();
                }

            }).start();
        }
        try {
            latch3.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("longAdderAdd执行耗时："+(System.currentTimeMillis()-startTime)+",执行结果:"+count.longAdder.sum());

        startTime=System.currentTimeMillis();
        for (int i=0;i<size;i++){
            new Thread(()->{
                try {
                    for (int j = 0; j < 1000000; j++) {
                        count.longAccumulatorAdd();
                    }
                }finally {
                    latch4.countDown();
                }

            }).start();
        }
        try {
            latch4.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("longAccumulatorAdd执行耗时："+(System.currentTimeMillis()-startTime)+",执行结果:"+count.accumulator.get());
    }

    static class LogKpiCount{
        int num;
        AtomicInteger atomicInteger = new AtomicInteger();
        LongAdder longAdder = new LongAdder();
        LongAccumulator accumulator = new LongAccumulator((x,y)->x+y,0);
        public synchronized void synchronizedAdd(){
            num++;
        }

        public void atomicIntegerAdd(){
            atomicInteger.getAndIncrement();
        }

        public void longAdderAdd(){
            longAdder.increment();
        }

        public void longAccumulatorAdd(){
            accumulator.accumulate(1);
        }
    }
}
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

执行结果如下，发现synchronized与其他三个存在数量级上的差异，AtomicInteger与其他两个存在数量级上的差异。

synchronizedAdd执行耗时：6763,执行结果:100000000
atomicIntegerAdd执行耗时：836,执行结果:100000000
longAdderAdd执行耗时：82,执行结果:100000000
longAccumulatorAdd执行耗时：99,执行结果:100000000
1
2
3
4

LongAdder架构
LongAdder是Striped64的子类

最重要的两个参数cells和base。

Cell是java.util.concurrent.atomic 下 Striped64的一个内部类

LongAdder为什么这么快

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

sum()会将所有Cell数组中的value和base累加作为返回值，核心的思想就是将之前AtomicLong一个value的更新压力分散到多个value中去，从而降级更新热点。

1.3 原子操作类总结

注：本文是学习B站周阳老师《尚硅谷2022版JUC并发编程》课程所做学习笔记。