JUC中的Atomic原子类

@[TOC]

Atomic 原子类


1. 原子类介绍

  1. 不可分割的

  2. 一个操作是不可中断的,即使多线程的情况下也可以保证, 即使是在多个线程一起执行的时候,一个操作一旦开始,就不会被其他线程干扰。

  3. 原子类的作用和锁类似,是为了保证并发情况下线程安全,不过相比于锁,更有优势

    优势: 粒度更细,效率更高

原子类纵览:

类型Value
Atomic*基本类型原子类AtomicInteger
AtomicLong
AtomicBoolean
Atomic*Arrays数组类型原子类AtomicIntegerArray
AtomicLongArray
AtomicReferenceArray
Atomic*Reference引用类型原子类AtomicReference
AtomicStampedReference
AtomicMarkableReference
Atomic*Fieldupdate升级类型原子类AtomicIntegerFieldUpdater
AtomicLongFieldUpdater
AtomicReferenceFieldUpdater
Adder累加器LongAdder
DoubleAdder
Accumulator累加器LongAccumulator
DoubleAccumulator

2. 基本类型原子类


  • AtomicInteger :整型原子类
  • AtomicLong :长整型原子类
  • AtomicBoolean :布尔型原子类

以 AtomicInteger 为例

AtomicInteger 类常用方法 :

public final int get()  //获取值

public final void set(int newValue) // 设置值

public final void lazySet(int newValue) //最终设置为给定的值

public final int getAndSet(int newValue) // 获取当前值,并设置新值

public final int getAndIncrement() //获取当前值 并自增

public final int getAndDecrement() //获取当前的值,并自减

public final int getAndAdd(int delta) // 获取当前值,并加上预期值

public final int getAndAdd(int delta) // 获取当前值,并加上预期值

public final boolean compareAndSet(int expect, int update) //比较并替换


使用 :

import java.util.concurrent.atomic.AtomicInteger;

public class AtomicIntegerTest {

    public static void main(String[] args) {
        
        int temvalue = 0;
        AtomicInteger i = new AtomicInteger(0);
        temvalue = i.getAndSet(3);
        System.out.println("temvalue:" + temvalue + ";  i:" + i);//temvalue:0;  i:3
        temvalue = i.getAndIncrement();
        System.out.println("temvalue:" + temvalue + ";  i:" + i);//temvalue:3;  i:4
        temvalue = i.getAndAdd(5);
        System.out.println("temvalue:" + temvalue + ";  i:" + i);//temvalue:4;  i:9
    }

}

案例 :

public class Test {


    public static void main(String[] args) {

        TestDemo thread = new TestDemo();
        Thread t1 = new Thread(thread,"窗口一");
        Thread t2 = new Thread(thread,"窗口二");
        t1.start();
        t2.start();
    }
}

class TestDemo implements Runnable{
    //共享的火车票变量
    private  int count = 100;

    //重寫run方法
    @Override
    public void run() {
        while (count > 0){
            try {
                //休眠一下 方便出现并发问题
                Thread.sleep(50);
            }catch (Exception e){
                e.getMessage();
            }
            sale();
        }
    }
    //卖票
    public void sale(){
        if(count > 0){
            System.out.println(Thread.currentThread().getName() +"出售 :" +(100 -  count + 1));
            count--;
        }
    }

}

窗口一出售 :37 窗口二出售 :39 窗口一出售 :40 窗口二出售 :41 窗口一出售 :41 窗口一出售 :43 窗口二出售 :43 窗口一出售 :45 窗口二出售 :45 窗口一出售 :47 窗口二出售 :47

多线程下会出现重复卖票的情况,我们解决这个问题可以使用 JDK 内置锁 (Synchronized)保证线程原子性,当某个线程取到锁后,其他线程就会等待,但是性能低下,我们可以使用AtomicInteger类,是一个专门提供可以保证原子性的类

package com.dimple.test;

import java.util.concurrent.atomic.AtomicInteger;

public class Test5 {


    public static void main(String[] args) {

        TestDemo thread = new TestDemo();
        Thread t1 = new Thread(thread,"窗口一");
        Thread t2 = new Thread(thread,"窗口二");
        t1.start();
        t2.start();
    }
}

class TestDemo implements Runnable{
    //共享的火车票变量
    private static AtomicInteger atomic = new AtomicInteger(100);

    //重寫run方法
    @Override
    public void run() {
        while (atomic.get() > 0){
            try {
                //休眠一下 方便出现并发问题
                Thread.sleep(50);
            }catch (Exception e){
                e.getMessage();
            }
            sale();
        }
    }
    //卖票
    public void sale(){
        if(atomic.get() > 0){
          	Integer count=  100 - atomic.getAndDecrement() + 1; //使用底层方法getAndDecrement()  自-1;
            System.out.println(Thread.currentThread().getName()+ "," + count);//获取当前值
        }
    }
}


窗口一出售 :91 窗口二出售 :92 窗口一出售 :93 窗口二出售 :94 窗口一出售 :95 窗口二出售 :96 窗口一出售 :97 窗口二出售 :98 窗口一出售 :99 窗口二出售 :100

原理分析 :

// setup to use Unsafe.compareAndSwapInt for updates
    private static final Unsafe unsafe = Unsafe.getUnsafe();
    private static final long valueOffset;

    static {
        try {
            valueOffset = unsafe.objectFieldOffset
                (AtomicInteger.class.getDeclaredField("value"));
        } catch (Exception ex) { throw new Error(ex); }
    }

    private volatile int value;

    public AtomicInteger(int initialValue) {
        value = initialValue;
    }
  1. AtomicInteger 类主要利用 CAS(compare and swap) + volatile native 方法来保证原子操作,从而避免 synchronized 的高开销,执行效率大为提升。
  2. CAS(compare and swap) 的原理是那期望值和原本的一个值作比较,如果相同则更新成新的值, UnSafe 类的 objectFieldOffset() 方法是一个本地方法,这个方法是用来拿到 “原来的值” 的内存地址。并且 value 是一个 volatile变量,在内存中可见,因此 JVM可以保证任何时刻线程总能拿到该变量的最新值。

3. 数组类型原子类


  • AtomicIntegerArray :整形数组原子类
  • AtomicLongArray :长整形数组原子类
  • AtomicReferenceArray :引用类型数组原子类

以 AtomicIntegerArray 为例

AtomicIntegerArray 类常用方法

public final int get(int i) //获取值

public final void set(int i, int newValue) // 设置值

public final void lazySet(int i, int newValue) //最终集的元素在位置 i到给定值。

public final int getAndSet(int i, int newValue) //自动设置元素的位置 i到给定值并返回旧值。

public final boolean compareAndSet(int i, int expect, int update) //自动设置元素的位置 i给更新后的值,如果预期值 ==期望值。

public final int getAndIncrement(int i) //自动递增一个指数 i元素。

public final int getAndDecrement(int i) //自动递减指数 i元素

使用 :


import java.util.concurrent.atomic.AtomicIntegerArray;

public class AtomicIntegerArrayTest {

    public static void main(String[] args) {
        int temvalue = 0;
        int[] nums = { 1, 2, 3, 4, 5, 6 };
        AtomicIntegerArray i = new AtomicIntegerArray(nums);
        for (int j = 0; j < nums.length; j++) {
            System.out.println(i.get(j));
        }
        temvalue = i.getAndSet(0, 2);
        System.out.println("temvalue:" + temvalue + ";  i:" + i);
        temvalue = i.getAndIncrement(0);
        System.out.println("temvalue:" + temvalue + ";  i:" + i);
        temvalue = i.getAndAdd(0, 5);
        System.out.println("temvalue:" + temvalue + ";  i:" + i);
    }

}

案例 :

public class AtmoicArray {
    public static void main(String[] args) throws InterruptedException {
      AtomicIntegerArray atomicIntegerArray=new AtomicIntegerArray(1000);

        Decrement decrement = new Decrement(atomicIntegerArray);
        Increment increment = new Increment(atomicIntegerArray);


        Thread[] threads= new Thread[100];
        Thread[] threads2= new Thread[100];

        for (int i = 0; i < 100 ; i++) {
            threads2[i]=new Thread(decrement);
            threads[i]=new Thread(increment);
            threads2[i].start();
            threads[i].start();
        }

        for (int i = 0; i < 100 ; i++) {
            threads2[i].join();
            threads[i].join();
        }

        for (int i = 0; i <  atomicIntegerArray.length(); i++) {
            if(atomicIntegerArray.get(i)!=0) {
                System.out.println("发现非0值" + i);
            }
        }
        System.out.println("运行结束");

    }


}

class  Decrement implements Runnable{
    private  AtomicIntegerArray array;

    Decrement(AtomicIntegerArray array) {
        this.array = array;
    }

    @Override
    public void run() {
        for (int i = 0; i < array.length() ; i++) {
            array.getAndDecrement(i);
        }

    }
}

class  Increment implements Runnable{
    private  AtomicIntegerArray array;

    Increment(AtomicIntegerArray array) {
        this.array = array;
    }

    @Override
    public void run() {
        for (int i = 0; i < array.length() ; i++) {
            array.getAndIncrement(i);
        }

    }
}


运行结束

运行结果 ,会发现我们数组线程每次 加100 减100 ,并不会出现不等于0的数据,数据并没有出现错乱,AtomicIntegerArray 给我们提供了数组的原子性

原理分析 :

 private static final Unsafe unsafe = Unsafe.getUnsafe();
 private static final int base = unsafe.arrayBaseOffset(int[].class);
 private static final int shift;
 private final int[] array;

static {
        int scale = unsafe.arrayIndexScale(int[].class);
        if ((scale & (scale - 1)) != 0)
            throw new Error("data type scale not a power of two");
        shift = 31 - Integer.numberOfLeadingZeros(scale);
    }
 
    private long checkedByteOffset(int i) {
        if (i < 0 || i >= array.length)
            throw new IndexOutOfBoundsException("index " + i);
 
        return byteOffset(i);
    }
 
    private static long byteOffset(int i) {
        return ((long) i << shift) + base;
    }


  1. unsafe.arrayBaseOffset 获取该类型的数组,在对象存储时,存放第一个元素的内存地址,相对于数组对象起始的内存偏移量,unsafe.arrayIndexSacle(int[].class) 获取该类型的数组中元素的大小,占用多少个字节
  2. 根据scale ,base 定位到任意一个下标的地址 举例 : int scale = 4;1个int类型,在java中占用4个字节, Integer.numberOfLeadingZeros(scale); 返回 scale 高位连续0的个数,得出shift = 2, 而shift在如下方法使用,shift就是用来定位数组中的内存位置,用来移位

4. 引用类型原子类


基本类型原子类只能更新一个变量,如果需要原子更新多个变量,需要使用引用类型原子类

  • AtomicReference :引用类型原子类
  • AtomicStampedReference :原子更新带有版本号的引用类型
  • AtomicMarkableReference :原子更新带有标记的引用类型

AtomicReference 常用方法

public final V get()  //获取值

public final void set(V newValue) //设置值

public final void lazySet(V newValue) //最终设置为给定的值

public final boolean compareAndSet(V expect, V update) //自动设置的值来指定更新值

public final V getAndSet(V newValue) //自动设置为给定的值并返回旧值。

public final V getAndUpdate(UnaryOperator<V> updateFunction) //自动更新当前值与结果应用给定的函数,返回前一个值。

AtomicReference 使用

public class Test {

    public static void main(String[] args) {
        AtomicReference<Person> atomicReference = new AtomicReference<Person>();
        Person person = new Person("abc", 22);
        atomicReference.set(person);
        Person updatePerson = new Person("Daisy", 20);
        atomicReference.compareAndSet(person, updatePerson);
        System.out.println(atomicReference.get().getName());
        System.out.println(atomicReference.get().getAge());
    }


}

@Data
class Person {

    private String name;

    private int age;

  
}

Daisy 20

AtomicStampedReference 使用

public class Test {


    public static void main(String[] args) {
        // 实例化、取当前值和 stamp 值
        final Integer initialRef = 0, initialStamp = 0;
        final AtomicStampedReference<Integer> asr = new AtomicStampedReference<Integer>(initialRef, initialStamp);
        System.out.println("currentValue=" + asr.getReference() + ", currentStamp=" + asr.getStamp());

        // compare and set
        final Integer newReference = 666, newStamp = 999;
        final boolean casResult = asr.compareAndSet(initialRef, newReference, initialStamp, newStamp);
        System.out.println("currentValue=" + asr.getReference()
                + ", currentStamp=" + asr.getStamp()
                + ", casResult=" + casResult);

        // 获取当前的值和当前的 stamp 值
        int[] arr = new int[1];
        final Integer currentValue = asr.get(arr);
        final int currentStamp = arr[0];
        System.out.println("currentValue=" + currentValue + ", currentStamp=" + currentStamp);

        // 单独设置 stamp 值
        final boolean attemptStampResult = asr.attemptStamp(newReference, 88);
        System.out.println("currentValue=" + asr.getReference()
                + ", currentStamp=" + asr.getStamp()
                + ", attemptStampResult=" + attemptStampResult);

        // 重新设置当前值和 stamp 值
        asr.set(initialRef, initialStamp);
        System.out.println("currentValue=" + asr.getReference() + ", currentStamp=" + asr.getStamp());

}

currentValue=0, currentStamp=0 currentValue=666, currentStamp=999, casResult=true currentValue=666, currentStamp=999 currentValue=666, currentStamp=88, attemptStampResult=true currentValue=0, currentStamp=0


AtomicMarkableReference 使用

public class Test {

    public static void main(String[] args) {
        // 实例化、取当前值和 mark 值
        final Boolean initialRef = null, initialMark = false;
        final AtomicMarkableReference<Boolean> amr = new AtomicMarkableReference<Boolean>(initialRef, initialMark);
        System.out.println("currentValue=" + amr.getReference() + ", currentMark=" + amr.isMarked());

        // compare and set
        final Boolean newReference1 = true, newMark1 = true;
        final boolean casResult = amr.compareAndSet(initialRef, newReference1, initialMark, newMark1);
        System.out.println("currentValue=" + amr.getReference()
                + ", currentMark=" + amr.isMarked()
                + ", casResult=" + casResult);

        // 获取当前的值和当前的 mark 值
        boolean[] arr = new boolean[1];
        final Boolean currentValue = amr.get(arr);
        final boolean currentMark = arr[0];
        System.out.println("currentValue=" + currentValue + ", currentMark=" + currentMark);

        // 单独设置 mark 值
        final boolean attemptMarkResult = amr.attemptMark(newReference1, false);
        System.out.println("currentValue=" + amr.getReference()
                + ", currentMark=" + amr.isMarked()
                + ", attemptMarkResult=" + attemptMarkResult);

        // 重新设置当前值和 mark 值
        amr.set(initialRef, initialMark);
        System.out.println("currentValue=" + amr.getReference() + ", currentMark=" + amr.isMarked());

    }

currentValue=null, currentMark=false currentValue=true, currentMark=true, casResult=true currentValue=true, currentMark=true currentValue=true, currentMark=false, attemptMarkResult=true currentValue=null, currentMark=false

5. 升级类型原子类

  • AtomicIntegerFieldUpdater : 原子更新整形字段的更新器
  • AtomicLongFieldUpdater :原子更新长整形字段的更新器
  • AtomicReferenceFieldUpdater :原子更新引用类型里的字段的更新器

AtomicIntegerFieldUpdater 常用方法

public final V get()  //获取值

public final void set(V newValue) //设置值

public final void lazySet(V newValue) //最终设置为给定的值

public final boolean compareAndSet(V expect, V update) //自动设置的值来指定更新值

public final V getAndSet(V newValue) //自动设置为给定的值并返回旧值。

public final V getAndUpdate(UnaryOperator<V> updateFunction) //自动更新当前值与结果应用给定的函数,返回前一个值。

public final V getAndAccumulate(T obj, V x,
                                    BinaryOperator<V> accumulatorFunction) //自动更新与应用给出的函数的值与给定值的结果指标 i元素,返回前一个值。

public final V accumulateAndGet(T obj, V x,BinaryOperator<V> accumulatorFunction) //自动更新与应用给出的函数的值与给定值的结果指标 i元素,返回更新后的值。

AtomicIntegerFieldUpdater 使用

public class Test {
    public static void main(String[] args) {
        AtomicIntegerFieldUpdater<User> a = AtomicIntegerFieldUpdater.newUpdater(User.class, "age");

        User user = new User("Java", 22);
        System.out.println(a.getAndIncrement(user));// 22
        System.out.println(a.get(user));// 23
    }
}

@Data
class User {

    private String name;
    
    public volatile int age;
}

22 23

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

6. Adder 累加器

  • 是JDK 1.8 中 引入的一个比较新的类
  • 高并发的情况下 LongAdder 比 AtomitLong 效率高,不过是空间换时间
  • 竞争激烈的时候,LingAdder 把不同的线程对应到不同的cell 上修改,降低了冲突的概率,是多段锁的理念,提高了并发性

测试 AtomicLong的性能

import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 演示高并发情况下LongAdder 比 AtomicLong
 * 性能好
 */
public class AtomicLongDemo {

    public static void main(String[] args) throws InterruptedException {
        AtomicLong atomicLong = new AtomicLong(0);
        //线程池开始时间
        long start = System.currentTimeMillis();
        ExecutorService executorService = Executors.newFixedThreadPool(20);
        for (int i = 0; i < 10000; i++) {
            executorService.submit(new Task(atomicLong));
        }
        //表示线程池执行完毕
        executorService.shutdown();
        while (!executorService.isTerminated()){

        }
        long end = System.currentTimeMillis();
        System.out.println(atomicLong.get());
        System.out.println("耗时"+(end -start));
    }


    public static class Task implements Runnable{

        private AtomicLong atomicLong;

        public Task(AtomicLong atomicLong) {
            this.atomicLong = atomicLong;
        }

        @Override
        public void run() {

            for (int i = 0; i < 10000; i++) {
                atomicLong.incrementAndGet();
            }
        }
    }
}

在这里插入图片描述

测试 LongAdder 性能

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;

/**
 * 演示高并发情况下LongAdder 比 AtomicLong
 * 性能好
 */
public class LongAdderDemo {

    public static void main(String[] args) throws InterruptedException {
        LongAdder atomicLong = new LongAdder();
        //线程池开始时间
        long start = System.currentTimeMillis();
        ExecutorService executorService = Executors.newFixedThreadPool(20);
        for (int i = 0; i < 10000; i++) {
            executorService.submit(new Task(atomicLong));
        }
        //表示线程池执行完毕
        executorService.shutdown();
        while (!executorService.isTerminated()){

        }
        long end = System.currentTimeMillis();
        System.out.println(atomicLong.sum() );
        System.out.println("耗时"+(end -start));
    }


    public static class Task implements Runnable{

        private LongAdder longAdder;

        public Task(LongAdder longAdder) {
            this.longAdder = longAdder;
        }

        @Override
        public void run() {

            for (int i = 0; i < 10000; i++) {
                longAdder.increment();
            }
        }
    }
}

在这里插入图片描述

会发现 LongAdder 比 AtomicLong 快了好多

  1. 他们内部实现有些不同,AtomicLong每次加法都需要同步,所以冲突比较多,也就降低了效率
  2. 而 LongAdder ,每个线程都有自己的计数器,仅用来线程计数,不会和其他线程打扰
  3. AtomicLong引入了分段锁的概念,内部有一个base变量 和 cell[] 数组共同参与计数
  4. base变量:竞争不激烈,直接累加到该变量上
  5. cell [] 数组: 竞争激烈,各个线程分累加到自己到cell[i] 卡槽中

7. Accumulator 累加器

  • Accumualtor 和 Adder 非常相似,Accumualtor就是更通用的版本Adder
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.LongAccumulator;
import java.util.stream.IntStream;

public class LongAccumulatorDemo {

    public static void main(String[] args) {
        LongAccumulator accumulator = new LongAccumulator((x,y)->x+y,0);
        ExecutorService executorService = Executors.newFixedThreadPool(8);
        IntStream.range(1,10).forEach(i->executorService.submit(()->accumulator.accumulate(i)));
        executorService.shutdown();
        while (!executorService.isTerminated())
            System.out.println(accumulator.getThenReset());
    }

45


个人博客地址:http://blog.yanxiaolong.cn   | 『纵有疾风起,人生不言弃』

end
  • 作者:yxl(联系作者)
  • 发表时间:2021-04-16 14:45
  • 版权声明:自由转载-非商用-非衍生-保持署名(创意共享3.0许可证)
  • 转载声明:如果是转载栈主转载的文章,请附上原文链接
  • 公众号转载:请在文末添加作者公众号二维码(公众号二维码见右边,欢迎关注)
  • 评论