摘要

简单记录一下CAS相关的基础知识。

正文

指路

注意：单核CPU(无超线程技术)依然存在线程安全问题。原因是如果任务耗时较长，通常会有多个时间片执行，就是由于多个时间片的原因，会导致线程安全问题。

具体细节，需详细学习操作系统。

一、背景

统计用户访问量。用户每发一次请求，访问量+1。

需要模拟100人同时访问，每个人进行10次请求。最后总访问次数应该是1000次。

如果不使用原子变量，示例代码如下。

java 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

/**
 * 统计用户访问量。用户每发一次请求，访问量+1。
 * <p>
 * 需要模拟100人同时访问，每个人进行10次请求。最后总访问次数应该是1000次。
 *
 * @author chenchuancheng github.com/meethigher
 * @since 2023/3/30 20:52
 */
public class Demo01 {

    /**
     * 总访问量
     */
    //private static AtomicInteger count = new AtomicInteger(0);
    private static int count = 0;

    private static void request() throws InterruptedException {
        //模拟耗时5毫秒
        TimeUnit.MILLISECONDS.sleep(5);
        //count.getAndIncrement();
        count++;
    }

    public static void main(String[] args) throws InterruptedException {
        long startTime = System.currentTimeMillis();
        int threadSize = 100;

        CountDownLatch countDownLatch = new CountDownLatch(100);
        for (int i = 0; i < threadSize; i++) {
            new Thread(() -> {
                //模拟用户行为：访问10次网站
                for (int j = 0; j < 10; j++) {
                    try {
                        request();
                    } catch (InterruptedException ignore) {
                    }
                }
                countDownLatch.countDown();
            }).start();
        }

        //等待所有模拟用户执行完
        countDownLatch.await();
        long endTime = System.currentTimeMillis();
        long d = endTime - startTime;
        System.out.println(Thread.currentThread().getName() + ", 耗时" + d + "毫秒, 总访问量" + count);

    }
}

并没有达到预期的1000，分析原因：count++实际上由jvm通过3步来完成

  1. 获取count的值，记做A：A=count
  2. 将A值+1，得到B：B=A+1
  3. 将B值赋值给count

如果有T1和T2两个线程，同时执行count++。当同时执行到第1步时，得到的A是一样的，3步操作结束后，就会导致实际count只加了1，这就是常说的线程安全的问题。

Q: 如何解决结果不正确的问题？

A: 加锁。保证当多个线程同时到达request方法的时候，只能允许一个线程可以进去操作，实现串行访问。Java中的synchronized关键字和ReentrantLock可以实现该效果

二、CAS

2.1 小试CAS(Java层面)

使用synchronized解决

java 
1
2
3
4
5
6

private synchronized static void request() throws InterruptedException {
    //模拟耗时5毫秒
    TimeUnit.MILLISECONDS.sleep(5);
    //count.getAndIncrement();
    count++;
}

但是发现效率变低了，因为串行执行时，每次都睡眠了5毫秒。我们针对不存在线程安全的问题，不需要加锁，修改为以下代码，会发现效率大大提高。

java 
1
2
3
4
5
6
7
8

private static void request() throws InterruptedException {
    //模拟耗时5毫秒
    TimeUnit.MILLISECONDS.sleep(5);
    //count.getAndIncrement();
    synchronized (Demo02.class) {
        count++;
    }
}

这个做法相当于是锁住了count++，也就是锁住了count++的三个步骤。为了再次提高效率，我们可以仅在第三步赋值时加锁，来保证数据的正确

  1. 获取count的值，记做A：A=count
  2. 将A值+1，得到B：B=A+1
  3. 获取锁
  4. 获取count最新的值，记做C
  5. 判断C是否等于A，如果相等，则将B赋值给count，并返回true，否则返回false，继续循环5这一步。
  6. 释放锁

3-6这几个步骤就叫做Compare And Swap，即Java中的CAS。

这个过程也叫做CAS自旋，说白了就是循环直到成功。

java 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

public class Demo03 {

    /**
     * 总访问量
     */
    private volatile static int count = 0;

    private static void request() throws InterruptedException {
        //模拟耗时5毫秒
        TimeUnit.MILLISECONDS.sleep(5);
        int expect;
        while (!compareAndSwap((expect = getCount()), expect + 1)) {
        }
    }

    private static int getCount() {
        return count;
    }

    /**
     * @param expect   期望值
     * @param newValue 新值
     * @return 成功true，失败false
     */
    private static synchronized boolean compareAndSwap(int expect, int newValue) {
        if (getCount() == expect) {
            count = newValue;
            return true;
        } else {
            return false;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        long startTime = System.currentTimeMillis();
        int threadSize = 100;

        CountDownLatch countDownLatch = new CountDownLatch(100);
        for (int i = 0; i < threadSize; i++) {
            new Thread(() -> {
                //模拟用户行为：访问10次网站
                for (int j = 0; j < 10; j++) {
                    try {
                        request();
                    } catch (InterruptedException ignore) {
                    }
                }
                countDownLatch.countDown();
            }).start();
        }

        //等待所有模拟用户执行完
        countDownLatch.await();
        long endTime = System.currentTimeMillis();
        long d = endTime - startTime;
        System.out.println(Thread.currentThread().getName() + ", 耗时" + d + "毫秒, 总访问量" + count);

    }
}

2.2 五问理解CAS(系统层面)

CAS 全称 CompareAndSwap，中文翻译过来为 比较与替换。

CAS操作包含三个操作数

  • 内存位置值（V）
  • 期望值（A）
  • 新值（B）

如果V与A匹配，那么处理器就会自动将V更新为B，否则处理器不做任何操作。

2.2.1 使用JDK提供的CAS

Q1：怎么使用JDK提供的CAS支持？

A1：Java中提供了对CAS操作的支持，具体在sun.misc.Unsafe类中，声明如下

首先，需要理解native关键字

其次，理解以下代码中变量含义

  • o：表示要操作的对象
  • offset：表示要操作对象中属性地址的偏移量
  • expected：表示需要修改的数据的期望值
  • x：表示需要修改的数据的新值
java 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

/**
 * Atomically update Java variable to <tt>x</tt> if it is currently
 * holding <tt>expected</tt>.
 * @return <tt>true</tt> if successful
 */
public final native boolean compareAndSwapObject(Object o, long offset,
                                                 Object expected,
                                                 Object x);

/**
 * Atomically update Java variable to <tt>x</tt> if it is currently
 * holding <tt>expected</tt>.
 * @return <tt>true</tt> if successful
 */
public final native boolean compareAndSwapInt(Object o, long offset,
                                              int expected,
                                              int x);

/**
 * Atomically update Java variable to <tt>x</tt> if it is currently
 * holding <tt>expected</tt>.
 * @return <tt>true</tt> if successful
 */
public final native boolean compareAndSwapLong(Object o, long offset,
                                               long expected,
                                               long x);

2.2.2 CAS原理

Q2：CAS实现原理是什么？

A2：CAS通过调用JNI的代码实现

JNI：Java Native Interface，允许Java调用其他语言。而compareAndSwapxxx系列方法就是借助C语言来调用CPU底层指定实现的。

以常用的Intel x86平台来说，CAS最终映射到CPU的指令是cmpxchg，即compare-exchange，这是一个原子指令。CPU执行此命令时，实现比较并替换的操作！

2.2.3 cmpxchg

Q3：cmpxchg如何保证多核下的线程安全？

A3：系统底层进行CAS操作的时候，会判断当前系统是否为多核系统。如果是，就给总线加锁（此时其他的CPU就无法进行运算了），只有一个线程会对总线加锁成功，加锁成功之后会执行CAS操作。也就是说CAS的原子性是操作系统级别的！2.1小试CAS的原子性是Java代码级别的。

2.2.4 CAS存在问题

Q4：CAS存在什么问题

A4：CAS存在问题如下

  1. 加锁带来的性能开销，要保证安全必然要牺牲效率。只是CAS是一种乐观锁，比常规synchronized悲观锁效率要高。
  2. ABA问题，这个比较好理解，就是一个值A，在CAS方法执行之前，被其他线程改为了B，又改回了A。那么CAS方法执行检查的时候，会发现他的值没有发生变化。这就是CAS的ABA问题。

看文字描述以及看图理解费劲，所以直接上代码。

java 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

@Slf4j
public class ABADemo {

    private final static AtomicInteger a = new AtomicInteger(1);

    public static void main(String[] args) throws Exception {
        new Thread(() -> {
            log.info("初始值：{}", a.get());
            try {
                int expectNum = a.get();
                int newNum = expectNum + 1;
                TimeUnit.SECONDS.sleep(1);
                //内部就是使用了cas
                boolean b = a.compareAndSet(expectNum, newNum);
                log.info("cas操作：{}, 最终值：{}", b, a.get());
            } catch (Exception ignore) {
            }
        }, "主线程").start();
        new Thread(() -> {
            try {
                //让主线程先获取值
                TimeUnit.MILLISECONDS.sleep(20);
                //开始搞事！++a
                a.incrementAndGet();
                log.info("进行++a操作，值：{}", a.get());
                //--a
                a.decrementAndGet();
                log.info("进行--a操作，值：{}", a.get());
            } catch (Exception ignore) {

            }
        }, "干扰线程").start();
        System.in.read();
    }
}

什么时候需要解决ABA问题？

我想了一下，像售票、金额这种的，哪怕存在ABA问题，也并不会出现数据混乱的问题。

但是像那种要求严格不变的，比如保险柜的机密文件，A线程去取机密文件时，另外B线程已经取出将信息窃取，又放回去了。这时候A获取的文件，已经不再机密，这时候的ABA问题是需要解决的。

2.2.5 解决ABA

Q5：如何解决ABA问题？

A5：大概有两种

  1. 不使用CAS，直接上synchronized。严格保证串行，但是性能会下降。
  2. 版本号。给值加一个版本号，每次值变化，都会修改它的版本号，CAS操作时都去对比此版本号。
    • 手动实现
    • 使用已有AtomicStampedReference实现，主要包含一个对象引用及一个可以自动更新的整数stamp的pair对象来解决ABA问题

AtomicStampedReference的CAS方法声明如下

变量含义

  • expectedReference：期望引用
  • newReference：新值引用
  • expectedStamp：期望引用的版本号
  • newStamp：新值版本号
java 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

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

使用AtomicStampedReference优化上述代码

java 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

@Slf4j
public class ABADemo02 {

    private final static AtomicStampedReference<Integer> a = new AtomicStampedReference<>(1, 1);

    public static void main(String[] args) throws Exception {
        new Thread(() -> {
            log.info("初始值：{}", a.getReference());
            try {
                int expectNum = a.getReference();
                int newNum = expectNum + 1;
                int expectStamp = a.getStamp();
                int newStamp=expectStamp+1;
                TimeUnit.SECONDS.sleep(1);
                //内部就是使用了cas
                boolean b = a.compareAndSet(expectNum, newNum,expectStamp,newStamp);
                log.info("cas操作：{}, 最终值：{}", b, a.getReference());
            } catch (Exception ignore) {
            }
        }, "主线程").start();
        new Thread(() -> {
            try {
                //让主线程先获取值
                TimeUnit.MILLISECONDS.sleep(20);
                //开始搞事！++a
                a.compareAndSet(a.getReference(),a.getReference()+1,a.getStamp(),a.getStamp()+1);
                log.info("进行++a操作，值：{}", a.getReference());
                //--a
                a.compareAndSet(a.getReference(),a.getReference()-1,a.getStamp(),a.getStamp()+1);
                log.info("进行--a操作，值：{}", a.getReference());
            } catch (Exception ignore) {

            }
        }, "干扰线程").start();
        System.in.read();
    }
}

三、参考致谢

乐观锁(CAS)和悲观锁(synchronized)的详细介绍_傻鱼爱编程的博客-CSDN博客

真实业务场景展现CAS原理的ABA问题及解决方案_cas的aba问题什么场景会出现问题_cauchy6317的博客-CSDN博客