程序员@IT·互联网技术干货

JDK原子类源码分析

2017-04-17  本文已影响201人  Justlearn

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实例实现。

private volatile long value;
public final long get() {
        return value;
    }
public final void set(long newValue) {
        value = newValue;
    }

由于value是volatile修饰的，故value在多线程间保持可见性，在多线程情况下可以无锁访问/修改。

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
}

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

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实现原子操作的问题

参考
上一篇 下一篇

猜你喜欢

热点阅读