Java并发——CAS原理分析

时间 2019-11-10

标签 java 并发 cas 原理分析栏目 Java 繁體版

原文原文链接

咱们知道多线程操做共享资源时，会出现三个问题：可见性、有序性以及原子性。java

通常状况下，咱们采用synchronized同步锁（独占锁、互斥锁），即同一时间只有一个线程可以修改共享变量，
其余线程必须等待。可是这样的话就至关于单线程，体现不出来多线程的优点。

那么咱们有没有另外一种方式来解决这三个问题呢？算法

Java中有一个volatile关键字，它能够解决可见性和有序性的问题。并且若是操做的共享变量是基本数据类型，
而且同一时间只对变量进行读取或者写入的操做，那么原子性问题也获得了解决，就不会产生多线程问题了。

可是一般，咱们都要先读取共享变量，而后操做共享变量，最后写入共享变量，那么这个时候怎么保证整个操做的原子性呢？一种解决方式就是CAS技术。在讲解这个以前，先了解两个重要概念：悲观锁与乐观锁。数据库

一. 悲观锁与乐观锁

悲观锁：老是假设最坏的状况，每次去拿数据的时候都认为别人会修改，因此每次在拿数据的时候都会上锁，这样别人想拿这个数据就会阻塞直到它拿到锁（共享资源每次只给一个线程使用，其它线程阻塞，用完后再把资源转让给其它线程）。传统的关系型数据库里边就用到了不少这种锁机制，好比行锁、表锁等、读锁、写锁等，都是在作操做以前先上锁。Java中Synchronized和ReentrantLock等独占锁就是悲观锁思想的实现。

// MySQL InnoDB持经过特定的语句进行显示锁定
SELECT … FOR UPDATE

乐观锁：老是假设最好的状况，每次去拿数据的时候都认为别人不会修改，因此不会上锁，可是在更新的时候会判断一下在此期间别人有没有去更新这个数据，能够使用版本号机制和CAS算法实现。乐观锁适用于多读的应用类型，这样能够提升吞吐量，像数据库提供的相似于write_condition机制，其实都是提供的乐观锁。在Java中java.util.concurrent.atomic包下面的原子变量类就是使用了乐观锁的一种实现方式CAS实现的。

两种锁各有优缺点，不可认为一种好于另外一种，像乐观锁适用于写比较少的状况下，即冲突真的不多发生的时候，这样能够省去了锁的开销，加大了系统的整个吞吐量。但若是常常产生冲突，上层应用会不断的进行retry，这样反却是下降了性能，因此这种状况下用悲观锁就比较合适。编程

悲观锁会阻塞其余线程。乐观锁不会阻塞其余线程，若是发生冲突，采用死循环的方式一直重试，直到更新成功。

能够参考《乐观锁、悲观锁，这一篇就够了！》segmentfault

二. CAS算法的实现原理

即compare and swap（比较与交换），是一种有名的无锁算法。无锁编程，即不使用锁的状况下实现多线程之间的变量同步，也就是在没有线程被阻塞的状况下实现变量的同步，因此也叫非阻塞同步（Non-blocking Synchronization）。CAS算法包含三个值：当前内存值(V)、预期原来的值(A)以及期待更新的值(B)。安全

若是内存位置V的值与预期原值A相匹配，那么处理器会自动将该位置值更新为新值B，返回true。不然处理器不作任何操做，返回false，这时会不断的重试，直到没有冲突，更新成功。

实现CAS最重要的一点，就是比较和交换操做的一致性，不然就会产生歧义。多线程

好比当前线程比较成功后，准备更新共享变量值的时候，这个共享变量值被其余线程更改了，那么CAS函数必须返回false。

要实现这个需求，java中提供了Unsafe类，它提供了三个函数，分别用来操做基本类型int和long，以及引用类型Object。并发

public final native boolean compareAndSwapObject(Object obj, long valueOffset, Object expect, Object update);

    public final native boolean compareAndSwapInt(Object obj, long valueOffset, int expect, int update);

    public final native boolean compareAndSwapLong(Object obj, long valueOffset, long expect, long update);

参数的意义：框架

obj 和 valueOffset：表示这个共享变量的内存地址。这个共享变量是obj对象的一个成员属性，valueOffset表示这个共享变量在obj类中的内存偏移量。因此经过这两个参数就能够直接在内存中修改和读取共享变量值。
expect: 表示预期原来的值。
update: 表示期待更新的值。

接下来咱们来看看Java并发框架下的atomic包是如何使用CAS的。ide

三. JUC并发框架下的原子类(atomic)

调用JUC并发框架下原子类的方法时，不须要考虑多线程问题。那么咱们分析它是怎么解决多线程问题的。以AtomicInteger类为例。

3.1 成员变量

// 经过它来实现CAS操做的。由于是int类型，因此调用它的compareAndSwapInt方法
    private static final Unsafe unsafe = Unsafe.getUnsafe();

    // value这个共享变量在AtomicInteger对象上内存偏移量，
    // 经过它直接在内存中修改value的值，compareAndSwapInt方法中须要这个参数
    private static final long valueOffset;

    // 经过静态代码块，在AtomicInteger类加载时就会调用
    static {
        try {
            // 经过unsafe类，获取value变量在AtomicInteger对象上内存偏移量
            valueOffset = unsafe.objectFieldOffset
                (AtomicInteger.class.getDeclaredField("value"));
        } catch (Exception ex) { throw new Error(ex); }
    }

    // 共享变量，AtomicInteger就保证了对它多线程操做的安全性。
    // 使用volatile修饰，解决了可见性和有序性问题。
    private volatile int value;

有三个重要的属性：

unsafe: 经过它实现CAS操做，由于共享变量是int类型，因此调用compareAndSwapInt方法。
valueOffset: 共享变量value在AtomicInteger对象上内存偏移量
value: 共享变量，使用volatile修饰，解决了可见性和有序性问题。

3.2 重要方法

3.2.1 get与set方法

// 直接读取。由于是volatile关键子修饰的，老是能看到(任意线程)对这个volatile变量最新的写入
    public final int get() {
        return value;
    }

    // 直接写入。由于是volatile关键子修饰的，因此它修改value变量也会当即被别的线程读取到。
    public final void set(int newValue) {
        value = newValue;
    }

由于value变量是volatile关键字修饰的，它老是能读取（任意线程）对这个volatile变量最新的写入。它修改value变量也会当即被别的线程读取到。

3.2.2 compareAndSet方法

// 若是value变量的当前值(内存值)等于指望值(expect)，那么就把update赋值给value变量，返回true。
    // 若是value变量的当前值(内存值)不等于指望值(expect)，就什么都不作，返回false。
    // 这个就是CAS操做，使用unsafe.compareAndSwapInt方法，保证整个操做过程的原子性
    public final boolean compareAndSet(int expect, int update) {
        return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
    }

经过调用unsafe的compareAndSwapInt方法实现CAS函数的。可是CAS函数只能保证比较并交换操做的原子性，可是更新操做并不必定会执行。好比咱们想让共享变量value自增。
共享变量value自增是三个操做，1.读取value值，2.计算value+1的值，3.将value+1的值赋值给value。分析这三个操做：

读取value值,由于value变量是volatile关键字修饰的，可以读取到任意线程对它最后一次修改的值，因此没问题。
计算value+1的值：这个时候就有问题了，可能在计算这个值的时候，其余线程更改了value值，由于没有加同步锁，因此其余线程能够更改value值。
将value+1的值赋值给value: 使用CAS函数，若是返回false，说明在当前线程读取value值到调用CAS函数方法前，共享变量被其余线程修改了，那么value+1的结果值就不是咱们想要的了，由于要从新计算。

3.2.3 getAndAddInt方法

public final int getAndAddInt(Object obj, long valueOffset, int var) {
        int expect;
        // 利用循环，直到更新成功才跳出循环。
        do {
            // 获取value的最新值
            expect = this.getIntVolatile(obj, valueOffset);
            // expect + var表示须要更新的值，若是compareAndSwapInt返回false，说明value值被其余线程更改了。
            // 那么就循环重试，再次获取value最新值expect，而后再计算须要更新的值expect + var。直到更新成功
        } while(!this.compareAndSwapInt(obj, valueOffset, expect, expect + var));

        // 返回当前线程在更改value成功后的，value变量原先值。并非更改后的值
        return expect;
    }

这个方法在Unsafe类中，利用do_while循环，先利用当前值，计算更新值，而后经过compareAndSwapInt方法设置value变量，若是compareAndSwapInt方法返回失败，表示value变量的值被别的线程更改了，因此循环获取value变量最新值，再经过compareAndSwapInt方法设置value变量。直到设置成功。跳出循环，返回更新前的值。

// 将value的值当前值的基础上加1，并返回当前值
    public final int getAndIncrement() {
        return unsafe.getAndAddInt(this, valueOffset, 1);
    }

    // 将value的值当前值的基础上加-1，并返回当前值
    public final int getAndDecrement() {
        return unsafe.getAndAddInt(this, valueOffset, -1);
    }

   
    // 将value的值当前值的基础上加delta，并返回当前值
    public final int getAndAdd(int delta) {
        return unsafe.getAndAddInt(this, valueOffset, delta);
    }

    
    // 将value的值当前值的基础上加1，并返回更新后的值(即当前值加1)
    public final int incrementAndGet() {
        return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
    }

    // 将value的值当前值的基础上加-1，并返回更新后的值(即当前值加-1)
    public final int decrementAndGet() {
        return unsafe.getAndAddInt(this, valueOffset, -1) - 1;
    }

    // 将value的值当前值的基础上加delta，并返回更新后的值(即当前值加delta)
    public final int addAndGet(int delta) {
        return unsafe.getAndAddInt(this, valueOffset, delta) + delta;
    }

都是利用unsafe.getAndAddInt方法实现的。

四.重要示例

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

class Data {
    AtomicInteger num;

    public Data(int num) {
        this.num = new AtomicInteger(num);
    }

    public int getAndDecrement() {
        return num.getAndDecrement();
    }
}

class MyRun implements Runnable {

    private Data data;
    /**
     * 用来记录全部卖出票的编号
     */
    private List<Integer> list;
    private CountDownLatch latch;

    public MyRun(Data data, List<Integer> list, CountDownLatch latch) {
        this.data = data;
        this.list = list;
        this.latch = latch;
    }

    @Override
    public void run() {
        try {
            action();
        }  finally {
            // 释放latch共享锁
            latch.countDown();
        }
    }

    /**
     * 进行买票操做，注意这里没有使用data.num>0做为判断条件，直到卖完线程退出。
     * 那么作会致使这两处使用了共享变量data.num，那么作多线程同步时，就要考虑更多条件。
     * 这里只for循环了5次，表示每一个线程只卖5张票，并将全部卖出去编号存入list集合中。
     */
    public void action() {
        for (int i = 0; i < 5; i++) {
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            int newNum = data.getAndDecrement();

            System.out.println("线程"+Thread.currentThread().getName()+"  num=="+newNum);
            list.add(newNum);
        }
    }
}

public class ThreadTest {
    public static void startThread(Data data, String name, List<Integer> list,CountDownLatch latch) {
        Thread t = new Thread(new MyRun(data, list, latch), name);
        t.start();
    }

    public static void main(String[] args) {
        // 使用CountDownLatch来让主线程等待子线程都执行完毕时，才结束
        CountDownLatch latch = new CountDownLatch(6);

        long start = System.currentTimeMillis();
        // 这里用并发list集合
        List<Integer> list = new CopyOnWriteArrayList<>();
        Data data = new Data(30);
        startThread(data, "t1", list, latch);
        startThread(data, "t2", list, latch);
        startThread(data, "t3", list, latch);
        startThread(data, "t4", list, latch);
        startThread(data, "t5", list, latch);
        startThread(data, "t6", list, latch);

        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 处理一下list集合，进行排序和翻转
        Collections.sort(list);
        Collections.reverse(list);
        System.out.println(list);

        long time = System.currentTimeMillis() - start;
        // 输出一共花费的时间
        System.out.println("\n主线程结束 time=="+time);
    }
}

结果输出

线程t2  num==30
线程t1  num==25
线程t5  num==29
线程t6  num==26
线程t4  num==28
线程t3  num==27
线程t4  num==24
线程t2  num==19
线程t1  num==20
线程t3  num==22
线程t5  num==21
线程t6  num==23
线程t5  num==17
线程t1  num==14
线程t6  num==13
线程t3  num==15
线程t2  num==18
线程t4  num==16
线程t4  num==10
线程t1  num==7
线程t6  num==12
线程t3  num==8
线程t2  num==9
线程t5  num==11
线程t5  num==6
线程t1  num==1
线程t6  num==2
线程t2  num==3
线程t4  num==4
线程t3  num==5
[30, 29, 28, 27, 26, 25, 24, 23, 22, 21, 20, 19, 18, 17, 16, 15, 14, 13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1]

主线程结束 time==57

咱们使用AtomicInteger，代替同步锁来解决多线程安全的。