AtomicIntegerFieldUpdater源码解析

AtomicIntegerFieldUpdater源码解析

AtomicIntegerFieldUpdater类介绍

AtomicIntegerFieldUpdater类是JDK提供的volatile int封装器。可以通过它来原子更新某些对象的volatile修饰的int字段。

看下面这个例子,假设NotUseAtomicIntegerFieldUpdate类有一个字段value,它在某些场景下需要具有原子更新的能力,但又不能使用AtomicInteger(可以认为是出于节省空间的考虑)。这个时候,AtomicIntegerFieldUpdater就派上用场了。

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
26
public class NotUseAtomicIntegerFieldUpdate implements Runnable {
private int value = 0;

@Override
public void run() {
for (int i = 0;i < 1000;i++) {
value++;
}
}

public static void main(String[] args) {
NotUseAtomicIntegerFieldUpdate update = new NotUseAtomicIntegerFieldUpdate();
for (int i = 0;i < 10;i++) {
Thread t = new Thread(update);
t.start();
}

try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}

System.out.println("Final value is " + update.value);
}
}

运行结果1

运行结果2

运行结果3

类图

AtomicIntegerFieldUpdater类图

主要属性

无。主要属性在AtomicIntegerFieldUpdaterImpl中。这里就拿AtomicIntegerFieldUpdaterImpl的属性来说明

1
2
3
4
5
6
7
8
9
private static final sun.misc.Unsafe U = sun.misc.Unsafe.getUnsafe();
private final long offset;
/**
* if field is protected, the subclass constructing updater, else
* the same as tclass
*/
private final Class<?> cclass;
/** class holding the field */
private final Class<T> tclass;

U: Unsafe类,实现CAS算法

offset:Field字段在整个AtomicIntegerFieldUpdaterImpl中的内存偏移。

cclass:如果字段受保护,cclass为调用者类的Class对象,否则cclass等于tclass

tclass:被操作对象类的Class对象

主要方法

newUpdater(Class, String)

1
2
3
4
5
public static <U> AtomicIntegerFieldUpdater<U> newUpdater(Class<U> tclass,
String fieldName) {
return new AtomicIntegerFieldUpdaterImpl<U>
(tclass, fieldName, Reflection.getCallerClass());
}

静态方法,创建一个AtomicIntegerFieldUpdater对象。

AtomicIntegerFieldUpdater()

1
2
3
4
5
/**
* Protected do-nothing constructor for use by subclasses.
*/
protected AtomicIntegerFieldUpdater() {
}

受限的构造函数,不做什么。

compareAndSet(T,int,int)

1
public abstract boolean compareAndSet(T obj, int expect, int update);

抽象方法,在AtomicIntegerFieldUpdaterImpl中实现

weakCompareAndSet(T,int,int)

1
public abstract boolean weakCompareAndSet(T obj, int expect, int update);

抽象方法,在AtomicIntegerFieldUpdaterImpl中实现

set(T, int)

1
public abstract void set(T obj, int newValue);

抽象方法,在AtomicIntegerFieldUpdaterImpl中实现

lazySet(T,int)

1
public abstract void lazySet(T obj, int newValue);

抽象方法,在AtomicIntegerFieldUpdaterImpl中实现

get(T)

1
public abstract int get(T obj);

抽象方法,在AtomicIntegerFieldUpdaterImpl中实现

getAndSet(T,int)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/**
* Atomically sets the field of the given object managed by this updater
* to the given value and returns the old value.
*
* @param obj An object whose field to get and set
* @param newValue the new value
* @return the previous value
*/
public int getAndSet(T obj, int newValue) {
int prev;
do {
prev = get(obj);
} while (!compareAndSet(obj, prev, newValue));
return prev;
}

返回当前值,并将当前值设置为newValue。这里是一个死循环,极端情况下,可能导致自旋过久。

getAndIncrement(T)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/**
* Atomically increments by one the current value of the field of the
* given object managed by this updater.
*
* @param obj An object whose field to get and set
* @return the previous value
*/
public int getAndIncrement(T obj) {
int prev, next;
do {
prev = get(obj);
next = prev + 1;
} while (!compareAndSet(obj, prev, next));
return prev;
}

返回当前值,并将当前值加上1.此处是一个死循环处理,极端情况下可能会自旋过久,

getAndDecrement(T)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/**
* Atomically decrements by one the current value of the field of the
* given object managed by this updater.
*
* @param obj An object whose field to get and set
* @return the previous value
*/
public int getAndDecrement(T obj) {
int prev, next;
do {
prev = get(obj);
next = prev - 1;
} while (!compareAndSet(obj, prev, next));
return prev;
}

返回当前值,并将当前值减去1。此处是一个死循环处理,极端情况下可能会自旋过久。

getAndAdd(T, int)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
/**
* Atomically adds the given value to the current value of the field of
* the given object managed by this updater.
*
* @param obj An object whose field to get and set
* @param delta the value to add
* @return the previous value
*/
public int getAndAdd(T obj, int delta) {
int prev, next;
do {
prev = get(obj);
next = prev + delta;
} while (!compareAndSet(obj, prev, next));
return prev;
}

返回当前值,并将当前值加上delta。此处是一个死循环处理,极端情况下可能会自旋过久。

incrementAndGet(T)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/**
* Atomically increments by one the current value of the field of the
* given object managed by this updater.
*
* @param obj An object whose field to get and set
* @return the updated value
*/
public int incrementAndGet(T obj) {
int prev, next;
do {
prev = get(obj);
next = prev + 1;
} while (!compareAndSet(obj, prev, next));
return next;
}

将当前的值自增1,并返回。此处是一个死循环处理,极端情况下可能会自旋过久。

decrementAndGet(T)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/**
* Atomically decrements by one the current value of the field of the
* given object managed by this updater.
*
* @param obj An object whose field to get and set
* @return the updated value
*/
public int decrementAndGet(T obj) {
int prev, next;
do {
prev = get(obj);
next = prev - 1;
} while (!compareAndSet(obj, prev, next));
return next;
}

将当前的值自减1,并返回。此处是一个死循环处理,极端情况下可能会自旋过久。

addAndGet(T,int)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
/**
* Atomically adds the given value to the current value of the field of
* the given object managed by this updater.
*
* @param obj An object whose field to get and set
* @param delta the value to add
* @return the updated value
*/
public int addAndGet(T obj, int delta) {
int prev, next;
do {
prev = get(obj);
next = prev + delta;
} while (!compareAndSet(obj, prev, next));
return next;
}

将当前值加上delta,并返回。此处是一个死循环处理,极端情况下可能会自旋过久。

getAndUpdate(T,IntUnaryOperator)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
/**
* Atomically updates the field of the given object managed by this updater
* with the results of applying the given function, returning the previous
* value. The function should be side-effect-free, since it may be
* re-applied when attempted updates fail due to contention among threads.
*
* @param obj An object whose field to get and set
* @param updateFunction a side-effect-free function
* @return the previous value
* @since 1.8
*/
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;
}

获取当前值,并将当前值应用单元函数updateFunction,并更新当前值。此处是一个死循环处理,极端情况下可能会自旋过久。

updateAndGet(T,IntUnaryOperator)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
/**
* Atomically updates the field of the given object managed by this updater
* with the results of applying the given function, returning the updated
* value. The function should be side-effect-free, since it may be
* re-applied when attempted updates fail due to contention among threads.
*
* @param obj An object whose field to get and set
* @param updateFunction a side-effect-free function
* @return the updated value
* @since 1.8
*/
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;
}

将当前值应用于单元函数updateFunction,并更新当前值,返回最新的当前值。此处是一个死循环处理,极端情况下可能会自旋过久。

getAndAccumulate(T,int,IntBinaryOperator)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
/**
* Atomically updates the field of the given object managed by this
* updater with the results of applying the given function to the
* current and given values, returning the previous value. The
* function should be side-effect-free, since it may be re-applied
* when attempted updates fail due to contention among threads. The
* function is applied with the current value as its first argument,
* and the given update as the second argument.
*
* @param obj An object whose field to get and set
* @param x the update value
* @param accumulatorFunction a side-effect-free function of two arguments
* @return the previous value
* @since 1.8
*/
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;
}

返回当前值。并将当前值和x应用于二元函数accumulateFunction,更新当前值。此处是一个死循环处理,极端情况下可能会自旋过久。

accumulateAndGet(T,int,IntBinaryOperator)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
/**
* Atomically updates the field of the given object managed by this
* updater with the results of applying the given function to the
* current and given values, returning the updated value. The
* function should be side-effect-free, since it may be re-applied
* when attempted updates fail due to contention among threads. The
* function is applied with the current value as its first argument,
* and the given update as the second argument.
*
* @param obj An object whose field to get and set
* @param x the update value
* @param accumulatorFunction a side-effect-free function of two arguments
* @return the updated value
* @since 1.8
*/
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;
}

将当前值和x应用于二元函数accumulatorFunction,并将结果更新于当前值,返回。此处是一个死循环处理,极端情况下可能会自旋过久。

AtomicIntegerFieldUpdaterImpl#UpdaterImpl(Class,String,Class)

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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
AtomicIntegerFieldUpdaterImpl(final Class<T> tclass,
final String fieldName,
final Class<?> caller) {
final Field field;
final int modifiers;
try {
// 反射获取tclass中的字段fieldName,如果字段不存在的话,会抛出异常
field = AccessController.doPrivileged(
new PrivilegedExceptionAction<Field>() {
public Field run() throws NoSuchFieldException {
return tclass.getDeclaredField(fieldName);
}
});
// 获取字段的访问修饰符
modifiers = field.getModifiers();
// 校验该字段的访问权限。fieldName对应的字段必须是可访问的
// 变量的描述符类型必须与调用者一致。如果调用者能够调用变量就能够通过反射操作保证原子性。
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);
}
// 如果对象类型不是int,抛出异常
if (field.getType() != int.class)
throw new IllegalArgumentException("Must be integer type");

// 如果字段不是被volatile修饰的,抛出异常
if (!Modifier.isVolatile(modifiers))
throw new IllegalArgumentException("Must be volatile type");

// Access to protected field members is restricted to receivers only
// of the accessing class, or one of its subclasses, and the
// accessing class must in turn be a subclass (or package sibling)
// of the protected member's defining class.
// If the updater refers to a protected field of a declaring class
// outside the current package, the receiver argument will be
// narrowed to the type of the accessing class.
this.cclass = (Modifier.isProtected(modifiers) && // 字段的修饰符是否是protect
tclass.isAssignableFrom(caller) && // 判断tclass和caller是否相同或者是另一个类的子类或接口
!isSamePackage(tclass, caller)) //是否具有相同的ClassLoader和包名称
? caller : tclass;
this.tclass = tclass;
// 获取该字段在对象内存中的偏移量
this.offset = U.objectFieldOffset(field);
}

从上述代码中可以看到,如果字段想要被AtomicIntegerFieldUpdater修改的话,必须要满足下面几个条件:

  1. 字段必须是volatile类型的,在线程之间共享变量时保证立即可见.eg:volatile int value = 3
  2. 字段的描述类型(修饰符public/protected/default/private)是与调用者与操作对象字段的关系一致。也就是说调用者能够直接操作对象字段,那么就可以反射进行原子操作。但是对于父类的字段,子类是不能直接操作的,尽管子类可以访问父类的字段。
  3. 只能是实例变量,不能是类变量,也就是说不能加static关键字。
  4. 只能是可修改变量,不能使final变量,因为final的语义就是不可修改。实际上final的语义和volatile是有冲突的,这两个关键字不能同时存在。
  5. 对于AtomicIntegerFieldUpdater和AtomicLongFieldUpdater只能修改int/long类型的字段,不能修改其包装类型(Integer/Long)。如果要修改包装类型就需要使用AtomicReferenceFieldUpdater。

AtomicIntegerFieldUpdaterImpl#isAncestor(ClassLoader, ClassLoader)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
/**
* Returns true if the second classloader can be found in the first
* classloader's delegation chain.
* Equivalent to the inaccessible: 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,否则false

AtomicIntegerFieldUpdaterImpl#isSamePackage(Class, Class)

1
2
3
4
5
6
7
8
/**
* Returns true if the two classes have the same class loader and
* package qualifier
*/
private static boolean isSamePackage(Class<?> class1, Class<?> class2) {
return class1.getClassLoader() == class2.getClassLoader()
&& Objects.equals(getPackageName(class1), getPackageName(class2));
}

判断两个类是否是同一个类加载器加载的并且同属于一个包

AtomicIntegerFieldUpdaterImpl#getPackageName(Class)

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

返回类的包名

AtomicIntegerFieldUpdaterImpl#accessCheck(T)

1
2
3
4
5
6
7
8
/**
* Checks that target argument is instance of cclass. On
* failure, throws cause.
*/
private final void accessCheck(T obj) {
if (!cclass.isInstance(obj))
throwAccessCheckException(obj);
}

判断对象obj是否是cclass的实例,不是的话抛出AccessCheckException

AtomicIntegerFieldUpdaterImpl#throwAccessCheckException(T)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
/**
* Throws access exception if accessCheck failed due to
* protected access, else ClassCastException.
*/
private final void throwAccessCheckException(T obj) {
if (cclass == tclass)
throw new ClassCastException();
else
throw new RuntimeException(
new IllegalAccessException(
"Class " +
cclass.getName() +
" can not access a protected member of class " +
tclass.getName() +
" using an instance of " +
obj.getClass().getName()));
}

如果访问检查失败或者Class转换失败,抛出异常。

AtomicIntegerFieldUpdaterImpl#compareAndSet(T,int,int)

1
2
3
4
public final boolean compareAndSet(T obj, int expect, int update) {
accessCheck(obj);
return U.compareAndSwapInt(obj, offset, expect, update);
}

如果当前值等于期望值except,就把当前值设置为update,返回true,否则返回false。

AtomicIntegerFieldUpdaterImpl#weakCompareAndSet(T,int,int)

1
2
3
4
public final boolean weakCompareAndSet(T obj, int expect, int update) {
accessCheck(obj);
return U.compareAndSwapInt(obj, offset, expect, update);
}

从父类的注释上看,这个方法的本意是,调用weakCompareAndSet方法时不能保证指令重排的发生,因此,这个方法有时候会毫无理由地失败。

AtomicIntegerFieldUpdaterImpl#set(T,int)

1
2
3
4
public final void set(T obj, int newValue) {
accessCheck(obj);
U.putIntVolatile(obj, offset, newValue);
}

将对象obj的fieldName字段设置为newValue

AtomicIntegerFieldUpdaterImpl#lazySet(T, int)

1
2
3
4
public final void lazySet(T obj, int newValue) {
accessCheck(obj);
U.putOrderedInt(obj, offset, newValue);
}

该方法是“慢慢”地将对象的obj的fieldName字段设置为newValue。这就意味着有段时间,多线程环境下,会有线程读取到旧值。

AtomicIntegerFieldUpdaterImpl#get(T)

1
2
3
4
public final int get(T obj) {
accessCheck(obj);
return U.getIntVolatile(obj, offset);
}

获取对象obj的fieldName字段的值

AtomicIntegerFieldUpdaterImpl#getAndSet(T,int)

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

返回当前值,并将当前值设置为newValue

AtomicIntegerFieldUpdaterImpl#getAndAdd(T,int)

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

返回当前值,并将当前值加上delta,更新。

AtomicIntegerFieldUpdaterImpl#getAndIncrement(T)

1
2
3
public final int getAndIncrement(T obj) {
return getAndAdd(obj, 1);
}

返回当前值,并将当前值加上1,更新当前值

AtomicIntegerFieldUpdaterImpl#getAndDecrement(T)

1
2
3
public final int getAndDecrement(T obj) {
return getAndAdd(obj, -1);
}

返回当前值,并将当前值减1,更新当前值。

AtomicIntegerFieldUpdaterImpl#incrementAndGet(T)

1
2
3
public final int incrementAndGet(T obj) {
return getAndAdd(obj, 1) + 1;
}

将当前值加1,并更新,返回当前值。

AtomicIntegerFieldUpdaterImpl#decrementAndGet(T)

1
2
3
public final int decrementAndGet(T obj) {
return getAndAdd(obj, -1) - 1;
}

将当前值加减1,并更新,返回当前值。

AtomicIntegerFieldUpdaterImpl#addAndGet(T,int)

1
2
3
public final int addAndGet(T obj, int delta) {
return getAndAdd(obj, delta) + delta;
}

将当前值加上delta,更新,并返回更新后的当前值。

应用

我们将上面的代码改写下,用一个静态的AtomicIntegerFieldUpdater对象来持有value字段,进行原子更新。

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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
public class NotUseAtomicIntegerFieldUpdate {
// 注意修饰符需要和下面的holder修饰符一致,并且要用volatile字段修饰
public volatile int value = 0;

public int getValue() {
return value;
}

public void setValue(int value) {
this.value = value;
}
}

public class UseAtomicIntegerFieldUpdater implements Runnable {
public static NotUseAtomicIntegerFieldUpdate update = new NotUseAtomicIntegerFieldUpdate();
public static AtomicIntegerFieldUpdater<NotUseAtomicIntegerFieldUpdate> holder = AtomicIntegerFieldUpdater.newUpdater(NotUseAtomicIntegerFieldUpdate.class, "value");
@Override
public void run() {
for (int i = 0;i < 1000;i++) {
holder.incrementAndGet(update);
}
}

public static void main(String[] args) {
UseAtomicIntegerFieldUpdater updater = new UseAtomicIntegerFieldUpdater();

for (int i = 0;i < 10;i++) {
Thread t = new Thread(updater);
t.start();
}

try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}

System.out.println("Final value of update is " + updater.update.getValue());
System.out.println("Final value of holder is " + updater.holder.get(update));

}
}

现在无论再怎么运行,结果都是一致的了。

运行结果4