Java并发 - CAS+Unsafe+原子操作

﹏ヽ暗。殇╰゛Y 2022-09-07 05:49 358阅读 0赞

# Java并发 - CAS+Unsafe+原子操作 #

线程的三大特性：**原子性**、有序性、可见性。

## 竞态条件与临界区 ##

在同一程序中运行多个线程本身不会导致问题，问题在于多个线程访问了相同的资源。向这些资源做了写操作时才有可能发生，只要资源没有发生变化，多个线程读取相同的资源就是安全的。

临界区：关键部分代码的多线程并发执行，会对执行结果产生影响，那么add 方法内部就是临界区域。

竞态条件：可能发生在临界区域内的特殊条件。多线程执行add方法中的 n++ 关键代码时，产生了竞态条件。

public int n = 0;
    public void add() { 
        n++;
    }

存在静态条件，线程不安全，需要转变为原子操作才能安全，

## 共享资源 ##

1、如果一段代码是线程安全的, 则它不包含竞态条件。只有当多个线程更新共享资源时，才会发生竞态条件。

2、局部对象引用本身不共享，但是引用的对象存储在共享堆中。如果方法内创建的对象，只是在方法中传递，并且不对其他线程可用，那么也是线程安全的。

3、栈封闭时，不会在线程之间共享的变量都是线程安全的。

## 不可变对象 ##

1、创建不可变的共享对象来保证对象在线程间共享时不会被修改，从而实现线程安全。

2、实例被创建，value变量就不能再被修改，这就是不可变性。

## 原子操作定义 ##

强调一下原子性的定义：不可在分割的最小单位、不可被外界干扰。

1、原子操作可以是一个步骤，也可以是多个操作步骤，但是其顺序不可以被打乱，也不可以被切割而只执行其中的一部分(不可中断性)。

2、将整个操作视作一个整体是原子性的核心特征。

至此，上面的 n++ 线程不安全的原因也是可以做一个回答了， n++ 使用Javap -v 编译后分成了以下几个步骤，而且存在被JIT编译成汇编时重排序的可能性，不符合原子操作的定义，存在竞态条件，线程不安全。

0: getstatic     #5 
    3: iconst_1
    4: iadd
    5: putstatic     #5

让 n++ 变成线程安全，只需要保证其原子性，方式有使用CAS机制和锁两种方式。

## CAS机制 ##

乐观锁就是使用的CAS机制+版本号的方式实现的。

Compare And Swap 比较和交换，属于硬件同步原语，处理器提供了基本内存操作的原子性保证。比较当前工作内存中的值和主内存中的值，如果这个值是期望的，那么执行操作，否则就一直循环。

在Java中，JDK提供了 sun.misc.Unsafe 类，提供了 compareAndSwapInt() 和 compareAndSwapLong() 等几个方法实现CAS。

public void test1() { 
        // 大坑：如果泛型是一个包装类，注意对象被缓存引用的问题
        AtomicInteger atomicInteger = new AtomicInteger(2020);
        // expect 期望，达到期望后更新为 update 的值
        boolean isUpdate = atomicInteger.compareAndSet(2020, 2021);
        System.out.println("是否更新了值：" + isUpdate);
        System.out.println("更新成了：" + atomicInteger.get());
    
        // 修改后并未通知其它线程
        atomicInteger.compareAndSet(2021, 2020);
    
        // 自旋锁
        atomicInteger.getAndIncrement();
    
        boolean isUpdate2 = atomicInteger.compareAndSet(2020, 2021);
        System.out.println("是否更新了值：" + isUpdate2);
        System.out.println("更新成了：" + atomicInteger.get());
    }

### Unsafe类 ###

Java 无法直接操作内存，Unsafe类相当于Java的后门，可以通过C++来操作内存。JDK不建议直接使用Unsafe类，JDK底层有很多针对不同平台运行的代码，使用Unsafe类可能会造成一些跨平台的问题。

Unsafe类中有以下三个方法：

/** * @param var1 操作的对象 * @param var2 操作对象的属性偏移量，用于获取该属性在内存中的具体地址 * @param var4 期望值，读取传入对象o在内存中偏移量为offset位置的值与期望值expected作比较 * 相等就把x值赋值给offset位置的值。方法返回true * @param var5 目标值 * @return 与目标的期望值不相等返回false */
    public final native boolean compareAndSwapObject(Object var1, long var2, Object var4, Object var5);
    
    public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5);
    
    public final native boolean compareAndSwapLong(Object var1, long var2, long var4, long var6);

使用Unsafe类的方法对线程不安全的add()方法进行修改：

private int value = 0;
    /** * 获取并使用 Unsafe 对象： * Java 无法直接操作内存，但可以通过C++来操作内存， * Unsafe类相当于Java的后门，可以通过这个类来操作内存、修改对象、数组内存 */
    private static Unsafe unsafe;
    /** * 获取到value属性偏移量，用于定义value属性在内存中的具体地址 */
    private static long valueOffset;
    
    static { 
        try { 
            // 通过反射获取Unsafe对象
            Field field = Unsafe.class.getDeclaredField("theUnsafe");
            field.setAccessible(true);
            // 获取Unsafe实例
            unsafe = (Unsafe) field.get(null);
            valueOffset = unsafe.objectFieldOffset(CASDemo.class.getDeclaredField("value"));
        } catch (Exception e) { 
            e.printStackTrace();
        }
    }
    
    /** * 该方法经过以下问题不断改进 */
    public void add() { 
        // 1、三个步骤
        // value++;
        // 2、可能会失败，不能每次都成功
        // unsafe.compareAndSwapInt(this, valueOffset, value, value + 1);
        // 3、CAS+死循环重试，确保每一次都能操作成功
        int current, target;
        do { 
            // 当前值可能被其它线程改了，每一次循环都重新获取一次值
            current = unsafe.getIntVolatile(this, valueOffset);
            target = current + 1;
            // current 是期望值，如果比较的期望值相同则修改为值为target，并返回true
        } while (!unsafe.compareAndSwapInt(this, valueOffset, current, target));
    }
    
    public void test2() throws InterruptedException { 
        for (int i = 0; i < 2; i++) { 
            new Thread(() -> { 
                for (int j = 0; j < 1000; j++) { 
                    add();
                }
            }).start();
        }
    
        TimeUnit.SECONDS.sleep(2);
        System.out.println(value);
    }

在 java.util.concurrent.atomic 包下，提供了一些原子操作类，举两个例子：

### AtomicInteger类 ###

对Compare And Swap的理解示例

public void test() { 
        // 大坑：如果泛型是一个包装类，注意对象被缓存引用的问题
        AtomicInteger atomicInteger = new AtomicInteger(2020);
        // expect 期望，达到期望后更新为 update 的值
        boolean isUpdate = atomicInteger.compareAndSet(2020, 2021);
        System.out.println("是否更新了值：" + isUpdate + "值为：" + atomicInteger.get());
    
        // 修改后并未通知其它线程
        atomicInteger.compareAndSet(2021, 2020);
    	// 进行加1操作
        atomicInteger.getAndIncrement();
    
        boolean isUpdate2 = atomicInteger.compareAndSet(2020, 2021);
        System.out.println("是否更新了值：" + isUpdate2 + "值为：" + atomicInteger.get());
    }

针对n++的问题，使用原子类解决

AtomicInteger n = new AtomicInteger(0);
    public void test4() throws InterruptedException { 
        for (int i = 0; i < 2; i++) { 
            new Thread(() -> { 
                for (int j = 0; j < 1000; j++) { 
                    add4();
                }
            }).start();
        }
    
        TimeUnit.SECONDS.sleep(2);
        System.out.println(n.get());
    }
    
    public void add4(){ 
        n.incrementAndGet();
    }

### CAS优缺点 ###

优点：

*  线程一直处于活跃状态（非阻塞），不需要CPU进行频繁调度
 *  可避免死锁

缺点：

*  不适用于大段代码，只使用与对单个变量进行同步。CAS只能把unsafe.compareAndSwapInt(this, valueOffset, current, target)这个修改内存地址的操作变成一个原子性操作，如果是大段的代码需要保证一个原子性操作，就不适用了。所以CAS适用于针对某一个线程共享的变量、属性。
 *  不适用于大量并发的环境。如果CAS中执行非常耗时的操作的话，线程会占用大量的CPU执行时间，如果线程数不断增加，CPU承载能力有限。
 *  ABA问题。

假设此处有三个线程A、B、C，都在修改一个value值，在A->B->C修改的整个过程不受干扰的情况下，A将value改为1，B又将value改为2，C又将value改回了1，但其它线程并不知道当前value已经修改过了。

CAS操作可能带来ABA问题，因为CAS操作需要在操作值的时候，检查值有没有发生变化，如果没有发发生变化则更新。

解决方案：像乐观锁一样，使用版本号。在变量前面追加版本号，每次变量更新时把版本号加1。

可用AtomicStampedReference类解决原子引用类型：

public void test3() { 
        // 初始期望值为2020，初始版本号为1
        AtomicStampedReference<Integer> reference = new AtomicStampedReference<>(1, 1);
        new Thread(() -> { 
            System.out.println("A1=>" + reference.getStamp());
            try { 
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            boolean b1 = reference.compareAndSet(1, 2, reference.getStamp(), reference.getStamp() + 1);
            System.out.println("A2=>" + b1 + "=>" + reference.getStamp());
    
            boolean b2 = reference.compareAndSet(2, 1, reference.getStamp(), reference.getStamp() + 1);
            System.out.println("A3=>" + b2 + "=>" + reference.getStamp());
        }, "A").start();
    
        new Thread(() -> { 
            System.out.println("B1=>" + reference.getStamp());
            try { 
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
    
            boolean b = reference.compareAndSet(1, 3, reference.getStamp(), reference.getStamp() + 1);
            System.out.println("B2=>" + b + "=>" + reference.getStamp());
        }, "B").start();
    
        // main+gc
        if (Thread.activeCount() == 4) { 
            System.out.println("执行完毕");
        }
    }