基础 | 并发编程 - [CAS & 原子类]

§1 CAS

CAS 即 Compare and Swap
用于判断内存某个位置的值是否为预期值，如果是 则更改为新的值
属于系统原语，CPU 原子指令(cmpxchg)，硬件级同步
执行过程中不允许被中断，原子性通过线程独占实现，不会造成所谓的数据不一致问题

CAS 涉及三个值：

• 内存值，是当前内存中存在的值
• 期望值，是工作线程修改内存值时内存的原始值
• 交换值，是工作线程希望将此内存值变为的值

CAS 工作流程：

• 主内存中值 x=1 时，工作线程从主内存复制 x
• 工作线程修改 x 的值为 5
• 工作线程希望讲 x=5 写回主内存
• 只有从 x 进入工作线程至 x 回写的整个过程中无其他线程修改过 x ，回写 x 才是安全的
• 因此使 比较 主内存中值 x 与 x 的原始值 1，以确认 x 是否被修改
• 如果比较结果返回 true，则 交换 主内存中 x 的值 1 为 工作内存中新值 5

§2 CAS 原理

CAS 的原理，直白说就是保证在一个操作期间，被修改的值被这个操作 线程独占
因为是独占的，所以没有线程安全问题
这主要依赖 自旋锁 和 Unsafe 类 实现

§2.1 CAS 原理 - 自旋锁

见 基础 | 并发编程 - [锁] #自旋锁

§2.2 CAS 原理 - Unsafe 类

Unsafe 类是CAS 核心类，在 rt.jar 中 sun.misc 包下
Unsafe 类的所有方法都是 native 的，此类可以帮助 java 按操作指针的方式直接操作内存，即直接操作特定内存的数据

以 new AtomicInteger().getAndIncrement() 为例

// 为 AtomicInteger 声明 Unsafe 实例，用于其中的更新操作
private static final Unsafe unsafe = Unsafe.getUnsafe();
// AtomicInteger 中 value 字段在字节码中的偏移量
private static final long valueOffset;

// 类加载阶段就尝试获取 valueOffset
static {
    try {
        valueOffset = unsafe.objectFieldOffset
            (AtomicInteger.class.getDeclaredField("value"));
    } catch (Exception ex) { throw new Error(ex); }
}
// valueOffset 对应的就是这个 value，此值就是 integer 的值
// volatile 关键字用在这里解决了可见性，后面在用 unsafe 解决原子性
private volatile int value;

/**
 * Atomically increments by one the current value.
 *
 * @return the previous value
 */
public final int getAndIncrement() {
	// 当前对象 this 的 valueoffset 对应的字段（就是value）,加 1
    return unsafe.getAndAddInt(this, valueOffset, 1);
}
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

Unsafe.getAndAddInt()
代码如下，为了理解方便替换了变量标识符

public final int getAndAddInt(Object obj, long offset, int addValue) {
    int currentValue;
    do {
        currentValue = this.getIntVolatile(obj, offset);
    } while(!this.compareAndSwapInt(obj, offset, currentValue , currentValue + addValue));

    return currentValue ;
}
1
2
3
4
5
6
7
8

这里的代码使用了一个自旋锁，只不过获取锁与获取锁后的动作在一个操作里都完成了

this.compareAndSwapInt() 包含三步

• 获取现在的值
• 与以前的值比较，相同代表刚刚获取了值没变，则没有线程进入，此时改值是安全的
• 如果成功就赋予新值

并且这三步是原子的，相当于加锁和加锁后的操作二合一了
此方法会尝试加锁并在成功后赋予新值，如果失败，就需要重新尝试，且重试之前获取新当前值

§2.3 CAS 缺点

• 长时间循环时会导致 CPU 效率变低，高并发下收益不高，因为一次自旋最多完成一个线程的修改
  因为底层是特殊的自旋锁
• 只能保证一个共享变量的原子操作
  因为底层依赖的是 CPU 原语，只支持单变量
• ABA 问题

ABA 问题
CAS 在使用时实际上有两次取值

• 第一次是在 CAS 外，取出的值作为 期望值 使用
• 第二次是在 CAS 中，取出的值作为 内存值 去和期望值进行比较

上面两次取值如果一致，认为（注意仅仅是认为）这期间没有其他线程操作做这个值

但是，若另一个线程在这期间操作过这个值，甚至多次操作，但最后一次操作将值改回了原来的值
这就是 ABA 问题，CAS 操作中前后两次取值经比较一致，但中间值实际上被修改过

对于正在通过 CAS 修改值的线程，只对修改值这个操作本身， ABA 问题是没有什么危害的
但在复杂业务场景下，ABA 问题会导致线程忽略已发生的事件，进而忽略应有的逻辑

ABA 危害示例
设想有一个 B2B 业务群，下设如下服务
有一库存服务 stock，有一库房服务 wms ，有一运单服务 trans，有一计费服务 charge
订单先流入 stock ，经过验证后回传订单，此时订单生效，对应计费信息下发 charge
订单再流入 wms，产生仓储服务费，计费信息下发 charge
最后订单流入 trans，产生运费，计费信息下发 charge
上述计费信息都到达 charge 后，charge 经过计费规则生成账单，并归账

假设，业务要求，charge 是实时结算的，
只有完全结算完成并归账，才允许有下一次库存变化，否则就可能出问题 （可能是系统问题，也可能是线下问题）
库存变化可能来自订单、采购入库、退供入库

问题流程如下

• 在一个时间点 t，整个系统是结算完成的，可以接受订单
• 线程 1 接单，流入 stock ，查询库存 10
• 中间线程 1 经过 rpc 与其他服务通信，获取仅仅是推送信息给中间件，网络问题，卡了
• 线程 2 接单，流入 stock 并扣减库存，流入 wms ，流入 trans（有些商品确实可以很快，比如数字商品）
• 线程 2 突发退单，假设这就是个数字商品，退单流程可以共用销售流程不用另行发起
• 线程 2 退 trans，退 wms ，退 stock，库存又是 10 了，但是计费信息已经给 charge 了
• charge 还没有结算完成，按理说不应接单
• 线程 1 的 rpc 回来了，比较库存，一致，所以下单成功
• 在 charge 没有结算完成并归档时，通过 stock 接单成功了

ABA 问题的解决
通过原子时间戳(AtomicStemptReference)解决 ABA 问题
注意，下面的例子中，线程 A 只是用来模拟其他线程多次修改了原子时间戳，并最终改了回去，因此不甚符合实际开发写法

static AtomicStampedReference<Integer> atomicStampedReference = new AtomicStampedReference<>(100,1);

public static void main(String[] args) {
    new Thread(()->{
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        atomicStampedReference.compareAndSet(100,101,atomicStampedReference.getStamp(),atomicStampedReference.getStamp()+1);
        atomicStampedReference.compareAndSet(101,100,atomicStampedReference.getStamp(),atomicStampedReference.getStamp()+1);
    },"A").start();
    new Thread(()->{
        int stamp = atomicStampedReference.getStamp();
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        boolean a = atomicStampedReference.compareAndSet(100,2020,stamp,stamp+1);
        System.out.println(a);
    },"B").start();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23