转载

Java并发 - atomic

原子类的特点

不可分割
一个操作是不可中断的,即使在多线程的情况下也可以保证

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

粒度更细:原子变量可以把竞争范围缩小到变量级别, 这是我们可以获得的最细粒度的情况了, 通常锁的粒度都要大于原子变量的粒度
效率更高:通常,使用原子类的效率会比使用锁的效率更高除了高度竞争的情况

原子类纵览

Atomic基本类演示

AtomicInteger常用方法

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

代码演示

public class CountExample {

    /**
     * 请求总数
     */
    public static int clientTotal = 5000;

    /**
     * 同事并发执行的线程数
     */
    public static int threadTotal = 200;

    /**
     * 计数值
     */
    private static final AtomicInteger atomicCount = new AtomicInteger(0);

    private static Integer count = 0;



    public static void main(String[] args) throws InterruptedException {
        Runnable runnable = () -> {
            // 原子类++操作
            atomicCount.incrementAndGet();
            count++;
        };

        ExecutorService executorService = Executors.newFixedThreadPool(threadTotal);
        for (int i = 0; i < clientTotal; i++) {
            executorService.execute(runnable);
        }
        
        // 使得主线程在累加完之后查看累加结果，也可以用join()
        Thread.sleep(1000);
        System.out.println("原子类多线程累加的结果 " + atomicCount.get());
        System.out.println("普通变量多线程累加的结果 " + count);

    }
}
复制代码

输出结果

原子类多线程累加的结果 5000 普通变量多线程累加的结果 4773

Atomic*Array演示

public class AtomicArrayDemo {

    public static void main(String[] args) {
        AtomicIntegerArray atomicIntegerArray = new AtomicIntegerArray(1000);
        Incrementer incrementer = new Incrementer(atomicIntegerArray) ;
        Decrementer decrementer = new Decrementer(atomicIntegerArray) ;

        Thread[] threadsIncrementer = new Thread[100];
        Thread[] threadsDecrementer = new Thread[100];
        for (int i = 0; i < 100; i++) {
            threadsDecrementer[i] = new Thread(decrementer);
            threadsIncrementer[i] = new Thread(incrementer);
            threadsDecrementer[i].start();
            threadsIncrementer[i].start();

        }

        for(int i=0;i<100;i++){
            try {
                threadsDecrementer[i].join() ;
                threadsIncrementer[i].join() ;
            } catch (InterruptedException  e) {
                e. printStackTrace() ;
            }
        }

        //正常情况下应该计算结果都是0 ，加了100次，减了100次
        for (int i = 0; i < atomicIntegerArray.length(); i++) {
            if (atomicIntegerArray.get(i) != 0) {
                System.err.println("发现错误，其数组下标为 " + i);
            }
        }

        System.out.println("运行结束");

    }
}

class Decrementer implements Runnable{

    private AtomicIntegerArray array;

    public Decrementer(AtomicIntegerArray atomicIntegerArray) {
        this.array = atomicIntegerArray;
    }

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

class Incrementer implements Runnable{

    private AtomicIntegerArray array;

    public Incrementer(AtomicIntegerArray atomicIntegerArray) {
        this.array = atomicIntegerArray;
    }

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

Atomic*Array就是数组中的每个元素都是Atomic基本类

Atomic*Reference演示

AtomicReference : AtomicReference类的作用,和AtomicInteger并没有本质区别，AtomicInteger可以让一个整数保证原子性,而AtomicReference可以 让一个对象保证原子性 ，当然,AtomicReference的功能明显比AtomicInteger强，因为一个对象里可以包含很多属性。用法和AtomicInteger类似。

使用AtomicReference实现自旋锁

public class SpinLock {

    AtomicReference<Thread> sign = new AtomicReference<>();

    public void lock() {
        Thread current = Thread.currentThread();
        while (!sign.compareAndSet(null,current)){
//            System.out.println("自旋获取失败，再次获取");
        }
    }

    public void unlock() {
        Thread current = Thread.currentThread();
        sign.compareAndSet(current,null);
    }

    public static void main(String[] args) throws InterruptedException {
        SpinLock lock = new SpinLock();
        Runnable runnable = new Runnable(){

            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName() + "开始尝试获取自旋锁");
                lock.lock();
                System.out.println(Thread.currentThread().getName() + "获取到了自旋锁。。。");
                try {
                    Thread.sleep(1000);
                }catch (Exception e) {
                    e.printStackTrace();
                }finally {
                    lock.unlock();
                    System.err.println(Thread.currentThread().getName() + "释放锁");
                }
            }
        };
        Thread thread1 = new Thread(runnable);
        Thread thread2 = new Thread(runnable);
        thread1.start();
        thread2.start();
        thread1.join();
        thread2.join();
    }
}
复制代码

Atomic*FieldUpdater演示

这个类型的原子类的作用主要就是将普通变量升级为具有原子功能的变量。可能这里大家会存在疑问，为什么不在一开始就直接使用原子类的来避免原子安全问题。 Atomic*FieldUpdater 有自己的适用场景：不允许对原有的类型进行直接修改，但是又存在并发问题的。但是偶尔又需要原子的 get-set 操作的。

原子类型的操作远比原子类型对应的基本类型的操作消耗的资源要大。

public class AtomicIntegerFieldUpdaterDemo implements Runnable{

    static Candidate tom;

    static Candidate peter;

    public static AtomicIntegerFieldUpdater<Candidate> scoreUpdater =
            AtomicIntegerFieldUpdater.newUpdater(Candidate.class,"score");

    @Override
    public void run() {
        for (int i = 0; i < 10000; i++) {
            peter.score++;
            scoreUpdater.getAndIncrement(tom);
        }
    }

    public static class Candidate{
        // updater要求进行升级的类型必须是使用volatile修饰的
        volatile int score;
    }


    public static void main(String[] args) throws InterruptedException {
        tom=new Candidate() ;
        peter=new Candidate( ) ;
        AtomicIntegerFieldUpdaterDemo r = new
                AtomicIntegerFieldUpdaterDemo() ;
        Thread t1 = new Thread(r) ;
        Thread t2 = new Thread(r) ;
        t1.start( ) ;
        t2.start( ) ;
        t1.join();
        t2.join();
        System.out.println("普通的变量：" + peter.score);
        tom.score++;
        // 使用tom.score和scoreUpdater.get(tom)结果一致
        System.out.println("升级的变量：" + tom.score + " - " +scoreUpdater.get(tom));
    }
}
复制代码

使用Atomic*Updater的注意点

可见范围

updater的底层是通过反射来实现的，所以可见性范围需要注意，不能设置为不可见的。

不支持static

Adder累加器演示

（JDK1.8引入）高并发下LongAdder比AtomicLong效率高,不过本质是空间换时间。竞争激烈的时候， LongAdder把不同线程对应到不同的Cell，上进行修改,降低了冲突的概率,是多段锁的理念,提高了并发性。

效率对比：AtomicLong VS LongAdder AtomicLong

public class AtomicLongDemo {

    public static void main(String[] args) throws InterruptedException {
        AtomicLong counter = new AtomicLong (0) ;
        ExecutorService service = Executors
                . newFixedThreadPool (20) ;
        Long start = System.currentTimeMillis();
        for(int i=0;i<10000;i++){
            service.submit(new Task(counter));
        }
        service.shutdown();
        while (!service.isTerminated()){

        }
        Long end = System.currentTimeMillis();
        System.out.println(counter.get());
        System.out.println("AtomicLong耗时：" + (end - start));
    }

    private static class Task implements Runnable{

        private AtomicLong counter;

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

        @Override
        public void run() {
            for (int i = 0; i < 10000; i++) {
                counter.incrementAndGet();
            }
        }
    }
}
复制代码

运行结果

100000000

LongAdder耗时：2014

LongAdder

public class LongAdderDemo {

    public static void main(String[] args) {
        LongAdder counter = new LongAdder () ;
        ExecutorService service = Executors
                . newFixedThreadPool (20) ;
        Long start = System.currentTimeMillis();
        for(int i=0;i<10000;i++){
            service.submit(new Task(counter));
        }
        service.shutdown();
        while (!service.isTerminated()){

        }
        Long end = System.currentTimeMillis();
        System.out.println(counter.sum());
        System.out.println("LongAdder耗时：" + (end - start));
    }

    private static class Task implements Runnable{

        private LongAdder counter;

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

        @Override
        public void run() {
            for (int i = 0; i < 10000; i++) {
                counter.increment();
            }
        }
    }
}
复制代码

运行结果

100000000

LongAdder耗时：224

原因分析： AtomicLong 每一次加法都需要flush和refresh ,导致很消耗资源

在内部,这个LongAdder的实现原理和刚才的AtomicLong是有不同的,刚才的AtomicLong的实现原理是,每一次加法都需要做同步 ,所以在高并发的时候会导致冲突比较多, 也就降低了效率。

而此时的LongAdder , 每个线程会有自己的一个计数器 ,仅用来在自己线程内计数,这样一来就不会和其他线程的计数器干扰。

LongAdder引入了分段累加的概念,内部有一个base变量和一个Cell[]数组共同参与计数:

base变量:竞争不激烈，直接累加到该变量上
Cell[]数组:竞争激烈,各个线程分散累加到自己的槽Cell[i]中

sum()源码分析

public long sum() {
        Cell[] as = cells; Cell a;
        long sum = base;
        // sum如果为空的话，直接加base，也就是竞争不激烈的情况下
        if (as != null) {
        // 可以使用cell进行求和的时候，没有进行加锁操作，内部元素可能在sum时被修改
        // 也就说，如果前面已经进行累加的结果被改变了，那么sum的结果会不精确
            for (int i = 0; i < as.length; ++i) {
                if ((a = as[i]) != null)
                    sum += a.value;
            }
        }
        return sum;
    }
复制代码

对比AtomicLong和LongAdder

在低争用下, AtomicLong和L ongAdder这两个类具有相似的特征。但是在竞争激烈的情况下, LongAdder的预期吞吐量要高得多,但要 消耗更多的空间
LongAdder适合的场景是统计求和计数的场景,而且 LongAdder基本只提供了add方法,而AtomicLong还具有 cas方法

Accumulator累加器

Accumulator和Adder非常相似, Accumulator就是一个更通用版本的Adder,其特点就是灵活

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

        }
        System.out.println(accumulator.getThenReset());
    }
}
复制代码

适用场景:适用于大量并行计算的场景。