AtomicLongFieldUpdater源码解析

Posted on | In , ,
Words count in article: 3.8k | Reading time ≈ 20

AtomicLongFieldUpdater源码解析

AtomicLongFieldUpdater类介绍

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

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

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
public class NotUseAtomicLongFieldUpdate implements Runnable {
	private long value = 0;

	@Override
	public void run() {
		for (int i = 0;i < 1000;i++) {
			value++;
		}
	}
	
	public static void main(String[] args) {
		NotUseAtomicLongFieldUpdate update = new NotUseAtomicLongFieldUpdate();
		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

类图

AtomicLongFieldUpdater

主要属性

无。主要属性在CASUpdater或者LockedUpdater中。这里就拿CASUpdater的属性来说明

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
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
/**
     * Creates and returns an updater for objects with the given field.
     * The Class argument is needed to check that reflective types and
     * generic types match.
     *
     * @param tclass the class of the objects holding the field
     * @param fieldName the name of the field to be updated
     * @param <U> the type of instances of tclass
     * @return the updater
     * @throws IllegalArgumentException if the field is not a
     * volatile long type
     * @throws RuntimeException with a nested reflection-based
     * exception if the class does not hold field or is the wrong type,
     * or the field is inaccessible to the caller according to Java language
     * access control
     */
    @CallerSensitive
    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);
    }

新建一个Updater返回。如果平台支持CAS，则返回CASUpdater否则返回LockedUpdater

AtomicLongFieldUpdater()

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

空函数

compareAndSet（T,long,long)

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 updated value if the current value {@code ==} the
     * expected value. This method is guaranteed to be atomic with respect to
     * other calls to {@code compareAndSet} and {@code set}, but not
     * necessarily with respect to other changes in the field.
     *
     * @param obj An object whose field to conditionally set
     * @param expect the expected value
     * @param update the new value
     * @return {@code true} if successful
     * @throws ClassCastException if {@code obj} is not an instance
     * of the class possessing the field established in the constructor
     */
    public abstract boolean compareAndSet(T obj, long expect, long update);

抽象函数，具体实现可看CASUpdater或者LockedUpdater。

weakCompareAndSet(T,long,long)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
/**
     * Atomically sets the field of the given object managed by this updater
     * to the given updated value if the current value {@code ==} the
     * expected value. This method is guaranteed to be atomic with respect to
     * other calls to {@code compareAndSet} and {@code set}, but not
     * necessarily with respect to other changes in the field.
     *
     * <p><a href="package-summary.html#weakCompareAndSet">May fail
     * spuriously and does not provide ordering guarantees</a>, so is
     * only rarely an appropriate alternative to {@code compareAndSet}.
     *
     * @param obj An object whose field to conditionally set
     * @param expect the expected value
     * @param update the new value
     * @return {@code true} if successful
     * @throws ClassCastException if {@code obj} is not an instance
     * of the class possessing the field established in the constructor
     */
    public abstract boolean weakCompareAndSet(T obj, long expect, long update);

抽象函数，具体实现可看CASUpdater或者LockedUpdater。

set(T,long)

1
2
3
4
5
6
7
8
9
/**
     * Sets the field of the given object managed by this updater to the
     * given updated value. This operation is guaranteed to act as a volatile
     * store with respect to subsequent invocations of {@code compareAndSet}.
     *
     * @param obj An object whose field to set
     * @param newValue the new value
     */
    public abstract void set(T obj, long newValue);

抽象函数，具体实现可看CASUpdater或者LockedUpdater。

lazySet(T,long)

1
2
3
4
5
6
7
8
9
/**
     * Eventually sets the field of the given object managed by this
     * updater to the given updated value.
     *
     * @param obj An object whose field to set
     * @param newValue the new value
     * @since 1.6
     */
    public abstract void lazySet(T obj, long newValue);

抽象函数，具体实现可看CASUpdater或者LockedUpdater。

long get(T)

1
2
3
4
5
6
7
8
/**
     * Gets the current value held in the field of the given object managed
     * by this updater.
     *
     * @param obj An object whose field to get
     * @return the current value
     */
    public abstract long get(T obj);

抽象函数，具体实现可看CASUpdater或者LockedUpdater。

getAndSet(T,long)

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 long getAndSet(T obj, long newValue) {
        long 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 long getAndIncrement(T obj) {
        long 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 long getAndDecrement(T obj) {
        long prev, next;
        do {
            prev = get(obj);
            next = prev - 1;
        } while (!compareAndSet(obj, prev, next));
        return prev;
    }

获取当前值，返回。并把当前值减1，更新。这里是一个死循环，极端情况下，可能导致自旋过久。

getAndAdd(T,long)

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 long getAndAdd(T obj, long delta) {
        long 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 long incrementAndGet(T obj) {
        long 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 long decrementAndGet(T obj) {
        long prev, next;
        do {
            prev = get(obj);
            next = prev - 1;
        } while (!compareAndSet(obj, prev, next));
        return next;
    }

将当前值减1并更新，返回当前值。这里是一个死循环，极端情况下，可能导致自旋过久。

addAndGet(T,long)

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 long addAndGet(T obj, long delta) {
        long prev, next;
        do {
            prev = get(obj);
            next = prev + delta;
        } while (!compareAndSet(obj, prev, next));
        return next;
    }

将当前值加上delta，更新，并返回。这里是一个死循环，极端情况下，可能导致自旋过久。

getAndUpdate(T,LongUnaryOperator)

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 long getAndUpdate(T obj, LongUnaryOperator updateFunction) {
        long prev, next;
        do {
            prev = get(obj);
            next = updateFunction.applyAsLong(prev);
        } while (!compareAndSet(obj, prev, next));
        return prev;
    }

返回当前值，并将单元函数应用于当前值，更新。这里是一个死循环，极端情况下，可能导致自旋过久。

updateAndGet(T,LongUnaryOperator)

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 long updateAndGet(T obj, LongUnaryOperator updateFunction) {
        long prev, next;
        do {
            prev = get(obj);
            next = updateFunction.applyAsLong(prev);
        } while (!compareAndSet(obj, prev, next));
        return next;
    }

将当前值应用于单元函数，并更新，返回。这里是一个死循环，极端情况下，可能导致自旋过久。

getAndAccumulate(T,long,LongBinaryOperator)

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 long getAndAccumulate(T obj, long x,
                                       LongBinaryOperator accumulatorFunction) {
        long prev, next;
        do {
            prev = get(obj);
            next = accumulatorFunction.applyAsLong(prev, x);
        } while (!compareAndSet(obj, prev, next));
        return prev;
    }

返回当前值，并对当前值和x应用双元函数，将得到的结果更新当前值。这里是一个死循环，极端情况下，可能导致自旋过久。

accumulateAndGet(T,long,LongBinaryOperator)

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 long accumulateAndGet(T obj, long x,
                                       LongBinaryOperator accumulatorFunction) {
        long prev, next;
        do {
            prev = get(obj);
            next = accumulatorFunction.applyAsLong(prev, x);
        } while (!compareAndSet(obj, prev, next));
        return next;
    }

对当前值和x应用双元函数，并将得到的结果更新至当前值，并返回。

CASUpdater源码

CASUpdater(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
CASUpdater(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();
                // 确认当前字段和AtomicLongFieldUpdater具有同样的访问级别
                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);
            }

            // 只能是long型
            if (field.getType() != long.class)
                throw new IllegalArgumentException("Must be long 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) &&
                           tclass.isAssignableFrom(caller) &&
                           !isSamePackage(tclass, caller))
                          ? caller : tclass;
            this.tclass = tclass;
            this.offset = U.objectFieldOffset(field);
        }

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

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

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的实例

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

compareAndSet(T,long,long)

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

如果当前值是expect，就通过调用unsafe的cas相关算法，更新当前值为update，返回true，否则返回false。

weakCompareAndSet(T,long,long)

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

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

但是当前实现就是compareAndSet。

set(T,long)

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

设置当前值为newValue

lazySet(T,long)

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

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

get(T)

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

获取当前值

getAndSet(T,long)

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

返回当前值，并将当前值更新为newValue

getAndAdd(T,long)

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

返回当前值，并将当前值加上delta

getAndIncrement(T)

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

返回当前值，并将当前值加1

getAndDecrement(T)

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

返回当前值，并将当前值减1

incrementAndGet(T)

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

将当前值加1，并返回

decrementAndGet(T)

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

将当前值减1，并返回

addAndGet(T,long)

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

将当前值加上delta，并返回。

LockedUpdater

由于LockUpdater部分方法与CASUpdater相同，这里不再赘述。仅列出和CASUpdater实现不同的方法。

compareAndSet(T,long,long)

1
2
3
4
5
6
7
8
9
10
public final boolean compareAndSet(T obj, long expect, long update) {
            accessCheck(obj);
            synchronized (this) {
                long v = U.getLong(obj, offset);
                if (v != expect)
                    return false;
                U.putLong(obj, offset, update);
                return true;
            }
        }

这里和CASUpdater不同，这里使用了锁。在更新当前值的时候，先锁上当前实例，如果当前值不等于expect，返回false，否则调用unsafe接口，更新当前值为update。

weakCompareAndSet(T,long,long)

1
2
3
public final boolean weakCompareAndSet(T obj, long expect, long update) {
            return compareAndSet(obj, expect, update);
        }

直接调用的compareAndSet方法

set(T,long)

1
2
3
4
5
6
public final void set(T obj, long newValue) {
            accessCheck(obj);
            synchronized (this) {
                U.putLong(obj, offset, newValue);
            }
        }

更新当前值的时候锁住实例。

lazySet(T,long)

1
2
3
public final void lazySet(T obj, long newValue) {
            set(obj, newValue);
        }

直接调用set方法

get(T)

1
2
3
4
5
6
public final long get(T obj) {
            accessCheck(obj);
            synchronized (this) {
                return U.getLong(obj, offset);
            }
        }

取值的时候，锁住当前实例，再调用Unsafe的方法获取值。

应用

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
public class NotUseAtomicLongFieldUpdate {
	public volatile long value = 0;
}


public class UseAtomicLongFieldUpdater implements Runnable {
	public static NotUseAtomicLongFieldUpdate update = new NotUseAtomicLongFieldUpdate();
	public static AtomicLongFieldUpdater<NotUseAtomicLongFieldUpdate> holder = AtomicLongFieldUpdater.newUpdater(NotUseAtomicLongFieldUpdate.class, "value");
	@Override
	public void run() {
		for (int i = 0;i < 1000;i++) {
			holder.incrementAndGet(update);
		}
	}
	
	public static void main(String[] args) {
		UseAtomicLongFieldUpdater updater = new UseAtomicLongFieldUpdater();
		
		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.value);
		System.out.println("Final value of holder is " + updater.holder.get(update));
		
	}
}

0%