Java并发编程之原子性Atomic(八)

旧城等待， 2022-04-15 02:41 311阅读 0赞

**一.Atomic包简介**

*  Atomic包中的类基本的特性就是在多线程环境下，当有多个线程同时对单个（包括基本类型及引用类型）变量进行操作时，具有排他性，即当多个线程同时对该变量的值进行更新时，仅有一个线程能成功，而未成功的线程可以向自旋锁一样，继续尝试，一直等到执行成功。
 *  Atomic系列的类中的核心方法都会调用unsafe类中的几个本地方法。我们需要先知道一个东西就是Unsafe类，全名为：sun.misc.Unsafe，这个类包含了大量的对C代码的操作，包括很多直接内存分配以及原子操作的调用，而它之所以标记为非安全的，是告诉你这个里面大量的方法调用都会存在安全隐患，需要小心使用，否则会导致严重的后果，例如在通过unsafe分配内存的时候，如果自己指定某些区域可能会导致一些类似C++一样的指针越界到其他进程的问题。

**二.Java中CAS的原理实现**

*  CAS就是Compare and Swap的意思，比较并操作。很多的cpu直接支持CAS指令。CAS是项乐观锁技术，当多个线程尝试使用CAS同时更新同一个变量时，只有其中一个线程能更新变量的值，而其它线程都失败，失败的线程并不会被挂起，而是被告知这次竞争中失败，并可以再次尝试。CAS有3个操作数，内存值V，旧的预期值A，要修改的新值B。当且仅当预期值A和内存值V相同时，将内存值V修改为B，否则什么都不做。
 *  JDK1.5中引入了底层的支持，在int、long和对象的引用等类型上都公开了CAS的操作，并且JVM把它们编译为底层硬件提供的最有效的方法，在运行CAS的平台上，运行时把它们编译为相应的机器指令。在java.util.concurrent.atomic包下面的所有的原子变量类型中，比如AtomicInteger，都使用了这些底层的JVM支持为数字类型的引用类型提供一种高效的CAS操作。
 *  在CAS操作中，会出现ABA问题。就是如果V的值先由A变成B，再由B变成A，那么仍然认为是发生了变化，并需要重新执行算法中的步骤。有简单的解决方案：不是更新某个引用的值，而是更新两个值，包括一个引用和一个版本号，即使这个值由A变为B，然后为变为A，版本号也是不同的。AtomicStampedReference和AtomicMarkableReference支持在两个变量上执行原子的条件更新。AtomicStampedReference更新一个“对象-引用”二元组，通过在引用上加上“版本号”，从而避免ABA问题，AtomicMarkableReference将更新一个“对象引用-布尔值”的二元组。

**三.原子性Atomic包分类**

*  按照操作的数据类型可以分成4组

AtomicBoolean，AtomicInteger，AtomicLong

*  线程安全的基本类型的原子性操作

AtomicIntegerArray，AtomicLongArray，AtomicReferenceArray

*  线程安全的数组类型的原子性操作，它操作的不是整个数组，而是数组中的单个元素

AtomicLongFieldUpdater，AtomicIntegerFieldUpdater，AtomicReferenceFieldUpdater

*  基于反射原理对象中的基本类型（长整型、整型和引用类型）进行线程安全的操作

AtomicReference，AtomicMarkableReference，AtomicStampedReference

**四.源码分析**  
以下以AtomicInteger为例子,对其源码进行分析

1.有参构造函数 ，成员变量value声明为volatile类型，说明了多线程下的可见性，即任何一个线程的修改，在其它线程中都会被立刻看到

private volatile int value;
    
        /** * Creates a new AtomicInteger with the given initial value. * @param initialValue the initial value */
        public AtomicInteger(int initialValue) { 
            value = initialValue;
        }

2.Unsafe的具体实现是为了获取value在堆内存中的偏移量。  
valueOffset表示value数据域相对于this的内存地址的偏移位置，偏移量在AtomicInteger中很重要，原子操作都靠它来实现。

> sun.misc.Unsafe是JDK内部用的工具类。它通过暴露一些Java意义上说“不安全”的功能给Java层代码，来让JDK能够更多的使用Java代码来实现一些原本是平台相关的、需要使用native语言（例如C或C++）才可以实现的功能。该类不应该在JDK核心类库之外使用。

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

3.关键代码,原子递增当前值

/** * Atomically increments by one the current value. * @return the updated value */
     public final int incrementAndGet() { 
       		 // this表示AtomicInteger实例
          return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
     }

compareAndSwapInt是native方法,java底层的方法，注意 Unsafe的对象不能直接new,要通过反射去获取，AtomicInteger类本质就是利用Unsafe.compareAndSwapInt这个CAS操作方法来保证Integer操作的原子性。  
逻辑如下：  
var1为当前对象值，var2为当前对象偏移量值，var4为累加量1  
var5为从底层方法getIntVolatite方法获取到的底层方法的值  
如果当前对象值var2与从底层获取过来的值var5一样,则进行加1操作,即var5+var4,如果当前对象值var2被其他线程更改过导致var2与var5不一致的话,则继续获取当前值var2来与var5判断,通过不断的获取最新值判断,直到当前值与底层值一致,则返回var4+var5的最新值

public final int getAndAddInt(Object var1, long var2, int var4) { 
            int var5;
            do { 
            //获取value在内存中的值，然后进行CAS操作
                var5 = this.getIntVolatile(var1, var2);
            } while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));
    
            return var5;
        }

AtomicInteger 中主要实现了整型的原子操作，防止并发情况下出现异常结果，其内部主要依靠JDK 中的unsafe 类操作内存中的数据来实现的。volatile 修饰符保证了value在内存中其他线程可以看到其值得改变。CAS操作保证了AtomicInteger 可以安全的修改value 的值。  
**五.示例**  
1.AtomicInteger测试类

public static AtomicInteger num=new AtomicInteger();
        public static int ThreadPoolNum=1000;//线程数量
        private static  int PREMITS=1111;//线程并发的信号量
        public static void main(String[] args) { 
            final Semaphore semaphore = new Semaphore(PREMITS);
            ExecutorService executorService = Executors.newCachedThreadPool();//线程池
            for(int i=0;i<ThreadPoolNum;i++){ 
                executorService.execute(new Runnable() { 
                    @Override
                    public void run() { 
                        try { 
                            semaphore.acquire();
                            num.incrementAndGet();
                            System.out.println("结果num:"+num);
                            semaphore.release();
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    }
                });
            }
            executorService.shutdown();

结果num:1000,AtomicLong与其基本相同  
2.AtomicBoolean测试类

public static AtomicBoolean isUpdate=new AtomicBoolean();
        public static int ThreadPoolNum=5000;//线程数量
        private static  int PREMITS=1111;//线程并发的信号量
        public static void main(String[] args) { 
            isUpdate.set(false);
    
            final Semaphore semaphore = new Semaphore(PREMITS);
            ExecutorService executorService = Executors.newCachedThreadPool();//线程池
            for(int i=0;i<ThreadPoolNum;i++){ 
                executorService.execute(new Runnable() { 
                    @Override
                    public void run() { 
                        try { 
                            semaphore.acquire();
                            //线程安全类,只执行一次
                            if(isUpdate.compareAndSet(false, true)){ 
                                System.out.println("更新成功");
                            }
                            semaphore.release();
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    }
                });
            }
            executorService.shutdown();
        }

结果只返回一次 更新成功  
3.AtomicIntergerFiledUpdater测试类

private volatile int age=20;
        private static AtomicIntegerFieldUpdater<AtomicIntergerFiledUpdaterTest> updater=AtomicIntegerFieldUpdater.newUpdater(AtomicIntergerFiledUpdaterTest.class,"age");
    
        public static void main(String[] args) { 
            AtomicIntergerFiledUpdaterTest test = new AtomicIntergerFiledUpdaterTest();
            if(updater.compareAndSet(test,20,30)){ 
                System.out.println("更新成功."+test.getAge());
            }else { 
                System.out.println("更新失败."+test.getAge());
            }
        }
    
        public Integer getAge() { 
            return age;
        }

结果:更新成功.30  
4. AtomicReferencer测试类

static AtomicReference<Integer> num=new AtomicReference<Integer>();
        public static void main(String[] args) { 
            num.set(0);
            num.compareAndSet(0,1);
            num.compareAndSet(2,1);
            num.compareAndSet(1,11);
            System.out.println(num.get());//11
        }

结果:11  
5.LongAdd测试类

public static LongAdder num=new LongAdder();
        public static int ThreadPoolNum=1000;//线程数量
        private static  int PREMITS=1111;//线程并发的信号量
        public static void main(String[] args) { 
            final Semaphore semaphore = new Semaphore(PREMITS);
            ExecutorService executorService = Executors.newSingleThreadExecutor();//线程池
            for(int i=0;i<ThreadPoolNum;i++){ 
                executorService.execute(new Runnable() { 
                    @Override
                    public void run() { 
                        try { 
                            semaphore.acquire();
                            num.increment();
                            System.out.println("结果num:"+num);
                            semaphore.release();
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                    }
                });
            }
            executorService.shutdown();
        }

结果:1000  
LongAdder在AtomicLong的基础上将单点的更新压力分散到各个节点，在低并发的时候通过对base的直接更新可以很好的保障和AtomicLong的性能基本保持一致，而在高并发的时候通过分散提高了性能。  
缺点是LongAdder在统计的时候如果有并发更新，可能导致统计的数据有误差。