Java并发基石CAS原理以及ABA问题

java1234

共 31311字,需浏览 63分钟

 ·

2021-04-29 17:31

点击上方蓝色字体,选择“标星公众号”

优质文章,第一时间送达

76套java从入门到精通实战课程分享

在学习CAS之前,先从一个简单的案例入手,进而引出CAS的基本使用:

1、基于CAS的网站计数器

需求:

我们开发一个网站,需要对访问量进行统计,用户每发送一次请求,访问量+1,如何实现?

我们模拟有100个人同时访问,并且每个人对咱们的网站发起10次请求,最后总访问次数应该是1000次。


1.1 网站访客统计Demo

代码如下:

public class Demo {
    // 网站总访问量
    static int count = 0;

    // 模拟用户访问的方法
    public static void request() throws InterruptedException {
        // 模拟耗时5毫秒
        TimeUnit.MILLISECONDS.sleep(5);
        
        // 访问量++
        count ++;// 这里 count 并不是原子的
    }

    public static void main(String[] args) throws InterruptedException {
        // 开始时间
        long startTime = System.currentTimeMillis();
        // 最大线程数100,模拟100个用户同时访问
        int threadSize = 100;
        //
        CountDownLatch countDownLatch = new CountDownLatch(threadSize);

        for(int i = 0; i < threadSize; i++) {

            Thread thread = new Thread(new Runnable() {
                @Override
                public void run() {
                    // 模拟用户行为,每个用户访问10次网站
                    try {
                        for(int j = 0; j < 10; j++) {
                            request();
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        countDownLatch.countDown();
                    }
                }
            });

            thread.start();
        }
        // 怎么保证100个线程结束之后,再执行后面代码?
        countDownLatch.await();
        // 100个线程执行结束时间
        long endTime = System.currentTimeMillis();

        System.out.println(Thread.currentThread().getName() + ",耗时:" + (endTime - startTime) + ", count = " + count);
    }
}

这里先对CountDownLtch 做一个简单介绍,之后会更新一篇它的源码分析。


CountDownLatch的概念

  • CountDownLatch是一个同步工具类,用来协调多个线程之间的同步,或者说起到线程之间的通信(而不是用作互斥的作用)。

  • CountDownLatch能够使一个线程在等待另外一些线程完成各自工作之后,再继续执行。使用一个计数器进行实现。计数器初始值为线程的数量。当每一个线程完成自己任务后,计数器的值就会减一。当计数器的值为0时,表示所有的线程都已经完成了任务,然后在CountDownLatch上等待的线程就可以恢复执行任务。

CountDownLatch的用法

  • CountDownLatch典型用法1:某一线程在开始运行前等待n个线程执行完毕。将CountDownLatch的计数器初始化为n —> new CountDownLatch(n) ,每当一个任务线程执行完毕,就将计数器减1 —> countdownlatch.countDown(),当计数器的值变为0时,在CountDownLatch上 await() 之后的线程就会被唤醒。一个典型应用场景就是启动一个服务时,主线程需要等待多个组件加载完毕,之后再继续执行。

  • CountDownLatch典型用法2:实现多个线程开始执行任务的最大并行性。注意是并行性,不是并发,强调的是多个线程在某一时刻同时开始执行。类似于赛跑,将多个线程放到起点,等待发令枪响,然后同时开跑。做法是初始化一个共享的CountDownLatch(1),将其计数器初始化为1,多个线程在开始执行任务前首先 coundownlatch.await(),当主线程调用 countDown() 时,计数器变为0,多个线程同时被唤醒。

上面案例代码的执行结果如下图:

如图所示,我们理论上应该是100个线程模拟用户,每个线程模拟访问10次,最终结果count 应该是1000才对,但是无论几次测试,最终count 都是达不到1000的。


原因分析:

/**
 * Q:分析一下问题出在哪呢?
 * A:count ++ 操作实际上是由3步来完成!(jvm执行引擎)
 *    1.获取count的值,记做A : A=count
 *    2.将A值+1,得到B :B=A+1
 *    3.将B值赋值给count
 *
 *    如果有A.B两个线程同时执行count++,他们通知执行到上面步骤的第一步,得到的
 *    count是一样的,3步操作结束后,count只加了1,导致count结果不正确!
 * Q:怎么解决结果不正确问题?
 * A:对count++操作的时候,我们让多个线程排队处理,多个线程同时到达request()方法的时候,
 * 只能允许一个线程可以进去操作,其它的线程在外面等着,等里面的处理完毕出来之后,外面等着的
 * 再进去一个,这样操作的count++就是排队进行的,结果一定是正确的。
 *
 * Q:怎么实现排队效果??
 * A:java中synchronized关键字和ReentrantLock都可以实现对资源枷锁,保证并发正确性,
 * 多线程的情况下可以保证被锁住的资源被“串行”访问。
 */

1.2 使用synchronized关键字改进Demo案例

改进代码如下:

public class Demo {
    // 网站总访问量
    static int count = 0;

    // 模拟用户访问的方法(加synchronized修饰)
    public synchronized static void request() throws InterruptedException {
        // 模拟耗时5毫秒
        TimeUnit.MILLISECONDS.sleep(5);
        
        // 访问量++
        count ++;
    }

    public static void main(String[] args) throws InterruptedException {
        // 开始时间
        long startTime = System.currentTimeMillis();
        // 最大线程数100,模拟100个用户同时访问
        int threadSize = 100;
        CountDownLatch countDownLatch = new CountDownLatch(threadSize);

        for(int i = 0; i < threadSize; i++) {

            Thread thread = new Thread(new Runnable() {
                @Override
                public void run() {
                    // 模拟用户行为,每个用户访问10次网站
                    try {
                        for(int j = 0; j < 10; j++) {
                            request();
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        countDownLatch.countDown();
                    }
                }
            });

            thread.start();
        }
        // 怎么保证100个线程结束之后,再执行后面代码?
        countDownLatch.await();
        // 100个线程执行结束时间
        long endTime = System.currentTimeMillis();

        System.out.println(Thread.currentThread().getName() + ",耗时:" + (endTime - startTime) + ", count = " + count);
    }
}

执行代码运行结果如下:

可以看出,我们要得到总访问量1000结果正确,但是当我们把synchronized关键字加在了request()方法上,由于锁住了方法,导致相比于不加锁时,线程执行效率严重降低!

/**
 * Q:耗时太长的原因是什么呢?
 * A:程序中的request方法使用synchronized关键字修饰,保证了并发情况下,request方法同一时刻
 * 只允许一个线程进入,request加锁相当于串行执行了,count的结果和我们预期的一致,但是耗时太长了..
 *
 * Q:如何解决耗时长的问题?
 * A:count ++ 操作实际上是由3步来完成!(jvm执行引擎)
 *    1.获取count的值,记做A : A=count
 *    2.将A值+1,得到B :B=A+1
 *    3.将B值赋值给count
 *    升级第3步的实现(只把锁加到第3步上,缩小加锁的范围):
 *       1.获取锁 
 *       2.获取以下count最新的值,记做LV
 *       3.判断LV是否等于A,如果相等,则将B的值赋值给count,并返回true,否则返回false
 *       4.释放锁
 */

1.3 缩小加锁范围再次改进Demo案例

代码如下:

public class Demo03 {
    // 网站总访问量:volatile保证线程可见性,便于在下面逻辑中 -> 保证多线程之间每次获取到的count是最新值
    volatile static int count = 0;

    // 模拟访问的方法
    public static void request() throws InterruptedException {
        // 模拟耗时5毫秒
        TimeUnit.MILLISECONDS.sleep(5);

        //count ++;

        int expectCount; // 表示期望值
        // 比较并交换
        while (!compareAndSwap((expectCount = getCount()), expectCount + 1)) {
        }
    }

    /**
     * 比较并交换
     *
     * @param expectCount 期望值count
     * @param newCount    需要给count赋值的新值
     * @return 成功返回 true 失败返回false
     */
    public static synchronized boolean compareAndSwap(int expectCount, int newCount) {
        // 判断count当前值是否和期望值expectCount一致,如果一致 将newCount赋值给count
        if (getCount() == expectCount) {
            count = newCount;
            return true;
        }
        return false;
    }

    public static int getCount() {
        return count;
    }

    public static void main(String[] args) throws InterruptedException {
        // 开始时间
        long startTime = System.currentTimeMillis();
        int threadSize = 100;
        CountDownLatch countDownLatch = new CountDownLatch(threadSize);

        for (int i = 0; i < threadSize; i++) {

            Thread thread = new Thread(new Runnable() {
                @Override
                public void run() {
                    // 模拟用户行为,每个用户访问10次网站
                    try {
                        for (int j = 0; j < 10; j++) {
                            request();
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        countDownLatch.countDown();
                    }
                }
            });

            thread.start();
        }
        // 保证100个线程 结束之后,再执行后面代码
        countDownLatch.await();
        long endTime = System.currentTimeMillis();

        System.out.println(Thread.currentThread().getName() + ",耗时:" + (endTime - startTime) + ", count = " + count);
    }
}

执行结果如下:

可以看到这种方式下,不仅可以达到期望的网站访问量结果,效率也很高!


这种比较并交换,且线程安全的方式就可以称作CAS:

/**
 * 比较并交换
 *
 * @param expectCount 期望值count
 * @param newCount    需要给count赋值的新值
 * @return 成功返回 true 失败返回false
 */
public static synchronized boolean compareAndSwap(int expectCount, int newCount) {
    // 判断count当前值是否和期望值expectCount一致,如果一致 将newCount赋值给count
    if (getCount() == expectCount) {
        count = newCount;
        return true;
    }
    return false;
}

2、CAS介绍与实现原理

CAS 全称“CompareAndSwap”,中文翻译过来为“比较并交换”。

定义:

  • CAS操作包含三个操作数————内存位置(V)、期望值(A)和新值(B)。

  • 如果内存位置的值与期望值匹配,那么处理器会自动将该位置值更新为新值。否则,处理器不作任何操作。

  • 无论哪种情况,它都会在CAS指令之前返回该位置的值。(CAS在一些特殊情况下仅返回CAS是否成功,而不提取当前值)

  • CAS有效的说明了 “我认为位置V应该包含值A;如果包含该值,则将B放到这个位置;否则,不要更改

         该位置的值,只告诉我这个位置现在的值即可。”

2.1 Java中的CAS

java中提供了对CAS操作的支持,具体在sun.misc.unsafe类中,声明如下:

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

  • 参数var1:表示要操作的对象。

  • 参数var2:表示要操作对象中属性地址的偏移量。

  • 参数var4:表示需要修改数据的期望的值。

  • 参数var5:表示需要修改为的新值。

Java中的CAS通过调用JNI的代码实现,JNI:java Native Interface,允许java调用其它语言。而compareAndSwapXXX系列的方法就是借助C语言来调用cpu底层指令实现的。


以常用的Intel x86平台来说,最终映射到的cpu的指令为“cmpxchg”,这是一个原子指令,cpu执行此命令时,实现比较并替换的操作!


2.2 CAS也会出现一些问题

2.2.1 ABA问题(狸猫换太子)

CAS需要在操作值的时候检查下值有没有发生变化,如果没有发生变化则更新,但是如果一个值原来是A,在CAS方法执行之前,被其它线程修改为了B、然后又修改回了A,那么CAS方法执行检查的时候会发现它的值没有发生变化,但是实际却变化了。这就是CAS的ABA问题。


提示,使用程序模拟ABA:

public class CasABADemo {

    public static AtomicInteger a = new AtomicInteger(1);

    public static void main(String[] args) {
        Thread main = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("操作线程" + Thread.currentThread().getName() + ", 初始值:" + a.get());
                try {

                    int expectNum = a.get();
                    int newNum = expectNum + 1;
                    Thread.sleep(1000);// 主线程休眠一秒钟,让出cpu
     
                    // CAS比较并交换
                    boolean isCASSccuess = a.compareAndSet(expectNum, newNum);
                    System.out.println("操作线程" + Thread.currentThread().getName() + ",CAS操作:" + isCASSccuess);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, "主线程");

        Thread other = new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(20);// 确保Thread-main线程优先执行

                    a.incrementAndGet();// a + 1,a=2
                    System.out.println("操作线程" + Thread.currentThread().getName() + ",【increment】,值=" +a.get());
                    a.decrementAndGet();// a - 1,a=1
                    System.out.println("操作线程" + Thread.currentThread().getName() + ",【decrement】,值=" +a.get());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, "干扰线程");

        main.start();
        other.start();
    }
}

输出结果如下:

我们看到结果中,在主线程“比较并交换之前”,干扰线程先是将a的值改成2,然后又重新改回1,之后才执行主线程的CAS!


2.2.2 如何解决ABA问题?

解决ABA最简单的方案就是给值加一个修改版本号,每次值变化,都会修改它的版本号,CAS操作时都去对比此版本号。


java中ABA解决方法(AtomicStampedReference),这种方式类似于乐观锁,即:通过当前版本号来控制CAS交换,如果当前版本号与期望版本号相等,才能交换,否则不可以交换,每执行一次交换当前版本号就+1。


AtomicStampedReference主要包含一个对象引用及一个可以自动更新的整数 stamp 版本号 的 Pair 对象来解决ABA问题。


[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-KziroKif-1619351907827)(Java并发基石-CAS原理.assets/image-20210425194204352.png)]


AtomicStampedReference中的compareAndSet()方法:

/**
 * Atomically sets the value of both the reference and stamp
 * to the given update values if the
 * current reference is {@code ==} to the expected reference
 * and the current stamp is equal to the expected stamp.
 *
 * @param expectedReference the expected value of the reference     期望值的引用
 * @param newReference the new value for the reference              新值的引用
 * @param expectedStamp the expected value of the stamp             期望引用的版本号
 * @param newStamp the new value for the stamp                      新值的版本号
 * @return {@code trueif successful
 */
public boolean compareAndSet(V   expectedReference,// 期望值的引用
                             V   newReference,// 新值的引用
                             int expectedStamp,// 期望引用的版本号
                             int newStamp) {// 新值的版本号
    Pair<V> current = pair;
    return
        expectedReference == current.reference && // 期望引用与当前引用一致
        expectedStamp == current.stamp &&         // 期望版本号与当前版本号一致
        ((newReference == current.reference &&    
          newStamp == current.stamp) ||
         casPair(current, Pair.of(newReference, newStamp)));
}

private boolean casPair(Pair<V> cmp, Pair<V> val) {
    return UNSAFE.compareAndSwapObject(this, pairOffset, cmp, val);
}

使用AtomicStampedReference解决ABA问题:

public class CasABADemo02 {
    
    public static AtomicStampedReference<Integer> a = new AtomicStampedReference(new Integer(1), 1);

    public static void main(String[] args) {
        Thread main = new Thread(() -> {
            System.out.println("操作线程" + Thread.currentThread().getName() + ", 初始值:" + a.getReference());
            try {
                Integer expectReference = a.getReference();
                Integer newReference = expectReference + 1;
                Integer expectStamp = a.getStamp();
                Integer newStamp = expectStamp + 1;

                Thread.sleep(1000);// 主线程休眠一秒钟,让出cpu
                // AtomicStampedReference下的compareAndSet来解决ABA问题
                boolean isCASSccuess = a.compareAndSet(expectReference, newReference, expectStamp, newStamp);
                System.out.println("操作线程" + Thread.currentThread().getName() + ",CAS操作:" + isCASSccuess);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "主线程");

        Thread other = new Thread(() -> {
            try {
                Thread.sleep(20);// 确保Thread-main线程优先执行

                a.compareAndSet(a.getReference(), (a.getReference() + 1), a.getStamp(), (a.getStamp() + 1));
                System.out.println("操作线程" + Thread.currentThread().getName() + ",【increment】,值=" + a.getReference());
                a.compareAndSet(a.getReference(), (a.getReference() - 1), a.getStamp(), (a.getStamp() + 1));
                System.out.println("操作线程" + Thread.currentThread().getName() + ",【decrement】,值=" + a.getReference());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "干扰线程");

        main.start();
        other.start();
    }
}

运行结果如图:

这时就解决了ABA问题,如果主线程执行CAS操作前,出现狸猫换太子的情况,那么这时候就不能进行比较并交换!

————————————————

版权声明:本文为CSDN博主「兴趣使然的草帽路飞」的原创文章,遵循CC 4.0 BY-SA版权协议,转载请附上原文出处链接及本声明。

原文链接:

https://blog.csdn.net/weixin_43591980/article/details/116136920






粉丝福利:Java从入门到入土学习路线图

👇👇👇

👆长按上方微信二维码 2 秒


感谢点赞支持下哈 

浏览 32
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报
评论
图片
表情
推荐
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报