Java原子类解析

CAS原理

CAS算法的过程是这样：
它包含三个参数CAS(V,E,N);V表示要更新的变量,E表示预期值,N表示新值.
仅当V值等于E值时,才会将V的值设为N,如果V值和E值不同,则说明已经有其他线程做了更新,则当前线程什么都不做.最后,CAS返回当前V的真实值
CAS操作是抱着乐观的态度进行的,它总是认为自己可以成功完成操作.当多个线程同时使用CAS操作一个变量时,只有一个会胜出并成功更新,其余均会失败.失败的线程不会被挂起,仅是被告知失败,并且允许再次尝试,当然也允许失败的线程放弃操作.基于这样的原理,CAS操作即使没有锁,也可以发现其他线程对当前线程的干扰,并进行恰当的处理
简单地说,CAS需要你额外给出一个期望值,也就是你认为这个变量现在应该是什么样子的.如果变量不是你想象的那样,那说明它已经被别人修改过了.你就重新读取,再次尝试修改就好了
在硬件层面,大部分的现代处理器都已经支持原子化的CAS指令.在JDK 5.0以后,虚拟机便可以使用这个指令来实现并发操作和并发数据结构,并且这种操作在虚拟机中可以说是无处不在

CAS优点

与锁相比,使用比较交换(下文简称CAS)会使程序看起来更加复杂一些.但由于其非阻塞性,它对死锁问题天生免疫,并且线程间的相互影响也远远比基于锁的方式要小.更为重要的是使用无锁的方式完全没有锁竞争带来的系统开销,也没有线程间频繁调度带来的开销,因此它要比基于锁的方式拥有更优越的性能

在高并发的情况下,它比有锁的程序拥有更好的性能
它天生就是死锁免疫的
就凭借这两个优势,就值得我们冒险尝试使用无锁的并发

CAS缺点

CAS存在一个很明显的问题,即ABA问题
问题：如果变量V初次读取的时候是A,并且在准备赋值的时候检查到它仍然是A,那能说明它的值没有被其他线程修改过了吗
如果在这段期间曾经被改成B，然后又改回A，那CAS操作就会误认为它从来没有被修改过。针对这种情况，java并发包中提供了一个带有标记的原子引用类AtomicStampedReference，它可以通过控制变量值的版本来保证CAS的正确性

CAS(乐观锁算法)的基本假设前提

CAS比较与交换的伪代码可以表示为：

do{
    备份旧数据；
    基于旧数据构造新数据；
}while(!CAS( 内存地址，备份的旧数据，新数据 )

就是指当两者进行比较时,如果相等,则证明共享数据没有被修改,替换成新值,然后继续往下运行;如果不相等,说明共享数据已经被修改,放弃已经所做的操作,然后重新执行刚才的操作.容易看出CAS操作是基于共享数据不会被修改的假设,采用了类似于数据库的commit-retry的模式.当同步冲突出现的机会很少时,这种假设能带来较大的性能提升

Unsafe解析

Java和C++语言的一个重要区别就是Java中我们无法直接操作一块内存区域,不能像C++中那样可以自己申请内存和释放内存.Java中的Unsafe类为我们提供了类似C++手动管理内存的能力.
Unsafe类：全限定名是sun.misc.Unsafe,从名字中我们可以看出来这个类对普通程序员来说是”危险”的,一般应用开发者不会用到这个类

Unsafe类是”final”的,不允许继承.且构造函数是private的:

public final class Unsafe {
    private static final Unsafe theUnsafe;
    public static final int INVALID_FIELD_OFFSET = -1;
    
    private static native void registerNatives();

    // 构造函数是private的,不允许外部实例化
    private Unsafe() {
    }
}    
```  
因此我们无法在外部对Unsafe进行实例化
### 获取Unsafe
Unsafe无法实例化,那么怎么获取Unsafe呢?答案就是通过反射来获取Unsafe：
```java
@CallerSensitive
public static Unsafe getUnsafe() {
    Class var0 = Reflection.getCallerClass();
    if (!VM.isSystemDomainLoader(var0.getClassLoader())) {
        throw new SecurityException("Unsafe");
    } else {
        return theUnsafe;
    }
}

主要功能

Unsafe的功能如下图：

普通读写

通过Unsafe可以读写一个类的属性,即使这个属性是私有的,也可以对这个属性进行读写

读写一个Object属性的相关方法

1 2	public native int getInt(Object var1, long var2); public native void putInt(Object var1, long var2, int var4);

getInt用于从对象的指定偏移地址处读取一个int.putInt用于在对象指定偏移地址处写入一个int.其他的primitive type也有对应的方法

Unsafe还可以直接在一个地址上读写

1 2	public native byte getByte(Object var1, long var2); public native void putByte(Object var1, long var2, byte var4);

getByte用于从指定内存地址处开始读取一个byte.putByte用于从指定内存地址写入一个byte

volatile读写

普通的读写无法保证可见性和有序性,而volatile读写就可以保证可见性和有序性

1 2	public native int getIntVolatile(Object var1, long var2); public native void putIntVolatile(Object var1, long var2, int var4);

getIntVolatile方法用于在对象指定偏移地址处volatile读取一个int.putIntVolatile方法用于在对象指定偏移地址处volatile写入一个int

volatile读写相对普通读写是更加昂贵的,因为需要保证可见性和有序性,而与volatile写入相比putOrderedXX写入代价相对较低,putOrderedXX写入不保证可见性,但是保证有序性,所谓有序性,就是保证指令不会重排序

有序写入

有序写入只保证写入的有序性,不保证可见性,就是说一个线程的写入不保证其他线程立马可见

1
2
3

public native void putOrderedObject(Object var1, long var2, Object var4);
public native void putOrderedInt(Object var1, long var2, int var4);
public native void putOrderedLong(Object var1, long var2, long var4);

直接内存操作

我们都知道Java不可以直接对内存进行操作,对象内存的分配和回收都是由JVM帮助我们实现的.但是Unsafe为我们在Java中提供了直接操作内存的能力

// 分配内存
public native long allocateMemory(long var1);
// 重新分配内存
public native long reallocateMemory(long var1, long var3);
// 内存初始化
public native void setMemory(long var1, long var3, byte var5);
// 内存复制
public native void copyMemory(Object var1, long var2, Object var4, long var5, long var7);
// 清除内存
public native void freeMemory(long var1);

CAS相关

JUC中大量运用了CAS操作,可以说CAS操作是JUC的基础,因此CAS操作是非常重要的.Unsafe中提供了int,long和Object的CAS操作：

1
2
3

public final native boolean compareAndSwapObject(Object var1, long var2, Object var4, Object var5);
public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5);
public final native boolean compareAndSwapLong(Object var1, long var2, long var4, long var6);

CAS一般用于乐观锁,它在Java中有广泛的应用,ConcurrentHashMap，ConcurrentLinkedQueue中都有用到CAS来实现乐观锁

偏移量相关

public native long staticFieldOffset(Field var1);
public native long objectFieldOffset(Field var1);
public native Object staticFieldBase(Field var1);
public native int arrayBaseOffset(Class<?> var1);
public native int arrayIndexScale(Class<?> var1);

staticFieldOffset方法用于获取静态属性Field在对象中的偏移量,读写静态属性时必须获取其偏移量.objectFieldOffset方法用于获取非静态属性Field在对象实例中的偏移量,读写对象的非静态属性时会用到这个偏移量.staticFieldBase方法用于返回Field所在的对象.arrayBaseOffset方法用于返回数组中第一个元素实际地址相对整个数组对象的地址的偏移量.arrayIndexScale方法用于计算数组中第一个元素所占用的内存空间

线程调度

public native void unpark(Object var1);
public native void park(boolean var1, long var2);
public native void monitorEnter(Object var1);
public native void monitorExit(Object var1);
public native boolean tryMonitorEnter(Object var1);

park方法和unpark方法相信看过LockSupport类的都不会陌生,这两个方法主要用来挂起和唤醒线程.LockSupport中的park和unpark方法正是通过Unsafe来实现的：

// 挂起线程
public static void park(Object blocker) {
    Thread t = Thread.currentThread();
    setBlocker(t, blocker); // 通过Unsafe的putObject方法设置阻塞阻塞当前线程的blocker
    UNSAFE.park(false, 0L); // 通过Unsafe的park方法来阻塞当前线程，注意此方法将当前线程阻塞后，当前线程就不会继续往下走了，直到其他线程unpark此线程
    setBlocker(t, null); // 清除blocker
}

// 唤醒线程
public static void unpark(Thread thread) {
    if (thread != null)
        UNSAFE.unpark(thread);
}

monitorEnter方法和monitorExit方法用于加锁,Java中的synchronized锁就是通过这两个指令来实现的

类加载

public native Class<?> defineClass(String var1, byte[] var2, int var3, int var4, ClassLoader var5, ProtectionDomain var6);
public native Class<?> defineAnonymousClass(Class<?> var1, byte[] var2, Object[] var3);
public native Object allocateInstance(Class<?> var1) throws InstantiationException;
public native boolean shouldBeInitialized(Class<?> var1);
public native void ensureClassInitialized(Class<?> var1);

defineClass方法定义一个类,用于动态地创建类.
defineAnonymousClass用于动态的创建一个匿名内部类
allocateInstance方法用于创建一个类的实例,但是不会调用这个实例的构造方法,如果这个类还未被初始化,则初始化这个类
shouldBeInitialized方法用于判断是否需要初始化一个类
ensureClassInitialized方法用于保证已经初始化过一个类

内存屏障

1
2
3

public native void loadFence();
public native void storeFence();
public native void fullFence();

loadFence：保证在这个屏障之前的所有读操作都已经完成
storeFence：保证在这个屏障之前的所有写操作都已经完成
fullFence：保证在这个屏障之前的所有读写操作都已经完成

原子类Atomic

java.util.concurrent.atomic包：原子类的小工具包,支持在单个变量上解除锁的线程安全编程
原子变量类相当于一种泛化的volatile变量,能够支持原子的和有条件的读-改-写操作.AtomicInteger表示一个int类型的值,并提供了get和set方法,这些 Volatile类型的int变量在读取和写入上有着相同的内存语义.它还提供了一个原子的compareAndSet方法(如果该方法成功执行,那么将实现与读取/写入一个volatile变量相同的内存效果),以及原子的添加、递增和递减等方法.AtomicInteger表面上非常像一个扩展的Counter类,但在发生竞争的情况下能提供更高的可伸缩性,因为它直接利用了硬件对并发的支持

为什么会有原子类

jdk5增加了并发包java.util.concurrent.,其下面的类使用CAS算法实现了区别于synchronouse同步锁的一种乐观锁.JDK5之前Java语言是靠synchronized关键字保证同步的,这是一种独占锁,也是是悲观锁
*如果同一个变量要被多个线程访问,则可以使用该包中的类**

常用原子类

Java中的原子操作类大致可以分为4类：原子更新基本类型、原子更新数组类型、原子更新引用类型、原子更新属性类型.这些原子类中都是用了无锁的概念,有的地方直接使用CAS操作的线程安全的类型

AtomicInteger
AtomicReference

AtomicInteger

一个可以自动更新的int值.AtomicInteger用于诸如原子递增计数器之类的应用程序,并且不能替代Integer.但是此类确实扩展了Number,以允许处理基于数字的类的工具和实用程序进行统一访问

主要变量

// 设置为使用Unsafe.compareAndSwapInt进行更新
private static final Unsafe unsafe = Unsafe.getUnsafe();
private static final long valueOffset;

static {
    try {
        valueOffset = unsafe.objectFieldOffset
            (AtomicInteger.class.getDeclaredField("value"));
    } catch (Exception ex) { throw new Error(ex); }
}

// 封装了一个int值对其进行加减
private volatile int value;

构造方法

/**
 * 使用给定的初始值创建一个新的AtomicInteger
 *
 * @param 初始值
 */
public AtomicInteger(int initialValue) {
    value = initialValue;
}

public AtomicInteger() {
}

主要方法

// 获取当前值
public final int get() {
    return value;
}

// 设置当前值
public final void set(int newValue) {
    value = newValue;
}

// 最终设置值
public final void lazySet(int newValue) {
    unsafe.putOrderedInt(this, valueOffset, newValue);
}

// 设置新值,并返回旧值
public final int getAndSet(int newValue) {
    return unsafe.getAndSetInt(this, valueOffset, newValue);
}

// 如果当前值为expect,则设置为u
public final boolean compareAndSet(int expect, int update) {
    return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
}

// 可能会失败,并且不提供排序保证,因此很少是compareAndSet的适当替代方法
public final boolean weakCompareAndSet(int expect, int update) {
    return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
}

// 当前值加1,返回旧值
public final int getAndIncrement() {
    return unsafe.getAndAddInt(this, valueOffset, 1);
}

// 当前值减1,返回旧值
public final int getAndDecrement() {
    return unsafe.getAndAddInt(this, valueOffset, -1);
}

// 当前值增加delta,返回旧值
public final int getAndAdd(int delta) {
    return unsafe.getAndAddInt(this, valueOffset, delta);
}

// 当前值加1,返回新值
public final int incrementAndGet() {
    return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
}

// 当前值减1,返回新值
public final int decrementAndGet() {
    return unsafe.getAndAddInt(this, valueOffset, -1) - 1;
}

// 当前值增加delta,返回新值
public final int addAndGet(int delta) {
    return unsafe.getAndAddInt(this, valueOffset, delta) + delta;
}

// 自旋式更新值,值为入参给定函数的结果,并返回先前的值
public final int getAndUpdate(IntUnaryOperator updateFunction) {
    int prev, next;
    do {
        prev = get();
        next = updateFunction.applyAsInt(prev);
    } while (!compareAndSet(prev, next));
    return prev;
}

// 自旋式更新值,值为入参给定函数的结果,并返回更新值
public final int updateAndGet(IntUnaryOperator updateFunction) {
    int prev, next;
    do {
        prev = get();
        next = updateFunction.applyAsInt(prev);
    } while (!compareAndSet(prev, next));
    return next;
}

// 将入参给定函数应用于当前值和给定值,以原子方式自旋的更新当前值,并返回先前的值
public final int getAndAccumulate(int x,  IntBinaryOperator accumulatorFunction) {
    int prev, next;
    do {
        prev = get();
        next = accumulatorFunction.applyAsInt(prev, x);
    } while (!compareAndSet(prev, next));
    return prev;
}

// 将入参给定函数应用于当前值和给定值,以原子方式自旋的更新当前值,并返回更新后的值
public final int accumulateAndGet(int x,  IntBinaryOperator accumulatorFunction) {
    int prev, next;
    do {
        prev = get();
        next = accumulatorFunction.applyAsInt(prev, x);
    } while (!compareAndSet(prev, next));
    return next;
}

AtomicReference

与AtomicInteger类似,只是里面封装了一个对象,而不是int,对引用进行修改

主要变量

private static final Unsafe unsafe = Unsafe.getUnsafe();
private static final long valueOffset;

static {
    try {
        valueOffset = unsafe.objectFieldOffset
            (AtomicReference.class.getDeclaredField("value"));
    } catch (Exception ex) { throw new Error(ex); }
}

private volatile V value;

构造方法

 /**
 * 使用给定的初始值创建一个新的AtomicReference
 */
public AtomicReference(V initialValue) {
    value = initialValue;
}

public AtomicReference() {
}

主要方法

/**
 * 获取当前值
 */
public final V get() {
    return value;
}

/**
 * 设置给定值
 */
public final void set(V newValue) {
    value = newValue;
}

/**
 * 最终设置为给定值,懒加载
 */
public final void lazySet(V newValue) {
    unsafe.putOrderedObject(this, valueOffset, newValue);
}

/**
 * 如果当前值==期望值,则以原子方式将该值设置为给定的更新值
 */
public final boolean compareAndSet(V expect, V update) {
    return unsafe.compareAndSwapObject(this, valueOffset, expect, update);
}

/**
 * 可能会失败,并且不提供保证,因此很少是compareAndSet的适当替代方案
 */
public final boolean weakCompareAndSet(V expect, V update) {
    return unsafe.compareAndSwapObject(this, valueOffset, expect, update);
}

/**
 * 以原子方式设置为给定值并返回旧值
 */
@SuppressWarnings("unchecked")
public final V getAndSet(V newValue) {
    return (V)unsafe.getAndSetObject(this, valueOffset, newValue);
}

/**
 * 使用入参给定函数的结果以原子方式自旋的更新当前值,并返回先前的值.
 */
public final V getAndUpdate(UnaryOperator<V> updateFunction) {
    V prev, next;
    do {
        prev = get();
        next = updateFunction.apply(prev);
    } while (!compareAndSet(prev, next));
    return prev;
}

/**
 * 使用入参给定函数的结果以原子方式自旋的更新当前值,并返回更新后的值
 */
public final V updateAndGet(UnaryOperator<V> updateFunction) {
    V prev, next;
    do {
        prev = get();
        next = updateFunction.apply(prev);
    } while (!compareAndSet(prev, next));
    return next;
}

/**
 * 将入参给定函数应用于当前值和入参给定值,以原子方式自旋的更新当前值,并返回先前的值
 */
public final V getAndAccumulate(V x, BinaryOperator<V> accumulatorFunction) {
    V prev, next;
    do {
        prev = get();
        next = accumulatorFunction.apply(prev, x);
    } while (!compareAndSet(prev, next));
    return prev;
}

/**
 * 将入参给定函数应用于当前值和入参给定值,以原子方式自旋的更新当前值,并返回更新后的值
 */
public final V accumulateAndGet(V x, BinaryOperator<V> accumulatorFunction) {
    V prev, next;
    do {
        prev = get();
        next = accumulatorFunction.apply(prev, x);
    } while (!compareAndSet(prev, next));
    return next;
}

AtomicStampedReference

AtomicStampedReference维护对象引用以及可以自动更新的整数”标记”,主要通过包装[对象引用, stamp],进行对象引用与stamp的对应绑定,更新stamp来解决ABA问题

ABA问题：
线程一准备用CAS将变量的值由A替换为B,在此之前线程二将变量的值由A替换为C,线程三又将C替换为A,然后线程一执行CAS时发现变量的值仍然为A,所以线程一CAS成功

主要变量

// 内部封装了一个Pair对象,每次对对象操作的时候,stamp + 1
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);
    }
}

private volatile Pair<V> pair;

构造方法

// 使用给定的初始值创建一个新的AtomicStampedReference
public AtomicStampedReference(V initialRef, int initialStamp) {
    pair = Pair.of(initialRef, initialStamp);
}

主要方法

// 获得当前对象引用
public V getReference() {
    return pair.reference;
}

// 获得当前时间戳
public int getStamp() {
    return pair.stamp;
}

// 返回当前的对象引用(返回值)和时间戳(入参返回)
public V get(int[] stampHolder) {
    Pair<V> pair = this.pair;
    stampHolder[0] = pair.stamp;
    return pair.reference;
}

// 可能会失败,不提供保证,因此很少是compareAndSet的适当替代方法
public boolean weakCompareAndSet(V expectedReference, V newReference, int expectedStamp, int newStamp) {
    return compareAndSet(expectedReference, newReference, expectedStamp, newStamp);
}

// // CAS操作,会比较stamp值 参数依次为：期望值 写入新值 期望时间戳 新时间戳
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)));
}

// 设置当前对象引用和时间戳
public void set(V newReference, int newStamp) {
    Pair<V> current = pair;
    if (newReference != current.reference || newStamp != current.stamp)
        this.pair = Pair.of(newReference, newStamp);
}

充值示例模拟

后台使用多个线程对用户充值,要求只能充值一次

public class AtomicStampedReferenceDemo {
 static AtomicStampedReference<Integer> money=new AtomicStampedReference<Integer>(19,0);
 public staticvoid main(String[] args) {
    //模拟多个线程同时更新后台数据库，为用户充值
    for(int i = 0 ; i < 3 ; i++) {
        final int timestamp=money.getStamp();
        newThread() {  
            public void run() { 
                while(true){
                   while(true){
                       Integerm=money.getReference();
                        if(m<20){
                     if(money.compareAndSet(m,m+20,timestamp,timestamp+1)){
      　　　　　　　　　　　　　　　 System.out.println("余额小于20元，充值成功，余额:"+money.getReference()+"元");
                                break;
                            }
                        }else{
                           //System.out.println("余额大于20元，无需充值");
                            break ;
                         }
                   }
                }
            } 
        }.start();
     }
    
    //用户消费线程，模拟消费行为
    new Thread() { 
         publicvoid run() { 
            for(int i=0;i<100;i++){
               while(true){
                    int timestamp=money.getStamp();
                    Integer m=money.getReference();
                    if(m>10){
                         System.out.println("大于10元");
                      　　if(money.compareAndSet(m, m-10,timestamp,timestamp+1)){
                  　　　　　　 System.out.println("成功消费10元，余额:"+money.getReference());
                             break;
                         }
                    }else{
                       System.out.println("没有足够的金额");
                         break;
                    }
                }
                try {Thread.sleep(100);} catch (InterruptedException e) {}
             }
        } 
    }.start(); 
  }
}

AtomicIntegerArray

支持无锁的数组,其中的元素可以原子更新

主要变量

private static final Unsafe unsafe = Unsafe.getUnsafe();
// 数组本身基地址
private static final int base = unsafe.arrayBaseOffset(int[].class);

private static final int shift;
// 封装了一个数组
private final int[] array;

static {
    // 数组中对象的宽度,int类型,4个字节,scale = 4;
    int scale = unsafe.arrayIndexScale(int[].class);
    if ((scale & (scale - 1)) != 0)
        throw new Error("data type scale not a power of two");
    
    // 前导0个数: 一个数字转为二进制后,他前面0的个数
    // 对于4来讲,他就是00000000 00000000 00000000 00000100,他的前导0个数就是29
    // 所以shift = 2    
    shift = 31 - Integer.numberOfLeadingZeros(scale);
}

构造方法

/**
 * 创建一个给定长度的新AtomicIntegerArray,所有元素最初为零
 */
public AtomicIntegerArray(int length) {
    array = new int[length];
}

/**
 * 创建一个新AtomicIntegerArray,其长度与给定数组相同,并从给定数组中复制所有元素
 */
public AtomicIntegerArray(int[] array) {
    // 最终现场保证保证可见性
    this.array = array.clone();
}

主要方法

// 第i个元素：在数组中的偏移量是多少
private long checkedByteOffset(int i) {
    if (i < 0 || i >= array.length)
        throw new IndexOutOfBoundsException("index " + i);

    return byteOffset(i);
}

// base: 数组基地址,i << shift,其实就是i * 4,因为这边是int array.
private static long byteOffset(int i) {
    return ((long) i << shift) + base;
}

// 获取第i个元素
public final int get(int i) {
    return getRaw(checkedByteOffset(i));
}

// 将指定位置i的元素设置为给定值
public final void set(int i, int newValue) {
    unsafe.putIntVolatile(array, checkedByteOffset(i), newValue);
}

// 将数组第i个下标设置为newValue,并返回旧的值
public final int getAndSet(int i, int newValue) {
    return unsafe.getAndSetInt(array, checkedByteOffset(i), newValue);
}

// 进行CAS操作,如果第i个下标的元素等于expect,则设置为update,设置成功返回true
public final boolean compareAndSet(int i, int expect, int update) {
    return compareAndSetRaw(checkedByteOffset(i), expect, update);
}

// 将第i个下标的元素加1
public final int getAndIncrement(int i) {
    return getAndAdd(i, 1);
}

// 将第i个下标的元素减1
public final int getAndDecrement(int i) {
    return getAndAdd(i, -1);
}

// 将第i个下标的元素增加delta(delta可以是负数)
public final int getAndAdd(int i, int delta) {
    return unsafe.getAndAddInt(array, checkedByteOffset(i), delta);
}

// 以原子方式将索引i处的元素加1
public final int incrementAndGet(int i) {
    return getAndAdd(i, 1) + 1;
}

// 以原子方式将索引i处的元素减1
public final int decrementAndGet(int i) {
    return getAndAdd(i, -1) - 1;
}

// 以原子方式将给定值添加到索引为i的元素
public final int addAndGet(int i, int delta) {
    return getAndAdd(i, delta) + delta;
}

// 使用入参给定函数的结果以原子方式自旋的更新索引i处的元素,并返回先前的值
public final int getAndUpdate(int i, IntUnaryOperator updateFunction) {
    long offset = checkedByteOffset(i);
    int prev, next;
    do {
        prev = getRaw(offset);
        next = updateFunction.applyAsInt(prev);
    } while (!compareAndSetRaw(offset, prev, next));
    return prev;
}

// 使用入参给定函数的结果以原子方式自旋的更新索引i处的元素,并返回更新后的值
public final int updateAndGet(int i, IntUnaryOperator updateFunction) {
    long offset = checkedByteOffset(i);
    int prev, next;
    do {
        prev = getRaw(offset);
        next = updateFunction.applyAsInt(prev);
    } while (!compareAndSetRaw(offset, prev, next));
    return next;
}

// 将入参的给定函数应用于当前值和入参的给定值,以原子方式自旋的更新索引i处的元素,并返回前一个值
public final int getAndAccumulate(int i, int x, IntBinaryOperator accumulatorFunction) {
    long offset = checkedByteOffset(i);
    int prev, next;
    do {
        prev = getRaw(offset);
        next = accumulatorFunction.applyAsInt(prev, x);
    } while (!compareAndSetRaw(offset, prev, next));
    return prev;
}

// 将入参的给定函数应用于当前值和入参的给定值,以原子方式自旋的更新索引i处的元素,并返回后一个值
public final int accumulateAndGet(int i, int x, IntBinaryOperator accumulatorFunction) {
    long offset = checkedByteOffset(i);
    int prev, next;
    do {
        prev = getRaw(offset);
        next = accumulatorFunction.applyAsInt(prev, x);
    } while (!compareAndSetRaw(offset, prev, next));
    return next;
}

AtomicIntegerFieldUpdater

让普通变量也享受原子操作,基于反射的实用程序,可通过原子更新来指定类别的volatile int个字段.此类设计用于原子数据结构,其中同一节点的多个字段独立地受原子限制更新

注意: compareAndSet的保证该类中的方法比其他原子类中的方法弱.因为此类无法确保该字段的所有用途适用于原子访问.它可以仅在其他调用方面保证原子性在同一更新程序上的compareAndSet和set

构造方法

// 为具有给定字段的对象创建并返回更新程序.需要使用Class参数来检查反射类型和泛型类型是否匹配
@CallerSensitive
public static <U> AtomicIntegerFieldUpdater<U> newUpdater(Class<U> tclass, String fieldName) {
    return new AtomicIntegerFieldUpdaterImpl<U>(tclass, fieldName, Reflection.getCallerClass());
}

protected AtomicIntegerFieldUpdater() {}

updater只能修改它可见范围内的变量.因为updater使用反射得到这个变量.如果变量不可见,就会出错.比如如果score申明为private,就是不可行的
为了确保变量被正确的读取,它必须是volatile类型的.如果我们原有代码中未申明这个类型,那么简单得申明一下就行,这不会引起什么问题
由于CAS操作会通过对象实例中的偏移量直接进行赋值,因此它不支持static字段(Unsafe.objectFieldOffset()不支持静态变量)

内部类

private static final class AtomicIntegerFieldUpdaterImpl<T> extends AtomicIntegerFieldUpdater<T> {
    private static final sun.misc.Unsafe U = sun.misc.Unsafe.getUnsafe();
    private final long offset;
    /**
     * 如果字段受保护,则子类构造Updater,否则与tclass相同
     */
    private final Class<?> cclass;
    /** class保持字段 */
    private final Class<T> tclass;
    
    AtomicIntegerFieldUpdaterImpl(final Class<T> tclass, final String fieldName,  final Class<?> caller) {
        final Field field;
        final int modifiers;
        try {
            field = AccessController.doPrivileged(
                new PrivilegedExceptionAction<Field>() {
                    public Field run() throws NoSuchFieldException {
                        return tclass.getDeclaredField(fieldName);
                    }
                });
            modifiers = field.getModifiers();
            sun.reflect.misc.ReflectUtil.ensureMemberAccess(
                caller, tclass, null, modifiers);
            ClassLoader cl = tclass.getClassLoader();
            ClassLoader ccl = caller.getClassLoader();
            if ((ccl != null) && (ccl != cl) &&
                ((cl == null) || !isAncestor(cl, ccl))) {
                sun.reflect.misc.ReflectUtil.checkPackageAccess(tclass);
            }
        } catch (PrivilegedActionException pae) {
            throw new RuntimeException(pae.getException());
        } catch (Exception ex) {
            throw new RuntimeException(ex);
        }
    
        if (field.getType() != int.class)
            throw new IllegalArgumentException("Must be integer type");
    
        if (!Modifier.isVolatile(modifiers))
            throw new IllegalArgumentException("Must be volatile type");
    
        // 对受保护字段成员的访问仅限于访问类或其子类之一的接收者,并且访问类又必须是受保护成员的定义类的子类(或程序包同级).如果更新程序在当前包外部引用了声明类的受保护字段,则接收器参数将缩小为访问类的类型
        this.cclass = (Modifier.isProtected(modifiers) &&
                       tclass.isAssignableFrom(caller) &&
                       !isSamePackage(tclass, caller))
                      ? caller : tclass;
        this.tclass = tclass;
        this.offset = U.objectFieldOffset(field);
    }
    
    /**
     * 如果可以在第一个类加载器的委托链中找到第二个类加载器,则返回true.等效于不可访问的：first.isAncestor(second)
     */
    private static boolean isAncestor(ClassLoader first, ClassLoader second) {
        ClassLoader acl = first;
        do {
            acl = acl.getParent();
            if (second == acl) {
                return true;
            }
        } while (acl != null);
        return false;
    }
    
    /**
     * 如果两个类具有相同的类加载器和包限定符,则返回true
     */
    private static boolean isSamePackage(Class<?> class1, Class<?> class2) {
        return class1.getClassLoader() == class2.getClassLoader()
               && Objects.equals(getPackageName(class1), getPackageName(class2));
    }

    private static String getPackageName(Class<?> cls) {
        String cn = cls.getName();
        int dot = cn.lastIndexOf('.');
        return (dot != -1) ? cn.substring(0, dot) : "";
    }

    /**
     * 检查目标参数是否是类的实例.失败时,抛出原因
     */
    private final void accessCheck(T obj) {
        if (!cclass.isInstance(obj))
            throwAccessCheckException(obj);
    }
    
    public final boolean compareAndSet(T obj, int expect, int update) {
        accessCheck(obj);
        return U.compareAndSwapInt(obj, offset, expect, update);
    }
    
    public final void set(T obj, int newValue) {
        accessCheck(obj);
        U.putIntVolatile(obj, offset, newValue);
    }
    
    public final int get(T obj) {
        accessCheck(obj);
        return U.getIntVolatile(obj, offset);
    }

    public final int getAndSet(T obj, int newValue) {
        accessCheck(obj);
        return U.getAndSetInt(obj, offset, newValue);
    }

    public final int getAndAdd(T obj, int delta) {
        accessCheck(obj);
        return U.getAndAddInt(obj, offset, delta);
    }
}

主要方法

// 如果当前值==期望值,则以原子方式将此更新程序管理的给定对象的字段设置为给定更新值.对于compareAndSet和set的其他调用,此方法保证是原子的,但对于该字段中的其他更改,则不一定是原子的
public abstract boolean compareAndSet(T obj, int expect, int update);

// 将此更新程序管理的给定对象的字段设置为给定的更新值.对于后续compareAndSet的调用,保证此操作充当易失性存储
public abstract void set(T obj, int newValue);

// 获取在此更新程序管理的给定对象的字段中保留的当前值
public abstract int get(T obj);

// 以原子方式将此更新程序管理的给定对象的字段设置为给定值,并返回旧值
public int getAndSet(T obj, int newValue) {
    int prev;
    do {
        prev = get(obj);
    } while (!compareAndSet(obj, prev, newValue));
    return prev;
}

// 以原子方式将此更新程序管理的给定对象的字段的当前值增加一
public int getAndIncrement(T obj) {
    int prev, next;
    do {
        prev = get(obj);
        next = prev + 1;
    } while (!compareAndSet(obj, prev, next));
    return prev;
}

/**
 * 以原子方式将此更新器管理的给定对象的字段的当前值减一
 */
public int getAndDecrement(T obj) {
    int prev, next;
    do {
        prev = get(obj);
        next = prev - 1;
    } while (!compareAndSet(obj, prev, next));
    return prev;
}

/**
 * 以原子方式将给定值添加到此更新程序管理的给定对象的字段的当前值
 */
public int getAndAdd(T obj, int delta) {
    int prev, next;
    do {
        prev = get(obj);
        next = prev + delta;
    } while (!compareAndSet(obj, prev, next));
    return prev;
}

/**
 * 以原子方式将此更新程序管理的给定对象的字段的当前值增加一
 */
public int incrementAndGet(T obj) {
    int prev, next;
    do {
        prev = get(obj);
        next = prev + 1;
    } while (!compareAndSet(obj, prev, next));
    return next;
}

/**
 * 以原子方式将此更新器管理的给定对象的字段的当前值减一
 */
public int decrementAndGet(T obj) {
    int prev, next;
    do {
        prev = get(obj);
        next = prev - 1;
    } while (!compareAndSet(obj, prev, next));
    return next;
}

/**
 * 以原子方式将给定值添加到此更新程序管理的给定对象的字段的当前值
 */
public int addAndGet(T obj, int delta) {
    int prev, next;
    do {
        prev = get(obj);
        next = prev + delta;
    } while (!compareAndSet(obj, prev, next));
    return next;
}

/**
 * 使用应用给定功能的结果以原子方式自旋式更新此更新程序管理的给定对象的字段,并返回先前的值
 */
public final int getAndUpdate(T obj, IntUnaryOperator updateFunction) {
    int prev, next;
    do {
        prev = get(obj);
        next = updateFunction.applyAsInt(prev);
    } while (!compareAndSet(obj, prev, next));
    return prev;
}

/**
 * 使用应用给定功能的结果以原子方式自旋式更新此更新程序管理的给定对象的字段,并返回更新后的值
 */
public final int updateAndGet(T obj, IntUnaryOperator updateFunction) {
    int prev, next;
    do {
        prev = get(obj);
        next = updateFunction.applyAsInt(prev);
    } while (!compareAndSet(obj, prev, next));
    return next;
}

/**
 * 将给定函数应用于当前值和给定值,并返回前一个值.以原子方式更新此更新程序管理的给定对象的字段
 */
public final int getAndAccumulate(T obj, int x, IntBinaryOperator accumulatorFunction) {
    int prev, next;
    do {
        prev = get(obj);
        next = accumulatorFunction.applyAsInt(prev, x);
    } while (!compareAndSet(obj, prev, next));
    return prev;
}

/**
 * 将给定函数应用于当前值和给定值,并返回更新后的值.以原子方式更新此更新程序管理的给定对象的字段
 */
public final int accumulateAndGet(T obj, int x, IntBinaryOperator accumulatorFunction) {
    int prev, next;
    do {
        prev = get(obj);
        next = accumulatorFunction.applyAsInt(prev, x);
    } while (!compareAndSet(obj, prev, next));
    return next;
}

示例代码

public class AtomicIntegerFieldUpdaterDemo {
    public static class Candidate {
        int id;
        // 如果直接把int改成atomicinteger, 可能对代码破坏比较大
        // 因此使用AtomicIntegerFieldUpdater对score进行封装
        volatile int score;
    }

    // 通过反射实现
    public final static AtomicIntegerFieldUpdater<Candidate> scoreUpdater = AtomicIntegerFieldUpdater.newUpdater(Candidate.class, "score");
    // 检查Updater是否工作正确, allScore的结果应该跟score一致
    public static AtomicInteger allScore = new AtomicInteger(0);

    public static void main(String[] args) throws InterruptedException {
        final Candidate stu = new Candidate();
        Thread[] t = new Thread[10000];
        for (int i = 0; i < 10000; i++) {
            t[i] = new Thread() {
                public void run() {
                    if (Math.random() > 0.4) {
                        scoreUpdater.incrementAndGet(stu);
                        allScore.incrementAndGet();
                    }
                }
            };
            t[i].start();
        }
        for (int i = 0; i < 10000; i++) {
            t[i].join();
        }

        System.out.println("score=" + stu.score);
        System.out.println("allScore=" + allScore);
    }
}

无锁Vector

jdk中Vector是加锁的, 网上找的一个无锁Vector LockFreeVector, 给他添加了源码中文注释.

主要关注push_back, 添加元素的函数

import java.util.AbstractList;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.atomic.AtomicReferenceArray;

/**
 * 它是线程安全且无锁的向量.
 * 该类实现算法来自：
 *     无锁动态可调整大小的阵列
 */
public class LockFreeVector<E> extends AbstractList<E> {
    private static final boolean debug = false;
    /**
     * 第一个存储桶的大小：sizeof(bucket[i+1])=2*sizeof(bucket[i])
     */
    private static final int FIRST_BUCKET_SIZE = 8;

    /**
     * 桶数：30 will allow 8*(2^30-1) elements
     */
    private static final int N_BUCKET = 30;

    /**
     * 我们最多只能有N_BUCKET个桶. 我们有sizeof(buckets.get(i))=FIRST_BUCKET_SIZE**(i+1)
     *
     * 为什么AtomicReferenceArray里再套一个AtomicReferenceArray呢,类似一个篮子(buckets)里放了很多篮子
     * 为了在容量扩展时希望尽可能少的改动原有数据,因此把一维数组扩展成二维数组.
     * 该二维数组并非均衡的分布.可能第一个数组8个元素,第二个数组16个元素,第三个数组32个......
     */
    private final AtomicReferenceArray<AtomicReferenceArray<E>> buckets;

    static class WriteDescriptor<E> {
        public E oldV;
        public E newV;
        public AtomicReferenceArray<E> addr;
        public int addr_ind;

        /**
         * 创建一个新的描述符
         * @param addr Operation address 对哪个数组进行写
         * @param addr_ind  Index of address 指定index
         * @param oldV old operand
         * @param newV new operand
         */
        public WriteDescriptor(AtomicReferenceArray<E> addr, int addr_ind, E oldV, E newV) {
            this.addr = addr;
            this.addr_ind = addr_ind;
            this.oldV = oldV;
            this.newV = newV;
        }

        public void doIt() {
            // 这边失败后重试的逻辑在另外的代码里.
            addr.compareAndSet(addr_ind, oldV, newV);
        }
    }

    static class Descriptor<E> {
        public int size;
        volatile WriteDescriptor<E> writeop;

        public Descriptor(int size, WriteDescriptor<E> writeop) {
            this.size = size;
            this.writeop = writeop;
        }

        public void completeWrite() {
            WriteDescriptor<E> tmpOp = writeop;
            if (tmpOp != null) {
                tmpOp.doIt();
                writeop = null; // 这是安全的,因为所有写入writeop的操作都将null用作r_value
            }
        }
    }

    private AtomicReference<Descriptor<E>> descriptor;
    private static final int zeroNumFirst = Integer.numberOfLeadingZeros(FIRST_BUCKET_SIZE);

    public LockFreeVector() {
        buckets = new AtomicReferenceArray<AtomicReferenceArray<E>>(N_BUCKET);
        buckets.set(0, new AtomicReferenceArray<E>(FIRST_BUCKET_SIZE));
        descriptor = new AtomicReference<Descriptor<E>>(new Descriptor<E>(0,null));
    }

    /**
     * 把元素e加到vector中
     */
    public void push_back(E e) {
        Descriptor<E> desc;
        Descriptor<E> newd;
        do {
            desc = descriptor.get();
            desc.completeWrite();
            // desc.size   Vector 本身的大小
            // FIRST_BUCKET_SIZE  第一个一维数组的大小
            int pos = desc.size + FIRST_BUCKET_SIZE;
            // 取出pos 的前导0
            int zeroNumPos = Integer.numberOfLeadingZeros(pos);
            // zeroNumFirst  为FIRST_BUCKET_SIZE 的前导0
            // bucketInd 数据应该放到哪一个一维数组(篮子)里的
            int bucketInd = zeroNumFirst - zeroNumPos;
            // 00000000 00000000 00000000 00001000 第一个篮子满 8
            // 00000000 00000000 00000000 00011000 第二个篮子满 8 + 16
            // 00000000 00000000 00000000 00111000 第三个篮子满 8 + 16 + 32
            // ... bucketInd其实通过前导0相减, 就是为了得出来当前第几个篮子是空的.

            // 判断这个一维数组是否已经启用, 可能是第一次初始化
            if (buckets.get(bucketInd) == null) {
                //newLen  一维数组的长度, 取前一个数组长度 * 2
                int newLen = 2 * buckets.get(bucketInd - 1).length();
                // 设置失败也没关系, 只要有人初始化成功就行
                buckets.compareAndSet(bucketInd, null,
                        new AtomicReferenceArray<E>(newLen));
            }

            // 在这个一位数组中，我在哪个位置
            // 0x80000000是 10000000 00000000 00000000 00000000
            // 这句话就是把上述111000, 第一个1变成了0, 得到011000, 即新值的位置.
            int idx = (0x80000000>>>zeroNumPos) ^ pos;
            // 通过bucketInd与idx来确定元素在二维数组中的位置
            // 期望写入的时候, 该位置值是null, 如果非null, 说明其他线程已经写了, 则继续循环.
            newd = new Descriptor<E>(desc.size + 1, new WriteDescriptor<E>(
                    buckets.get(bucketInd), idx, null, e));
            // 循环cas设值
        } while (!descriptor.compareAndSet(desc, newd));
        descriptor.get().completeWrite();
    }

    /**
     * 删除向量中的最后一个元素
     */
    public E pop_back() {
        Descriptor<E> desc;
        Descriptor<E> newd;
        E elem;
        do {
            desc = descriptor.get();
            desc.completeWrite();

            int pos = desc.size + FIRST_BUCKET_SIZE - 1;
            int bucketInd = Integer.numberOfLeadingZeros(FIRST_BUCKET_SIZE)
                    - Integer.numberOfLeadingZeros(pos);
            int idx = Integer.highestOneBit(pos) ^ pos;
            elem = buckets.get(bucketInd).get(idx);
            newd = new Descriptor<E>(desc.size - 1, null);
        } while (!descriptor.compareAndSet(desc, newd));

        return elem;
    }

    /**
     * 获取带有索引的元素
     */
    @Override
    public E get(int index) {
        int pos = index + FIRST_BUCKET_SIZE;
        int zeroNumPos = Integer.numberOfLeadingZeros(pos);
        int bucketInd = zeroNumFirst - zeroNumPos;
        int idx = (0x80000000>>>zeroNumPos) ^ pos;
        return buckets.get(bucketInd).get(idx);
    }

    /**
     * 将具有索引的元素设置为e。
     */
    public E set(int index, E e) {
        int pos = index + FIRST_BUCKET_SIZE;
        int bucketInd = Integer.numberOfLeadingZeros(FIRST_BUCKET_SIZE)
                - Integer.numberOfLeadingZeros(pos);
        int idx = Integer.highestOneBit(pos) ^ pos;
        AtomicReferenceArray<E> bucket = buckets.get(bucketInd);
        while (true) {
            E oldV = bucket.get(idx);
            if (bucket.compareAndSet(idx, oldV, e))
                return oldV;
        }
    }

    /**
     * 保留更多空间
     */
    public void reserve(int newSize) {
        int size = descriptor.get().size;
        int pos = size + FIRST_BUCKET_SIZE - 1;
        int i = Integer.numberOfLeadingZeros(FIRST_BUCKET_SIZE)
                - Integer.numberOfLeadingZeros(pos);
        if (i < 1)
            i = 1;

        int initialSize = buckets.get(i - 1).length();
        while (i < Integer.numberOfLeadingZeros(FIRST_BUCKET_SIZE)
                - Integer.numberOfLeadingZeros(newSize + FIRST_BUCKET_SIZE - 1)) {
            i++;
            initialSize *= FIRST_BUCKET_SIZE;
            buckets.compareAndSet(i, null, new AtomicReferenceArray<E>(
                    initialSize));
        }
    }

    /**
     * 向量的大小
     */
    public int size() {
        return descriptor.get().size;
    }
    
    @Override
    public boolean add(E object) {
        push_back(object);
        return true;
    }
}