下面列出了怎么用java.util.concurrent.atomic.AtomicIntegerFieldUpdater的API类实例代码及写法,或者点击链接到github查看源代码。
/**
* Offer {@code item} to the queue, try to acquire the processing lock, and if successful return an item for
* single-consumer style processing. If non-{@code null} is returned the caller is responsible for releasing
* the lock!
* @param queue The {@link Queue#offer(Object)} and {@link Queue#poll()} (assuming lock was acquired).
* @param lockUpdater Used to acquire the lock via
* {@link ConcurrentUtils#tryAcquireLock(AtomicIntegerFieldUpdater, Object)}.
* @param item The item to {@link Queue#offer(Object)}.
* @param <T> The type of item in the {@link Queue}.
* @return {@code null} if the queue was empty, or the lock couldn't be acquired. otherwise the lock has been
* acquired and it is the caller's responsibility to release!
*/
@Nullable
private <T> T addAndTryPoll(final Queue<T> queue,
@SuppressWarnings("rawtypes") final AtomicIntegerFieldUpdater<NettyPipelinedConnection> lockUpdater, T item) {
queue.add(item);
while (tryAcquireLock(lockUpdater, this)) {
// exceptions are not expected from poll, and if they occur we can't reliably recover which would involve
// draining the queue. just throw with the lock poisoned, callers will propagate the exception to related
// subscriber and close the connection.
final T next = queue.poll();
if (next != null) {
return next; // lock must be released when the returned task completes!
} else if (releaseLock(lockUpdater, this)) {
return null;
}
}
return null;
}
/**
* Poll the {@code queue} and attempt to process an item. The lock must be acquired on entry into this method and
* if this method return non-{@code null} the lock will not be released (caller's responsibility to later release)
* to continue the single-consumer style processing.
* @param queue The queue to {@link Queue#poll()}.
* @param lockUpdater Used to release via
* {@link ConcurrentUtils#releaseLock(AtomicIntegerFieldUpdater, Object)} if the queue is empty
* @param <T> The type of item in the {@link Queue}.
* @return {@code null} if the queue was empty. otherwise the lock remains acquired and it is the caller's
* responsibility to release (via subsequent calls to this method).
*/
@Nullable
private <T> T pollWithLockAcquired(final Queue<T> queue,
@SuppressWarnings("rawtypes") final AtomicIntegerFieldUpdater<NettyPipelinedConnection> lockUpdater) {
// the lock has been acquired!
try {
do {
final T next = queue.poll();
if (next != null) {
return next; // lock must be released when the returned task completes!
} else if (releaseLock(lockUpdater, this)) {
return null;
}
} while (tryAcquireLock(lockUpdater, this));
return null;
} catch (Throwable cause) {
// exceptions are not expected from poll, and if they occur we can't reliably recover which would involve
// draining the queue. just throw with the lock poisoned and close the connection.
connection.closeAsync().subscribe();
throw cause;
}
}
/**
* compareAndSet in one thread enables another waiting for value
* to succeed
*/
public void testCompareAndSetInMultipleThreads() throws Exception {
x = 1;
final AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a;
a = updaterFor("x");
Thread t = new Thread(new CheckedRunnable() {
public void realRun() {
while (!a.compareAndSet(AtomicIntegerFieldUpdaterTest.this, 2, 3))
Thread.yield();
}});
t.start();
assertTrue(a.compareAndSet(this, 1, 2));
t.join(LONG_DELAY_MS);
assertFalse(t.isAlive());
assertEquals(3, a.get(this));
}
/**
* compareAndSet in one thread enables another waiting for value
* to succeed
*/
public void testCompareAndSetInMultipleThreads() throws Exception {
x = 1;
final AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a;
a = updaterFor("x");
Thread t = new Thread(new CheckedRunnable() {
public void realRun() {
while (!a.compareAndSet(AtomicIntegerFieldUpdaterTest.this, 2, 3))
Thread.yield();
}});
t.start();
assertTrue(a.compareAndSet(this, 1, 2));
t.join(LONG_DELAY_MS);
assertFalse(t.isAlive());
assertEquals(3, a.get(this));
}
private static AtomicHelper getAtomicHelper() {
AtomicHelper helper;
try {
helper =
new FieldUpdaterAtomicHelper(
AtomicIntegerFieldUpdater.newUpdater(SerializingExecutor.class, "runState"));
} catch (Throwable t) {
log.log(Level.SEVERE, "FieldUpdaterAtomicHelper failed", t);
helper = new SynchronizedAtomicHelper();
}
return helper;
}
/**
* Acquire a lock that is exclusively held with no re-entry, but attempts to acquire the lock while it is
* held can be detected by {@link #releaseLock(AtomicIntegerFieldUpdater, Object)}.
* @param lockUpdater The {@link AtomicIntegerFieldUpdater} used to control the lock state.
* @param owner The owner of the lock object.
* @param <T> The type of object that owns the lock.
* @return {@code true} if the lock was acquired, {@code false} otherwise.
*/
public static <T> boolean tryAcquireLock(AtomicIntegerFieldUpdater<T> lockUpdater, T owner) {
for (;;) {
final int prevEmitting = lockUpdater.get(owner);
if (prevEmitting == CONCURRENT_IDLE) {
if (lockUpdater.compareAndSet(owner, CONCURRENT_IDLE, CONCURRENT_EMITTING)) {
return true;
}
} else if (lockUpdater.compareAndSet(owner, prevEmitting, CONCURRENT_PENDING)) {
return false;
}
}
}
public static void main(String[] args) {
Person person = new Person("zhangsan", 11, 170);
person.setHobby(new Hobby("打球", "足球,篮球"));
AtomicIntegerFieldUpdater<Person> atomicIntegerFieldUpdater = AtomicIntegerFieldUpdater.newUpdater(Person.class, "age");
atomicIntegerFieldUpdater.addAndGet(person, 12);
AtomicLongFieldUpdater<Person> atomicLongFieldUpdater = AtomicLongFieldUpdater.newUpdater(Person.class, "height");
atomicLongFieldUpdater.addAndGet(person, 180);
AtomicReferenceFieldUpdater<Person, Hobby> atomicReferenceFieldUpdater = AtomicReferenceFieldUpdater.newUpdater(Person.class, Hobby.class, "hobby");
atomicReferenceFieldUpdater.getAndSet(person, new Hobby("打球", "排球,羽毛球"));
}
/**
* AtomicIntegerFieldUpdater getAndUpdate returns previous value and updates
* result of supplied function
*/
public void testIntegerFieldUpdaterGetAndUpdate() {
AtomicIntegerFieldUpdater a = anIntFieldUpdater();
a.set(this, 1);
assertEquals(1, a.getAndUpdate(this, Atomic8Test::addInt17));
assertEquals(18, a.getAndUpdate(this, Atomic8Test::addInt17));
assertEquals(35, a.get(this));
assertEquals(35, anIntField);
}
/**
* AtomicIntegerFieldUpdater updateAndGet updates with supplied function and
* returns result.
*/
public void testIntegerFieldUpdaterUpdateAndGet() {
AtomicIntegerFieldUpdater a = anIntFieldUpdater();
a.set(this, 1);
assertEquals(18, a.updateAndGet(this, Atomic8Test::addInt17));
assertEquals(35, a.updateAndGet(this, Atomic8Test::addInt17));
assertEquals(35, a.get(this));
assertEquals(35, anIntField);
}
/**
* AtomicIntegerFieldUpdater getAndAccumulate returns previous value
* and updates with supplied function.
*/
public void testIntegerFieldUpdaterGetAndAccumulate() {
AtomicIntegerFieldUpdater a = anIntFieldUpdater();
a.set(this, 1);
assertEquals(1, a.getAndAccumulate(this, 2, Integer::sum));
assertEquals(3, a.getAndAccumulate(this, 3, Integer::sum));
assertEquals(6, a.get(this));
assertEquals(6, anIntField);
}
/**
* AtomicIntegerFieldUpdater accumulateAndGet updates with supplied
* function and returns result.
*/
public void testIntegerFieldUpdaterAccumulateAndGet() {
AtomicIntegerFieldUpdater a = anIntFieldUpdater();
a.set(this, 1);
assertEquals(7, a.accumulateAndGet(this, 6, Integer::sum));
assertEquals(10, a.accumulateAndGet(this, 3, Integer::sum));
assertEquals(10, a.get(this));
assertEquals(10, anIntField);
}
/**
* Object arguments for parameters of type T that are not
* instances of the class passed to the newUpdater call will
* result in a ClassCastException being thrown.
*/
public void testFieldUpdaters_ClassCastException() {
// Use raw types to allow passing wrong object type, provoking CCE
final AtomicLongFieldUpdater longUpdater = aLongFieldUpdater();
final AtomicIntegerFieldUpdater intUpdater = anIntFieldUpdater();
final AtomicReferenceFieldUpdater refUpdater = anIntegerFieldUpdater();
final Object obj = new Object();
for (Object x : new Object[]{ new Object(), null }) {
Runnable[] throwingActions = {
() -> longUpdater.get(x),
() -> intUpdater.get(x),
() -> refUpdater.get(x),
() -> longUpdater.set(x, 17L),
() -> intUpdater.set(x, 17),
() -> refUpdater.set(x, (Integer) 17),
() -> longUpdater.addAndGet(x, 17L),
() -> intUpdater.addAndGet(x, 17),
() -> longUpdater.getAndUpdate(x, y -> y),
() -> intUpdater.getAndUpdate(x, y -> y),
() -> refUpdater.getAndUpdate(x, y -> y),
() -> longUpdater.compareAndSet(x, 17L, 42L),
() -> intUpdater.compareAndSet(x, 17, 42),
() -> refUpdater.compareAndSet(x, (Integer) 17, (Integer) 42),
};
assertThrows(ClassCastException.class, throwingActions);
}
}
public void checkPrivateAccess() {
try {
AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a =
AtomicIntegerFieldUpdater.newUpdater
(AtomicIntegerFieldUpdaterTest.class, "privateField");
shouldThrow();
} catch (RuntimeException success) {
assertNotNull(success.getCause());
}
}
public void checkCompareAndSetProtectedSub() {
AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a =
AtomicIntegerFieldUpdater.newUpdater
(AtomicIntegerFieldUpdaterTest.class, "protectedField");
this.protectedField = 1;
assertTrue(a.compareAndSet(this, 1, 2));
assertTrue(a.compareAndSet(this, 2, -4));
assertEquals(-4, a.get(this));
assertFalse(a.compareAndSet(this, -5, 7));
assertEquals(-4, a.get(this));
assertTrue(a.compareAndSet(this, -4, 7));
assertEquals(7, a.get(this));
}
public void checkPackageAccess(AtomicIntegerFieldUpdaterTest obj) {
obj.x = 72;
AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a =
AtomicIntegerFieldUpdater.newUpdater
(AtomicIntegerFieldUpdaterTest.class, "x");
assertEquals(72, a.get(obj));
assertTrue(a.compareAndSet(obj, 72, 73));
assertEquals(73, a.get(obj));
}
public void checkPrivateAccess(AtomicIntegerFieldUpdaterTest obj) {
try {
AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a =
AtomicIntegerFieldUpdater.newUpdater
(AtomicIntegerFieldUpdaterTest.class, "privateField");
throw new AssertionError("should throw");
} catch (RuntimeException success) {
assertNotNull(success.getCause());
}
}
/**
* get returns the last value set or assigned
*/
public void testGetSet() {
AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a;
a = updaterFor("x");
x = 1;
assertEquals(1, a.get(this));
a.set(this, 2);
assertEquals(2, a.get(this));
a.set(this, -3);
assertEquals(-3, a.get(this));
}
/**
* get returns the last value lazySet by same thread
*/
public void testGetLazySet() {
AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a;
a = updaterFor("x");
x = 1;
assertEquals(1, a.get(this));
a.lazySet(this, 2);
assertEquals(2, a.get(this));
a.lazySet(this, -3);
assertEquals(-3, a.get(this));
}
/**
* compareAndSet succeeds in changing value if equal to expected else fails
*/
public void testCompareAndSet() {
AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a;
a = updaterFor("x");
x = 1;
assertTrue(a.compareAndSet(this, 1, 2));
assertTrue(a.compareAndSet(this, 2, -4));
assertEquals(-4, a.get(this));
assertFalse(a.compareAndSet(this, -5, 7));
assertEquals(-4, a.get(this));
assertTrue(a.compareAndSet(this, -4, 7));
assertEquals(7, a.get(this));
}
/**
* compareAndSet succeeds in changing protected field value if
* equal to expected else fails
*/
public void testCompareAndSetProtected() {
AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a;
a = updaterFor("protectedField");
protectedField = 1;
assertTrue(a.compareAndSet(this, 1, 2));
assertTrue(a.compareAndSet(this, 2, -4));
assertEquals(-4, a.get(this));
assertFalse(a.compareAndSet(this, -5, 7));
assertEquals(-4, a.get(this));
assertTrue(a.compareAndSet(this, -4, 7));
assertEquals(7, a.get(this));
}
/**
* repeated weakCompareAndSet succeeds in changing value when equal
* to expected
*/
public void testWeakCompareAndSet() {
AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a;
a = updaterFor("x");
x = 1;
do {} while (!a.weakCompareAndSet(this, 1, 2));
do {} while (!a.weakCompareAndSet(this, 2, -4));
assertEquals(-4, a.get(this));
do {} while (!a.weakCompareAndSet(this, -4, 7));
assertEquals(7, a.get(this));
}
/**
* getAndSet returns previous value and sets to given value
*/
public void testGetAndSet() {
AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a;
a = updaterFor("x");
x = 1;
assertEquals(1, a.getAndSet(this, 0));
assertEquals(0, a.getAndSet(this, -10));
assertEquals(-10, a.getAndSet(this, 1));
}
/**
* getAndAdd returns previous value and adds given value
*/
public void testGetAndAdd() {
AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a;
a = updaterFor("x");
x = 1;
assertEquals(1, a.getAndAdd(this, 2));
assertEquals(3, a.get(this));
assertEquals(3, a.getAndAdd(this, -4));
assertEquals(-1, a.get(this));
}
/**
* getAndDecrement returns previous value and decrements
*/
public void testGetAndDecrement() {
AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a;
a = updaterFor("x");
x = 1;
assertEquals(1, a.getAndDecrement(this));
assertEquals(0, a.getAndDecrement(this));
assertEquals(-1, a.getAndDecrement(this));
}
/**
* getAndIncrement returns previous value and increments
*/
public void testGetAndIncrement() {
AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a;
a = updaterFor("x");
x = 1;
assertEquals(1, a.getAndIncrement(this));
assertEquals(2, a.get(this));
a.set(this, -2);
assertEquals(-2, a.getAndIncrement(this));
assertEquals(-1, a.getAndIncrement(this));
assertEquals(0, a.getAndIncrement(this));
assertEquals(1, a.get(this));
}
/**
* addAndGet adds given value to current, and returns current value
*/
public void testAddAndGet() {
AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a;
a = updaterFor("x");
x = 1;
assertEquals(3, a.addAndGet(this, 2));
assertEquals(3, a.get(this));
assertEquals(-1, a.addAndGet(this, -4));
assertEquals(-1, a.get(this));
}
/**
* decrementAndGet decrements and returns current value
*/
public void testDecrementAndGet() {
AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a;
a = updaterFor("x");
x = 1;
assertEquals(0, a.decrementAndGet(this));
assertEquals(-1, a.decrementAndGet(this));
assertEquals(-2, a.decrementAndGet(this));
assertEquals(-2, a.get(this));
}
/**
* incrementAndGet increments and returns current value
*/
public void testIncrementAndGet() {
AtomicIntegerFieldUpdater<AtomicIntegerFieldUpdaterTest> a;
a = updaterFor("x");
x = 1;
assertEquals(2, a.incrementAndGet(this));
assertEquals(2, a.get(this));
a.set(this, -2);
assertEquals(-1, a.incrementAndGet(this));
assertEquals(0, a.incrementAndGet(this));
assertEquals(1, a.incrementAndGet(this));
assertEquals(1, a.get(this));
}
/** compare raw numbers for atomic ops using JDK vs unsafe wrapper classes */
public void SW_testCompareAtomicOps() {
final AtomicIntegerFieldUpdater<ConcurrentMapOpsTest> intJDKCounter =
AtomicIntegerFieldUpdater.newUpdater(ConcurrentMapOpsTest.class,
"intJDKCounter");
final AtomicLongFieldUpdater<ConcurrentMapOpsTest> longJDKCounter =
AtomicLongFieldUpdater.newUpdater(ConcurrentMapOpsTest.class,
"longJDKCounter");
final AtomicReferenceFieldUpdater<ConcurrentMapOpsTest, LongRef>
refJDKCounter = AtomicReferenceFieldUpdater.newUpdater(
ConcurrentMapOpsTest.class, LongRef.class, "refJDKCounter");
final AtomicIntegerFieldUpdater<ConcurrentMapOpsTest> intUnsafeCounter =
AtomicUpdaterFactory.newIntegerFieldUpdater(ConcurrentMapOpsTest.class,
"intUnsafeCounter");
final AtomicLongFieldUpdater<ConcurrentMapOpsTest> longUnsafeCounter =
AtomicUpdaterFactory.newLongFieldUpdater(ConcurrentMapOpsTest.class,
"longUnsafeCounter");
final AtomicReferenceFieldUpdater<ConcurrentMapOpsTest, LongRef>
refUnsafeCounter = AtomicUpdaterFactory.newReferenceFieldUpdater(
ConcurrentMapOpsTest.class, LongRef.class, "refUnsafeCounter");
// some warmups
runAtomicOps(1, 50000, intJDKCounter, longJDKCounter, refJDKCounter,
intUnsafeCounter, longUnsafeCounter, refUnsafeCounter);
// timed runs with single threads to see the raw overheads with no
// concurrency (as we would expect in most usual cases)
runAtomicOps(1, 50000000, intJDKCounter, longJDKCounter, refJDKCounter,
intUnsafeCounter, longUnsafeCounter, refUnsafeCounter);
// now with concurrency
runAtomicOps(5, 2000000, intJDKCounter, longJDKCounter, refJDKCounter,
intUnsafeCounter, longUnsafeCounter, refUnsafeCounter);
}
/**
* Creates and returns an updater for objects with the given integer field.
*/
public static <T> AtomicIntegerFieldUpdater<T> newIntegerFieldUpdater(
Class<T> tclass, String fieldName) {
if (UnsafeHolder.hasUnsafe()) {
return new UnsafeAtomicIntegerFieldUpdater<T>(tclass, fieldName);
}
else {
return AtomicIntegerFieldUpdater.newUpdater(tclass, fieldName);
}
}