JDK中的java.util.concurrent.atomic包提供了一系列支持无锁线程安全修改操作的基础变量。这些原子类是对volatile机制的扩展，并且提供了一系列CAS原子操作方法如：
boolean compareAndSet(expectedValue, updateValue);
这些CAS方法使得我们能够利用机器底层的原子指令，虽然在某些机器平台上，这些指令可能会牵涉到一些内部的加锁，这使得这些方法不是严格遵守非阻塞-即线程可能在执行CAS指令前阻塞。
AtomicXxx原子类
AtomicLong/AtomicInteger/AtomicBoolean/AtomicReference这几个原子类提供了对这些原子类相关的基础类型的get/set操作，同时也提供了一些工具方法。如AtomicLong/AtomicInteger其中一种应用场景就是序列号生成：
class Sequencer {
private final AtomicLong sequenceNumber
= new AtomicLong(0);
public long next() {
return sequenceNumber.getAndIncrement();
}}
同时我们还可以根据这些原子类提供的工具方法，实现我们自己的工具方法，举个例子：
private long transform(long aa){
return aa+2;
}
public long casTransform(AtomicLong var){//原子transform
long pre,next;
do{
pre = var.get();
next = transform(pre);
}while (!var.compareAndSet(pre, next));
return pre;
}
这种用法相当于我们通过AtomicXxxx类代理使用了CPU层面的CAS操作。
下面我们一起看一下这一类原子类的代码实现(以AtomicLong为例)，会发现十分简洁，大部分工作透过UNSAFE实例实现。
- get()/set()
private volatile long value;
public final long get() {
return value;
}
public final void set(long newValue) {
value = newValue;
}
由于value是volatile修饰的，故value在多线程间保持可见性，在多线程情况下可以无锁访问/修改。
- compareAndSet
public final boolean compareAndSet(long expect, long update) {
return unsafe.compareAndSwapLong(this, valueOffset, expect, update);
}
原子的设置value为update，实际的操作是由UNSAFE实例代理的。
public final native boolean compareAndSwapLong((Object) o, long offset, long expected,
long x);
在Unsafe中，compareAndSwapLong是一个native方法，也就是该方法的具体实现是由JVM的本地方法实现的。我们跟踪了openjdk中HotSpotVM的源码，其实也很简单：
UNSAFE_ENTRY(jboolean, Unsafe_CompareAndSwapLong(JNIEnv *env, jobject unsafe, jobject obj, jlong offset, jlong e, jlong x))
UnsafeWrapper("Unsafe_CompareAndSwapLong");
Handle p (THREAD, JNIHandles::resolve(obj));
jlong* addr = (jlong*)(index_oop_from_field_offset_long(p(), offset));
if (VPC_Version::supports_cx8())
<b>return (jlong)(Atomic::cmpxchg(x, addr, e)) == e;</b>
else {
jboolean success = false;
ObjectLocker ol(p, THREAD);
if (*addr == e) { *addr = x; success = true; }
return success;
}
UNSAFE_END
//linux_x86
inline jint Atomic::cmpxchg (jint exchange_value, volatile jint* dest, jint compare_value) {
int mp = os::is_PCP();
__asm__ volatile (LOCK_IF_PCP(%4) "cmpxchgl %1,(%3)"
: "=a" (exchange_value)
: "r" (exchange_value), "a" (compare_value), "r" (dest), "r" (mp)
: "cc", "memory");
return exchange_value;
}
最终执行的是汇编指令，CPU执行cmpxchgl指令对expextVal和updateVal进行比较交换操作。同时，虚拟机也根据不同的系统平台改用对应的操作指令：
//linux_sparc
inline jlong Atomic::cmpxchg (jlong exchange_value, volatile jlong* dest, jlong compare_value) {
#ifdef _LP64
jlong rv;
__asm__ volatile(
" casx [%2], %3, %0" //casx指令
: "=r" (rv)
: "0" (exchange_value), "r" (dest), "r" (compare_value)
: "memory");
return rv;
#else
assert(VM_Version::v9_instructions_work(), "cas only supported on v9");
volatile jlong_accessor evl, cvl, rv;
evl.long_value = exchange_value;
cvl.long_value = compare_value;
__asm__ volatile(
" sllx %2, 32, %2\n\t"
" srl %3, 0, %3\n\t"
" or %2, %3, %2\n\t"
" sllx %5, 32, %5\n\t"
" srl %6, 0, %6\n\t"
" or %5, %6, %5\n\t"
" casx [%4], %5, %2\n\t"
" srl %2, 0, %1\n\t"
" srlx %2, 32, %0\n\t"
: "=r" (rv.words[0]), "=r" (rv.words[1])
: "r" (evl.words[0]), "r" (evl.words[1]), "r" (dest), "r" (cvl.words[0]), "r" (cvl.words[1])
: "memory");
return rv.long_value;
#endif
}
- 循环CAS操作函数
getAndAdd/getAndSet/getAndIncrement/getAndDecrement/incrementAndGet/decrementAndGet/addAndGet这一系列的方法的实现原理都是一样的，所以放在一起说。它们的套路都是先得到value的值，然后自己做操作（加/减），然后CAS设置value为目标值，失败就循环的执行上述步骤，成功就可以返回（这些方法的区别就是，getAndXxxx系列返回的是value的原始值，而xxxAndGet系列返回的是修改成功后的值）。这个实现的核心就在于循环的CAS操作知道CAS成功。以getAndIncrement为例：
<pre>
public final long getAndIncrement() {
while (true) {
long current = get();
long next = current + 1;
if (compareAndSet(current, next))
return current;
}
}
</pre>
AtomicXxxxArray
原子更新数组，这一系列的类可以让我们原子的更新数组的元素。
AtomicXxxxFieldUpdater
如果我们要原子更新一个类的某个字段，就可以使用AtomicXxxxFieldUpdater这个系列的原子类。下面以AtomicLongFieldUpdater为例说明。AtomicLongFieldUpdater<T>可以让你原子的修改类T的long类型成员，AtomicLongFieldUpdater是一个抽象类，内部实现了两个静态私有类CASUpdater<T>与LockedUpdater<T>,所以创建AtomicLongFieldUpdater实例的时候，需要调用静态方法newUpdater。AtomicLongFieldUpdater的使用实例如下：
public class AtomicTest {
static class Person {
private long id;
private String name;
private short sex;
}
public static void main(String[] args) {
AtomicLongFieldUpdater<Person> updater = AtomicLongFieldUpdater.newUpdater(Person.class, "id");
Person person = new Person();
updater.addAndGet(person, 123353647L);
}
}
- getAndXxx/incrementAndGet/decrementAndGet
该系列方法的实现与AtomicLong这些原子类的实现是类似的，也即通过循环CAS来设置目标值。而具体调用CASUpdater<T>还是LockedUpdater<T>的CAS函数则是根据JVM是否支持long类型的无锁cas操作而异。在new一个AtomicLongFieldUpdater时，如果运行该程序的JVM支持long类型的无锁cas操作，就会new一个CASUpdater，反之会new一个LockedUpdater。
public static <U> AtomicLongFieldUpdater<U> newUpdater(Class<U> tclass, String fieldName) {
Class<?> caller = Reflection.getCallerClass();
if (AtomicLong.VM_SUPPORTS_LONG_CAS)
return new CASUpdater<U>(tclass, fieldName, caller);
else
return new LockedUpdater<U>(tclass, fieldName, caller);
}
/**
* Records whether the underlying JVM supports lockless
* compareAndSwap for longs. While the Unsafe.compareAndSwapLong
* method works in either case, some constructions should be
* handled at Java level to avoid locking user-visible locks.
*/
static final boolean VM_SUPPORTS_LONG_CAS = VMSupportsCS8();
/**
* Returns whether underlying JVM supports lockless CompareAndSet
* for longs. Called only once and cached in VM_SUPPORTS_LONG_CAS.
*/
private static native boolean VMSupportsCS8();//该native方法只执行一次
这两个类的compareAndSet函数实现如下：
//CASUpdater
public boolean compareAndSet(T obj, long expect, long update) {
if (obj == null || obj.getClass() != tclass || cclass != null) fullCheck(obj);
return unsafe.compareAndSwapLong(obj, offset, expect, update);
}
//LockedUpdater
public boolean compareAndSet(T obj, long expect, long update) {
if (obj == null || obj.getClass() != tclass || cclass != null) fullCheck(obj);
synchronized (this) {
long v = unsafe.getLong(obj, offset);
if (v != expect)
return false;
unsafe.putLong(obj, offset, update);
return true;
}
}
我们可以非常清楚的看到它们的异同。由于LockedUpdater.compareAndSet是在JVM不支持long类型的无锁cas操作下调用的，所以该cas操作需要显示加锁来实现CAS操作，因此性能相比CASUpdater类会下降。对于其他get/set操作，也是类似情况。
AtomicReference
AtomicReference用来原子的更新一个对象引用。
AtomicReferenceArray用来原子的更新一个对象引用数组中的一个元素。
AtomicStampedReference维护了一个对象引用以及该引用对应的一个版本号。在AtomicStampedReference内部有一个私有的Pair数据结构保存这两个值。同时，AtomicStampedReference有一个volatile的Pair类型成员变量。
private volatile Pair<V> pair;
private static class Pair<T> {
final T reference;
final int stamp;
private Pair(T reference, int stamp) {
this.reference = reference;
this.stamp = stamp;
}
static <T> Pair<T> of(T reference, int stamp) {
return new Pair<T>(reference, stamp);
}
}
AtomicStampedReference在设置它的引用时，必须同时设置版本号。
public void set(V newReference, int newStamp) {
Pair<V> current = pair;
if (newReference != current.reference || newStamp != current.stamp)
this.pair = Pair.of(newReference, newStamp);//pair为volitile
}
AtomicStampedReference提供的cas接口与其他Atomic类都不太一样，该方法在reference与stamp都与当前的两个值一样的情况下才能成功设置至两个值为新的值。AtomicStampedReference这种特殊的cas可以解决CAS实现原子操作的ABA问题,我们在最后会解释。
//最后的casPair通过UNSAFE原子的修改两个Pair变量
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可以视作是带版本号的AtomicReference,每次更新AtomicStampedReference都需要设置新的版本号。而AtomicMarkableReference可以视作为带布尔标记的AtomicReference。AtomicMarkableReference与AtomicStampedReference的操作方法实现基本一样。
CAS实现原子操作的问题
- ABA问题
CAS操作的原理就是在对目标变量进行操作时，先检查值是否发生了变化，如果没有变化则更新，否则返回失败并重试。但可能会存在这样一种问题，如果一个值原来是A，变成了B，又变成了A，则使用CAS进行检查时就会认为它的值没有变化，而实际却是变化了。这个问题的解决方案就是我们给每次变化加一个版本号，只要每次变量被修改，就对版本号加一，版本号是线性增长的，不会出现A->B->A这种情况，每次CAS比较的时候同时比较这两个的值，如果这两个值没有同时相等，就返回修改失败，重试直到成功。JDK里为我们提供了解决方案，就是AtomicStampedReference，上一节里对其原理有分析。 - 只能保证一个共享变量是原子操作
当一个共享变量出现并发修改，我们可以使用循环CAS的方式保证原子操作，但是对于多个共享变量，我们就无法使用循环CAS操作来保证对多个共享变量的原子操作。这时候除了可以使用锁以外，我们还可以把这两个变量放到一个类中，通过AtomicReference引用这个类的实例，当要对这两个变量进行原子修改时，调用其compareAndSet方法修改指向新的变量的引用即可。
参考
- 《Java并发编程的艺术》
- JDK1.7源码
网友评论