当前位置：首页 > article >正文

并发编程中的CAS思想

article 2024/11/5 14:59:35

共享变量操作的原子性

分析如下代码片段：

// 获取共享变量时，为了保证该变量的可见性，需要使用 volatile 修饰。
static volatile int count = 0;

public static void add(){
    count++;
}

public static void main(String[] args) throws InterruptedException {

    // t1线程对count加5000次
    Thread t1 = new Thread(() -> {
        for (int i = 0; i < 5000; i++) {
            add();
        }
    });
    // t2线程对count加5000次
    Thread t2 = new Thread(() -> {
        for (int i = 0; i < 5000; i++) {
            add();
        }
    });

    t1.start();
    t2.start();
    // 确保t1、t2都完成
    t1.join();
    t2.join();

    System.out.println(count);  // 正常情况下应该为10000
}

出现线程安全的原因：count++ 包含了三个独立的操作：读取count的值，将值加1，然后将计算结果写入count。这是一个 “读取-修改-写入” 的操作序列

如果计数器的初始值为9，那么在某些情况下，每个线程读到的值都为9，接着执行递增操作，并且都将计数器的值设为10。显然，这并不是我们希望看到的情况，如果有一次递增操作丢失了，计数器的值就将偏差1

本质上线程可能会基于一种 可能失效的观察结果 来做出判断或者执行某个计算

解决思路：要递增一个计数器，你必须知道它之前的值，在将更新结果写回之前，确保其它线程没有修改过这个值

解决方式一：在某个线程修改该变量时，防止其他线程使用这个变量，从而确保其他线程只能在修改操作完成之前或之后读取和修改状态，而不是在修改状态的过程中（利用synchronized锁）

解决方式二：线程在计数器原始值基础上做修改操作，在要把修改后的值写回计数器前，判断此时计数器的值还是不是原始值了（利用CAS思想）

若不是，则说明在当前线程执行更新的过程中已经有其它线程修改了计数器，所以当前线程修改的结果值不能再写回。只能终止，再尝试下一轮的修改操作，直到成功

若是，则说明在当前线程执行更新的过程中没有其他线程修改过计数器（此处会产生ABA问题），所以当前线程可以将修改的值写回计数器，修改成功

CAS（比较和交换）本身不就是"先判断，后执行"的操作？如何保证CAS的原子性呢？——底层利用的是 cmpxchg 指令，该指令从底层硬件上确保了 “比较” 和 “交换” 这两个操作的原子性

方式一实现：为 add 方加上 synchronized 锁，缺点：性能较低，当一个线程获取锁后，其它线程只能无条件的阻塞，会伴随着线程的阻塞与唤醒：上下文切换（核心态 --> 用户态）的开销

方式二实现：使用原子类 AtomicInteger 变量，配合它的 getAndAdd 方法（CAS）来实现并发状态下的正确递增操作，即使操作失败，也没有放弃CPU执行权，不会阻塞。缺点：可能会出现 ABA 问题

乐观锁与悲观锁

悲观锁：总是认为不加锁，就一定会产生线程安全问题（不加锁，一定会出现问题），这样的话就会伴随线程的阻塞与唤醒：上下文切换（核心态 --> 用户态）的开销

乐观锁：CAS是一种无锁的同步机制（失败了无所谓，大不了重试），可以在不使用锁的情况下实现数据的同步和并发控制，优点：不会导致上下文切换，即使操作失败，也不是直接放弃CPU，而是继续重试

在这里插入图片描述

JUC包下的并发原子类

仿照 AtomicInteger 类，实现并发条件下共享变量递增的线程安全：

public class AtomicIntegerTest {

	static AtomicInteger count = new AtomicInteger(0);

	public static void increment() {
		while (true) {
			// 旧值
			int except = count.get();
			// 要更新的值
			int update = except + 1;
			// CAS
			if (count.compareAndSet(except, update))
				break;
		}
	}

	public static void main(String[] args) throws InterruptedException {

		Thread t1 = new Thread(() -> {
			for (int i = 0; i < 5000; i++) {
				increment();
			}
		}, "t1");

		Thread t2 = new Thread(() -> {
			for (int i = 0; i < 5000; i++) {
				increment();
			}
		}, "t2");

		t1.start();
		t2.start();
		// 确保t1、t2都执行完，看到的是最终count的值
		t1.join();
		t2.join();

		System.out.println(count.get());

	}

}