【Java核心-进阶】CAS（compare-and-swap）

CAS简介

CAS（compare-and-swap）是一种对数据进行原子性操作的技术。
它提供了一系列操作指令用于读取数值，或并发修改。
它是Java并发中所谓 “lock-free” 机制的基础。
CAS的底层依赖于CPU提供的指令。如，x86 CPU 的 cmpxchg

 

CAS使用方式

AtomicInteger

AtomicInteger使用了CAS技术，
它底层依赖于 Unsafe 对数据进行操作，
并使用 volatile 字段 value 记录数据，以保证可见性。

public class AtomicInteger extends Number implements java.io.Serializable {
  private static final jdk.internal.misc.Unsafe U = jdk.internal.misc.Unsafe.getUnsafe();
  private static final long VALUE = U.objectFieldOffset(AtomicInteger.class, "value");
  private volatile int value;

  public final int getAndIncrement() {
    return U.getAndAddInt(this, VALUE, 1);
  }
  ...
}

public final class Unsafe {
  public final int getAndAddInt(Object o, long offset, int delta) {
    int v;
    do {
      v = getIntVolatile(o, offset);
    } while (!weakCompareAndSetInt(o, offset, v, v+delta);
    return v;
  }
  ...
}

 

自身业务代码使用CAS技术

注：此处仅说明较为底层的一些CAS使用方式。在实际开发中我们可以借助JDK并发包中一些比较高级的辅助类来实现原子化操作。如，ReentrantLock、Semaphore 等。
 

我们主要利用CAS的原子性操作来实现排他性地更改数据。
如，通过对某个字段进行原子操作来实现加锁与释放锁。

public class AtomicDataBlock {
  private volatile long lock;
  void acquireLock() {}
  void releaseLock() {}
}

 

直接调用 Unsafe 通常是不合适的。
这个类太底层，过于抽象，也发生过新版JDK不兼容的问题（部分方法被删除，甚至命名空间更改）。
 

JDK Atomic 包中提供了一些常用的原子性数据类型及相关辅助类。这些是比较好的选择。
如：AtomicBoolean、AtomicLong、AtomicReference、LongAdder、AtomicLongFieldUpdater

private final AtomicLong lock;
void acquireLock() {
  long threadId = Thread.currentThread().getId();
  while (!lock.compareAndSet(0, threadId)) {
    // 等待其它线程完成操作，然后重试
    ...
  }
}

 

从 Java 9 开始，可以使用 VarHandle 处理：

private volatile long lock;
private static final VarHandle HANDLE = MethodHandles.lookup()
    .findStaticVarHandle(AtomicDataBlock.class, "lock");
private void acquireLock() {
  long threadId = Thread.currentThread().getId;
  while (!HANDLE.compareAndSet(this, 0, threadId)) {
    // 等待其它线程完成操作，然后重试
    ...
  }
}

 

使用CAS的注意事项

避免过度自旋 —— 控制失败重试次数

使用CAS技术时，我们通常会结合失败重试机制。这隐含这一个假设：竞争情况很短暂，即使失败也很快能重试成功。
大多数应用场景中，这个假设确实成立。
但是一些特殊场景中，还是要考虑限制重试（自旋）的次数，以免过度消耗CPU。
【示例】AtomicInteger（Unsafe）中的失败重试：

public class AtomicInteger {
  public final int incrementAndGet() {
    return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
  }
  ...
}

public final class Unsafe {
  public final int getAndAddInt(Object o, long offset, int delta) {
    int v;
    // 失败重试（自旋）：
    do {
      v = getIntVolatile(o, offset);
    } while (!weakCompareAndSetInt(o, offset, v, v + delta));
    return v;
  }
  ...
}

 

避免ABA问题 —— 使用 AtomicStampedReference

CAS机制在更新数值时会比较当前值，如果它与期望的值相同，则成功更新，否则失败。
（这就是 C（cmpare）的含义。）
 

但是在某些业务场景中，仅有目标字段的当前值符合期望并不足以保证后续业务操作的合理性。
如，目标字段发生了 A -> B -> A 的更新，后续操作看到A时会认为自己的操作是合理，而实际上B操作使得后续操作不合理。
 

为了应对这种情况，JDK提供了 AtomicStampedReference，通过增加类似版本号的 stamp 来避免ABA。
这是一个双重校验的机制。只有在 当前值（Reference）和 stamp 都符合预期时才会成功更新。

public class AtomicStampedReference {
  public boolean compareAndSet(V expectedReference, V newReference,
                               int expectedStamp, int newStamp);
  ...
}

 

AQS，AbstractQueuedSynchronizer —— 一个底层的原子化操作基础结构

AQS是JDK并发包中很多同步结构的基础。ReentrantLock、Semaphore、CountDownLatch等同步结构的内部、ThreadPoolExecutor中的Worker都基于AQS实现。
AQS的设计初衷是，从原理上，一种同步结构可以利用其它的结构实现。AQS内部就是包含了一些比较抽象的同步相关基础操作。
 

AQS内部大致可拆解为以下3部分：

# 一个 volatile 字段 state 表示状态。

private volatile int state;

# 一个等待线程队列（先入先出）现实多线程间竞争和等待。

# 各种基于CAS的基础操作方法，及同步结构具备的一些方法（如，acquire、release）。

 

如，ReentrantLock 内部的 Sync 类继承自AQS，利用 state 字段来反映锁的持有情况。
lock 和 unlock 分别使用了 AQS 的 acquire 和 release 方法：

public void lock() {
  sync.acquire(1);
}

public void unlock() {
  sync.release(1);
}

 

AQS 的 acquire() 方法

public final void acquire(int arg) {
  if (!tryAcquire(arg) &&
      acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) {
    selfInterrupt();
  }
}

AQS中的 tryAcquire() 方法仅仅是一个“空壳”（直接抛异常 UnsupportedOperationException）。
在 ReentrantLock 内部的 FairSync 和 NonfairSync 类分别实现了该方法的公平性和非公平性版本。
（创建ReentrantLock时可指定是否为公平锁。）
两者的主要区别在于，非公平版本检测到无其它线程占用时，会直接尝试获取锁，而不检查是否已有其它线程在等待。

 

AQS 的 acquireQueued() 方法

在 acquire() 方法内部，如果 tryAcquire() 失败，就会进入排队竞争阶段。
当前线程会被包装为一个排他模式的节点（exclusive），由addWaiter方法将其添加到等待队列中。
 

此节点类Node包含一个 volatile 字段 waitStatus 来辅助实现线程间的同步。
如：其，。还有 Signal（-1）、Condition（-2）、Propagate（-3）等。

• 初始值 0 表示普通同步节点
• Cancelled 1 表示超时或中断，即节点（线程）不可用
• Signal -1 表示节点（线程）释放资源时会唤醒后面的节点（线程）
• Condition -2 表示节点（线程）处于 条件队列 中；当条件成立后，此节点会被添加到获取资源的队列中
• Progagate -3 在共享模式下使用。如果节点（线程）获取到资源后还有足够的资源，它的后继节点（线程）会尝试获取