Java 原子操作类之18罗汉增强类

Java开发手册

17.【参考】volatile 解决多线程内存不可见问题对于一写多读，是可以解决变量同步问题，但是如果多

写，同样无法解决线程安全问题。

说明：如果是 count++操作，使用如下类实现：

AtomicInteger count = new AtomicInteger();

count.addAndGet(1);

如果是 JDK8，推荐使用 LongAdder 对象，比 AtomicLong 性能更好（减少乐观锁的重试次数）

基本类型原子类

• AtomicInteger
• AtomicBoolean
• AtomicLong

常用API简介

• public final int get() //获取当前的值
• public final int getAndSet(int newValue)//获取当前的值，并设置新的值
• public final int getAndIncrement()//获取当前的值，并自增
• public final int getAndDecrement() //获取当前的值，并自减
• public final int getAndAdd(int delta) //获取当前的值，并加上预期的值
• boolean compareAndSet(int expect, int update) //如果输入的数值等于预期值，则以原子方式将该值设置为输入值（update）

举个栗子

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

class MyNumber {

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


public class AtomicIntegerDemo {

    public static final int SIZE = 50;

    public static void main(String[] args) throws InterruptedException {

        MyNumber myNumber = new MyNumber();

        CountDownLatch countDownLatch = new CountDownLatch(SIZE);
        for (int i = 1; i <= SIZE; i++) {
            new Thread(() -> {
                try {
                    for (int j = 1 ;j <=1000; j++) {
                        myNumber.addPlusPlus();
                    }
                }catch (Exception e){
                    e.printStackTrace();
                }finally {
                    countDownLatch.countDown();
                }
            },String.valueOf(i)).start();
        }

        countDownLatch.await();

        System.out.println(Thread.currentThread().getName()+"t"+"---result : "+myNumber.atomicInteger.get());


    }
}

上述案例使用的AtomicInteger进行的类似累加的操作，底层使用的volatile来实现的，已经保障了可见性，所以数据一定是正确的。

之所以是CountDownLatch 是为了保障main线程输出结果的时候，所有的线程都已经完成了计算。

数组类型原子类

• AtomicIntegerArray
• AtomicLongArray
• AtomicReferenceArray

demo

import java.util.concurrent.atomic.AtomicIntegerArray;


public class AtomicIntegerArrayDemo {

    public static void main(String[] args) {
        AtomicIntegerArray atomicIntegerArray = new AtomicIntegerArray(5);

        for (int i = 0; i <atomicIntegerArray.length(); i++) {
            System.out.println(atomicIntegerArray.get(i));
        }
        System.out.println();
        System.out.println();
        System.out.println();
        int tmpInt = 0;

        tmpInt = atomicIntegerArray.getAndSet(0,1122);
        System.out.println(tmpInt+"t"+atomicIntegerArray.get(0));
        atomicIntegerArray.getAndIncrement(1);
        atomicIntegerArray.getAndIncrement(1);
        tmpInt = atomicIntegerArray.getAndIncrement(1);
        System.out.println(tmpInt+"t"+atomicIntegerArray.get(1));

    }
}

引用类型原子类

• AtomicReference
• AtomicStampedReference
• AtomicMarkableReference

AtomicReference

使用场景

解决并发修改多个属性

AtomicInteger、AtomicBoolean等java.util.concurrent包下面的类，但是这个只能并发修改一个属性，如果我需要对多个属性同时进行并发修改，并且保证原子性呢？

AtomicReference和AtomicInteger非常类似，不同之处就在于AtomicInteger是对整数的封装，而AtomicReference则对应普通的对象引用，是操控多个属性的原子性的并发类。

举个栗子

public class AtomicReferenceDemo {
    public static void main(String[] args) {
        User z3 = new User("z3",24);
        User li4 = new User("li4",26);

        AtomicReference<User> atomicReferenceUser = new AtomicReference<>();

        atomicReferenceUser.set(z3);

        System.out.println(atomicReferenceUser.compareAndSet(z3,li4)+"t"+atomicReferenceUser.get().toString());
        System.out.println(atomicReferenceUser.compareAndSet(z3,li4)+"t"+atomicReferenceUser.get().toString());

    }
}

使用AtomicReference实现CAS

/**
 * 题目：实现一个自旋锁
 * 自旋锁好处：循环比较获取没有类似wait的阻塞。
 *
 * 通过CAS操作完成自旋锁，A线程先进来调用myLock方法自己持有锁5秒钟，B随后进来后发现
 * 当前有线程持有锁，不是null，所以只能通过自旋等待，直到A释放锁后B随后抢到。
 */
public class SpinLockDemo {
    AtomicReference<Thread> atomicReference = new AtomicReference<>();

    public void MyLock() {
        System.out.println(Thread.currentThread().getName()+"t"+"---come in");
        while(!atomicReference.compareAndSet(null,Thread.currentThread())) {

        }
        System.out.println(Thread.currentThread().getName()+"t"+"---持有锁成功");
    }

    public void MyUnLock() {
        atomicReference.compareAndSet(Thread.currentThread(),null);
        System.out.println(Thread.currentThread().getName()+"t"+"---释放锁成功");
    }

    public static void main(String[] args) {
        SpinLockDemo spinLockDemo = new SpinLockDemo();

        new Thread(() -> {
            spinLockDemo.MyLock();
            try { TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e) { e.printStackTrace(); }
            spinLockDemo.MyUnLock();
        },"t1").start();

        new Thread(() -> {
            spinLockDemo.MyLock();
            spinLockDemo.MyUnLock();
        },"t2").start();
    }
}

AtomicStampedReference

携带版本号的引用类型原子类，可以解决ABA问题

demo

public class ABADemo {
    static AtomicInteger atomicInteger = new AtomicInteger(100);
    static AtomicStampedReference<Integer> atomicStampedReference = new AtomicStampedReference<>(100,1);

    public static void main(String[] args) {
        new Thread(() -> {
            int stamp = atomicStampedReference.getStamp();
            System.out.println(Thread.currentThread().getName()+"t"+"---默认版本号: "+stamp);
            //让后面的t4获得和t3一样的版本号，都是1，好比较
            try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }

            atomicStampedReference.compareAndSet(100,101,stamp,stamp+1);
            System.out.println(Thread.currentThread().getName()+"t"+"---1次版本号: "+atomicStampedReference.getStamp());
            atomicStampedReference.compareAndSet(101,100,atomicStampedReference.getStamp(),atomicStampedReference.getStamp()+1);
            System.out.println(Thread.currentThread().getName()+"t"+"---2次版本号: "+atomicStampedReference.getStamp());
        },"t3").start();

        new Thread(() -> {
            int stamp = atomicStampedReference.getStamp();
            System.out.println(Thread.currentThread().getName()+"t"+"---默认版本号: "+stamp);
            //上前面的t3完成ABA问题
            try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); }
            boolean result = atomicStampedReference.compareAndSet(100, 20210308, stamp, stamp + 1);
            System.out.println(Thread.currentThread().getName()+"t"+"---操作成功否："+result+"t"+atomicStampedReference.getStamp()+"t"+atomicStampedReference.getReference());
        },"t4").start();
    }

    public static void abaProblem() {
        new Thread(() -> {
            atomicInteger.compareAndSet(100,101);
            atomicInteger.compareAndSet(101,100);
        },"t1").start();

        //暂停毫秒
        try { TimeUnit.MILLISECONDS.sleep(10); } catch (InterruptedException e) { e.printStackTrace(); }

        new Thread(() -> {
            boolean b = atomicInteger.compareAndSet(100, 20210308);
            System.out.println(Thread.currentThread().getName()+"t"+"修改成功否："+b+"t"+atomicInteger.get());
        },"t2").start();
    }
}

AtomicMarkableReference

原子更新带有标记位的引用类型对象，它的定义就是将状态戳简化为true|false，

可以理解为上面AtomicStampedReference的简化版，就是不关心修改过几次，仅仅关心是否修改过。因此变量mark是boolean类型，仅记录值是否有过修改。不建议使用。

demo

public class ABADemo {

    static AtomicMarkableReference markableReference = new AtomicMarkableReference(100,false);

    public static void main(String[] args) {

        System.out.println("============AtomicMarkableReference不关心引用变量更改过几次，只关心是否更改过======================");

        new Thread(() -> {
            boolean marked = markableReference.isMarked();
            System.out.println(Thread.currentThread().getName()+"t 1次版本号"+marked);
            try { TimeUnit.MILLISECONDS.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); }
            markableReference.compareAndSet(100,101,marked,!marked);
            System.out.println(Thread.currentThread().getName()+"t 2次版本号"+markableReference.isMarked());
            markableReference.compareAndSet(101,100,markableReference.isMarked(),!markableReference.isMarked());
            System.out.println(Thread.currentThread().getName()+"t 3次版本号"+markableReference.isMarked());
        },"t5").start();

        new Thread(() -> {
            boolean marked = markableReference.isMarked();
            System.out.println(Thread.currentThread().getName()+"t 1次版本号"+marked);
            //暂停几秒钟线程
            try { TimeUnit.MILLISECONDS.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); }
            markableReference.compareAndSet(100,2020, marked, !marked);
            System.out.println(Thread.currentThread().getName()+"t"+markableReference.getReference()+"t"+markableReference.isMarked());
        },"t6").start();
    }
}

对象的属性修改原子类

• AtomicIntegerFieldUpdater：原子更新对象中int类型字段的值
• AtomicLongFieldUpdater：原子更新对象中Long类型字段的值
• AtomicReferenceFieldUpdater：原子更新引用类型字段的值

为什么有这些东西？

使用目的：以一种线程安全的方式操作非线程安全对象内的某些字段。

使用要求

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

demo

AtomicIntegerFieldUpdaterDemo

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;

class BankAccount {
    String bankName = "ccb";

    //以一种线程安全的方式操作非线程安全对象内的某些字段

    //1 更新的对象属性必须使用 public volatile 修饰符。
    public volatile int money = 0;

    //2 因为对象的属性修改类型原子类都是抽象类，所以每次使用都必须
    // 使用静态方法newUpdater()创建一个更新器，并且需要设置想要更新的类和属性。
    private static final AtomicIntegerFieldUpdater<BankAccount> FieldUpdater = AtomicIntegerFieldUpdater.newUpdater(BankAccount.class,"money");

    public void transfer(BankAccount bankAccount) {
        FieldUpdater.incrementAndGet(bankAccount);
    }
}


public class AtomicIntegerFieldUpdaterDemo {

    public static void main(String[] args) throws InterruptedException {
        BankAccount bankAccount = new BankAccount();

        for (int i = 1; i <=1000; i++) {
            new Thread(() -> {
                bankAccount.transfer(bankAccount);
            },String.valueOf(i)).start();
        }

        //暂停几秒钟线程
        try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }

        System.out.println(Thread.currentThread().getName()+"t"+"---bankAccount: "+bankAccount.money);
    }
}

AtomicReferenceFieldUpdater

package com.atguigu.juc.atomics;

import lombok.Data;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;

@Data
class MyVar {
    public volatile String isInit = "111";
    private static final AtomicReferenceFieldUpdater<MyVar,String> FieldUpdater = AtomicReferenceFieldUpdater.newUpdater(MyVar.class,String.class,"isInit");

    public void init(MyVar myVar) {
        if(FieldUpdater.compareAndSet(myVar,"111", "222")) {
            System.out.println(Thread.currentThread().getName()+"t"+"---start init");
            try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); }
            System.out.println(Thread.currentThread().getName()+"t"+"---end init -- " + myVar.getIsInit());
        }else{
            System.out.println(Thread.currentThread().getName()+"t"+"---抢夺失败，已经有线程在修改中 --" + myVar.getIsInit());
        }
    }

}


/**
 * 
 *  多线程并发调用一个类的初始化方法，如果未被初始化过，将执行初始化工作，要求只能初始化一次
 */
public class AtomicReferenceFieldUpdaterDemo {
    public static void main(String[] args) {
        MyVar myVar = new MyVar();
        for (int i = 1; i <=5; i++) {
            new Thread(() -> {
                myVar.init(myVar);
            },String.valueOf(i)).start();
        }
    }
}

关联面试题

1. 面试官问你：你在哪里用了volatile ：AtomicReferenceFieldUpdater ：
2. 既然已经有了AtomicInteger，为什么又多此一举弄出个AtomicIntegerFieldUpdater来呢？其实主要有两方面原因：

1. 要使用AtomicInteger需要修改代码，将原来int类型改造成AtomicInteger，使用该对象的地方都要进行调整（多进行一次get()操作获取值），但是有时候代码不是我们想改就能改动的。
2. 也是比较重要的一个特性，AtomicIntegerFieldUpdater可以节省内存消耗。
3. 引用：https://juejin.cn/post/6907610980428021773#comment

原子操作增强类

分类

• DoubleAccumulator
• DoubleAdder
• LongAccumulator
• LongAdder

前言

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

入门讲解

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

demo

import java.util.concurrent.atomic.LongAccumulator;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.LongBinaryOperator;

public class LongAdderAPIDemo {
    public static void main(String[] args) {
        LongAdder longAdder = new LongAdder();//只能做加法

        longAdder.increment();
        longAdder.increment();
        longAdder.increment();

        System.out.println(longAdder.longValue());

        LongAccumulator longAccumulator = new LongAccumulator((left, right) -> left - right, 100);

        longAccumulator.accumulate(1);//1
        longAccumulator.accumulate(2);//3
        longAccumulator.accumulate(3);//6

        System.out.println(longAccumulator.longValue());


    }
}

demo–LongAdder高性能对比Code

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;

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();
        //longAdder.sum();
    }

    LongAccumulator longAccumulator = new LongAccumulator(Long::sum,0);
    public void add_LongAccumulator() {
        longAccumulator.accumulate(1);
    }

}


/**
 *
 *  50个线程，每个线程100W次，总点赞数出来
 */
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;
        long endTime;

        CountDownLatch countDownLatch1 = new CountDownLatch(SIZE_THREAD);
        CountDownLatch countDownLatch2 = new CountDownLatch(SIZE_THREAD);
        CountDownLatch countDownLatch3 = new CountDownLatch(SIZE_THREAD);
        CountDownLatch countDownLatch4 = new CountDownLatch(SIZE_THREAD);
        CountDownLatch countDownLatch5 = 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 {
                    countDownLatch1.countDown();
                }
            },String.valueOf(i)).start();
        }
        countDownLatch1.await();
        endTime = System.currentTimeMillis();
        System.out.println("----costTime: "+(endTime - startTime) +" 毫秒"+"t add_Synchronized"+"t"+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 {
                    countDownLatch2.countDown();
                }
            },String.valueOf(i)).start();
        }
        countDownLatch2.await();
        endTime = System.currentTimeMillis();
        System.out.println("----costTime: "+(endTime - startTime) +" 毫秒"+"t add_AtomicInteger"+"t"+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 {
                    countDownLatch3.countDown();
                }
            },String.valueOf(i)).start();
        }
        countDownLatch3.await();
        endTime = System.currentTimeMillis();
        System.out.println("----costTime: "+(endTime - startTime) +" 毫秒"+"t add_AtomicLong"+"t"+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 {
                    countDownLatch4.countDown();
                }
            },String.valueOf(i)).start();
        }
        countDownLatch4.await();
        endTime = System.currentTimeMillis();
        System.out.println("----costTime: "+(endTime - startTime) +" 毫秒"+"t add_LongAdder"+"t"+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 {
                    countDownLatch5.countDown();
                }
            },String.valueOf(i)).start();
        }
        countDownLatch5.await();
        endTime = System.currentTimeMillis();
        System.out.println("----costTime: "+(endTime - startTime) +" 毫秒"+"t add_LongAccumulator"+"t"+clickNumber.longAccumulator.longValue());
    }
}

结果

源码、原理分析

官方api

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

LongAdder是Striped64的子类，Striped64有几个比较重要的成员函数

• base变量：非竞争状态条件下,直接累加到该变量上
• Cell[ ]数组:竞争条件下(高并发下),累加各个线程自己的槽Cell[i]中

/** Number of CPUS, to place bound on table sizeCPU数量，即cells数组的最大长度 */
static final int NCPU = Runtime.getRuntime().availableProcessors();

/**
 * Table of cells. When non-null, size is a power of 2.
cells数组，为2的幂，2,4,8,16.....，方便以后位运算
 */
transient volatile Cell[] cells;

/**基础value值，当并发较低时，只累加该值主要用于没有竞争的情况，通过CAS更新。
 * Base value, used mainly when there is no contention, but also as
 * a fallback during table initialization races. Updated via CAS.
 */
transient volatile long base;

/**创建或者扩容Cells数组时使用的自旋锁变量调整单元格大小（扩容），创建单元格时使用的锁。
 * Spinlock (locked via CAS) used when resizing and/or creating Cells. 
 */
transient volatile int cellsBusy;

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

LongAdder为什么这么快-分散热点

LongAdder在无竞争的情况,跟AtomicLong一样，对同一个base进行操作。

当出现竞争关系时则采用化整为零的做法，用空间换时间，用一个数组cells将一个value拆分进这个数组cells。多个线程需要同时对value进行操作时候，

对线程id进行hash，再根据hash值映射到这个数组cells的某个下标，再对该下标所对应的值进行自增操作。当所有线程操作完毕，将数组cells的所有值和无竞争值base都加起来作为最终结果。

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