Multithreaded atomic Operations | more challenging in August

CompareAndSwap comparison and swap

define

CAS has three operands, the memory value V, the old expected value A, and the new value B to be modified. The usual way to use CAS for synchronization is to read the value A from address V, perform A multi-step calculation to get the new value B, and then use CAS to change the value of V from A to B. If the value at V has not changed at the same time, the CAS operation succeeds.

Disadvantages: Although CAS effectively solves the problem of atomic operation, it still has three major problems.

1. Loop time is long and expensive. (High CPU overhead)
2. Atomic operations that guarantee only one shared variable. (Multiple use locks)
3. ABA problem. (Do not affect the use is not considered)

AtomicInteger i = new AtmoicInteger();
i.incrementAndGet();// Increment and return the result
 
// The incrementAndGet() method traces in and finds that the underlying method is called in the native method (unsafe.class)
//public final native boolean compareAndSwapInt(...) )
Lock CMPXCHG: lock CMPXCHG: lock northbridge
 
// Synchronized and volatile use the underlying lock CMPXCHG instruction
Copy the code

ABA problem

When thread 1 reads data 0 and does ++, it comes back to check the original value. During this process, other threads make changes to data 0, but the final result becomes 0 again, that is, the original value is A, the process is changed to B, and finally changed to A, which is the ABA problem

/ * * *@Author blackcat
 * @version: 1.0
 * @description: * / ABA problem
@Slf4j
public class ABATest {

    static AtomicReference<String> ref = new AtomicReference<>("A");

    public static void main(String[] args) throws InterruptedException {
        log.debug("main start...");
        // Get the value A
        // Has this shared variable been modified by its thread?
        String prev = ref.get();
        other();   //
        sleep(2);
        // Try to change to C
        log.debug("change A->C {}", ref.compareAndSet(prev, "C"));
    }


    private static void other(a) throws InterruptedException {
        new Thread(() -> {
            log.debug("change A->B {}", ref.compareAndSet(ref.get(), "B"));
        }, "t1").start();
        sleep(1);
        new Thread(() -> {
            log.debug("change B->A {}", ref.compareAndSet(ref.get(), "A"));
        }, "t2").start();
    }

    private static void sleep(int i) {
        try {
            Thread.sleep(i * 1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

 [main] DEBUG com.blact.cat.first.AtomicTest - main start...
 [t1] DEBUG com.blact.cat.first.AtomicTest - change A->B true
 [t2] DEBUG com.blact.cat.first.AtomicTest - change B->A true
 [main] DEBUG com.blact.cat.first.AtomicTest - change A->C true
Copy the code

** Data 0 plus the version number, each thread read data 0 with the version number, each time after processing back with the version number can be compared. AtomicStampedReference Available version

/ * * *@Author blackcat
 * @create2021/8/2 that *@version: 1.0
 * @description: * /
@Slf4j
public class AtomicStampedReferenceTest {

    static AtomicStampedReference<String> ref = new AtomicStampedReference<>("A".0);

    public static void main(String[] args) throws InterruptedException {
        log.debug("main start...");
        // Get the value A
        String prev = ref.getReference();  
        // Get the version number
        int stamp = ref.getStamp();
        log.debug("Version {}", stamp);
        // If there is interference from other threads, ABA occurs
        other();
        sleep(2);
        // Try to change to C
        log.debug("change A->C {}", ref.compareAndSet(prev, "C", stamp, stamp + 1));
    }

    private static void other(a) throws InterruptedException {
        new Thread(() -> {
            log.debug("change A->B {}", ref.compareAndSet(ref.getReference(), "B",
                    ref.getStamp(), ref.getStamp() + 1));
            log.debug("Update to {}", ref.getStamp());
        }, "t1").start();
        sleep(1);
        new Thread(() -> {
            log.debug("change B->A {}", ref.compareAndSet(ref.getReference(), "A",
                    ref.getStamp(), ref.getStamp() + 1));
            log.debug("Update to {}", ref.getStamp());
        }, "t2").start();
    }

    private static void sleep(int i) {
        try {
            Thread.sleep(i * 1000);
        } catch(InterruptedException e) { e.printStackTrace(); } } } [main] DEBUG com.blact.cat.first.AtomicStampedReferenceTest - main start... [the main] the DEBUG com. Blact. The first. AtomicStampedReferenceTest - version0
 [t1] DEBUG com.blact.cat.first.AtomicStampedReferenceTest - change A->B true(t1) DEBUG com. Blact. The first. AtomicStampedReferenceTest - for updated versions1
 [t2] DEBUG com.blact.cat.first.AtomicStampedReferenceTest - change B->A true(t2) DEBUG com. Blact. The first. AtomicStampedReferenceTest - for updated versions2
 [main] DEBUG com.blact.cat.first.AtomicStampedReferenceTest - change A->C false
Copy the code

atomicupdateBasic types of

Updating base types atomically, the Atomic package provides the following three classes.

AtomicBoolean: Atomic update Boolean type.

AtomicInteger: Atomic update integer.

AtomicLong: AtomicLong.

/ * * *@Author blackcat
 * @create 2021/8/2 14:42
 * @version: 1.0
 * @description:AtomicInteger Related method */
@Slf4j
public class AtomicIntegerTest {

    public static void main(String[] args) {
        AtomicInteger i = new AtomicInteger(0);
        // Get and increment (I = 0, result I = 1, return 0), similar to I ++
        log.info("getAndIncrement:{}", i.getAndIncrement());
        // Increment and get (I = 1, result I = 2, return 2), similar to ++ I
        log.info("incrementAndGet:{}", i.incrementAndGet());
        // Decrement and get (I = 2, result I = 1, return 1), similar to -- I
        log.info("decrementAndGet:{}", i.decrementAndGet());
        // Get and decrement (I = 1, result I = 0, return 1), similar to I --
        log.info("getAndDecrement:{}", i.getAndDecrement());
        // Get and add (I = 0, result I = 2, return 0)
        log.info("getAndAdd:{}", i.getAndAdd(2));
        // Add value and get (I = 2, result I = 0, return 0)
        log.info("addAndGet:{}", i.addAndGet(-2));
        // Get and update (I = 0, p = current value of I, result I = -2, return 0)
        // The operation in the function is atomic, but the function needs to have no side effects
        log.info("getAndUpdate:{}", i.getAndUpdate(p -> p - 2));
        // update and get (I = -2, p = current value of I, result I = 0, return 0)
        // The operation in the function is atomic, but the function needs to have no side effects
        log.info("updateAndGet:{}", i.updateAndGet(p -> p + 2));
        // Get and calculate (I = 0, p is the current value of I, x is parameter 1, result I = 10, return 0)
        // The operation in the function is atomic, but the function needs to have no side effects
        // getAndUpdate If an external local variable is referenced ina lambda, ensure that the local variable is final
        GetAndAccumulate can refer to the external local variable as argument 1, but because it is not in the lambda it does not have to be final
        log.info("getAndAccumulate:{}", i.getAndAccumulate(10, (p, x) -> p + x));
        // Calculate and get (I = 10, p is the current value of I, x is parameter 1, result I = 0, return 0)
        // The operation in the function is atomic, but the function needs to have no side effects
        log.info("accumulateAndGet:{}", i.accumulateAndGet(-10, (p, x) -> p + x)); }}Copy the code

atomicupdateAn array type

To update an element of an array atomically, the Atomic package provides the following classes.

AtomicIntegerArray: AtomicIntegerArray updates the elements of an integer array.

AtomicLongArray: AtomicLongArray: AtomicLongArray: AtomicLongArray: AtomicLongArray.

AtomicReferenceArray: AtomicReferenceArray: AtomicReferenceArray updates elements in an array of reference types.

/ * * *@Author blackcat
 * @create2021/8/2 therefore *@version: 1.0
 * @description: * /
@Slf4j
public class AtomicIntegerArrayTest {

    static int[] value = new int[] {1.2};

    static AtomicIntegerArray ai = new AtomicIntegerArray(value);

    public static void main(String[] args) {
        // Atomically sets the element in the array position I to newValue, which is 3
        ai.getAndSet(0.3);
        log.info("AtomicIntegerArray:{}", ai.get(0));
        // If the current value is equal to the expected value, then atomically set the element in the array position I to the update value, resetting it to 1
        boolean compareAndSet = ai.compareAndSet(0.3.1);
        log.info("{} -- -- -- -- -- -- -- {}", compareAndSet, ai.get(0));
    //int addAndGet (int I, int delta) : atomically adds the input value to the elements of the array indexed by I. Return to 2
        ai.getAndAdd(0.1);
        boolean addCompare = ai.compareAndSet(0.3.1);
        log.info("add after------{}-------{}", addCompare, ai.get(0));
        log.info("value; {}", value[0]); }}Copy the code

Note that the value of the array is passed in by the constructor, and the AtomicIntegerArray makes a copy of the current array, so that when AtomicIntegerArray makes changes to its inner array elements, it does not affect the array passed in.

Atomic update reference types

AtomicReference: Indicates the type of an atomic update reference.

AtomicReferenceFieldUpdater: atomic updates a reference type in the field.

AtomicMarkableReference: atomic updates with mark a reference type.

* * *@Author blackcat
 * @create 2021/8/2 21: 08 *@version: 1.0
 * @description: * /@Slf4j
public class AtomicReferenceTest {

    public static AtomicReference<User> atomicUserRef = new AtomicReference<User>();

    public static void main(String[] args) {
        User user = new User("Black".15);
        atomicUserRef.set(user);
        log.info("before:{}",atomicUserRef.get().toString());
        User updateUser = new User("She ling".17);
        atomicUserRef.compareAndSet(user, updateUser);
        log.info("after:{}",atomicUserRef.get().toString());
    }

    static class User {
        private String name;
        private int age;

        public User(String name, int age) {
            this.name = name;
            this.age = age;
        }

        public String getName(a) {
            return name;
        }

        public int getAge(a) {
            return age;
        }

        @Override
        public String toString(a) {
            return "User{" +
                    "name='" + name + '\' ' +
                    ", age=" + age +
                    '} '; }}}Copy the code

Atomic field updater

AtomicIntegerFieldUpdater: atomic updates integer field updater.

AtomicLongFieldUpdater: AtomicLongFieldUpdater is an updater for atomically updating long integer fields.

AtomicStampedReference: AtomicStampedReference updates a reference type with a version number. This class associates an integer value with a reference

The update data available for atoms and the version number of the data can be addressed when atomic updates are made using CAS

ABA problem.

Field updaters allow atomic operations to be performed on a Field of an object. They can only be used with volatile fields. Otherwise it will appear unusual Exception in the thread “is the main” Java. Lang. IllegalArgumentException: Must be volatile type

/ * * *@Author blackcat
 * @createAct 2021/8/2 *@version: 1.0
 * @description: * /
@Slf4j
public class AtomicIntegerFieldUpdaterTest {

    private volatile int field;

    public static void main(String[] args) {
        AtomicIntegerFieldUpdater fieldUpdater = AtomicIntegerFieldUpdater.newUpdater(AtomicIntegerFieldUpdaterTest.class, "field");
        AtomicIntegerFieldUpdaterTest old = new AtomicIntegerFieldUpdaterTest();
        fieldUpdater.compareAndSet(old, 0.10);
        // Modify field = 10
        System.out.println(old.field);
        // Field = 20
        fieldUpdater.compareAndSet(old, 10.20);
        System.out.println(old.field);
        // Failed to modify field = 20
        fieldUpdater.compareAndSet(old, 10.30); System.out.println(old.field); }}Copy the code

Atomic accumulator

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.Consumer;
import java.util.function.Supplier;

/ * * *@Author blackcat
 * @create 2021/8/2 22:07
 * @version: 1.0
 * @description: Accumulator comparison */
public class AccCompare {

    private static <T> void demo(Supplier<T> adderSupplier, Consumer<T> action) {
        T adder = adderSupplier.get();
        long start = System.nanoTime();
        List<Thread> ts = new ArrayList<>(10);
        // Four threads add up to 500,000 each
        for (int i = 0; i < 4; i++) {
            ts.add(new Thread(() -> {
                for (int j = 0; j < 500000; j++) { action.accept(adder); }})); } ts.forEach(t -> t.start()); ts.forEach(t -> {try {
                t.join();
            } catch(InterruptedException e) { e.printStackTrace(); }});long end = System.nanoTime();
        System.out.println(adder + " cost:" + (end - start) / 1000 _000);
    }

    public static void main(String[] args) {
        for (int i = 0; i < 5; i++) {
            demo(() -> new LongAdder(), adder -> adder.increment());
        }
        for (int i = 0; i < 5; i++) {
            demo(() -> newAtomicLong(), adder -> adder.getAndIncrement()); }}}20000000 cost:59
20000000 cost:46
20000000 cost:51
20000000 cost:60
20000000 cost:45
20000000 cost:340
20000000 cost:563
20000000 cost:519
20000000 cost:527
20000000 cost:523
Copy the code

AtomicLong is an atomic operation performed by multiple threads on a single hot spot value. In LongAdder, each thread has its own slot, and each thread generally only performs CAS operations on the value in its slot.

For example, there are three threads a, ThreadB, and ThreadC, and each thread increments the value by 10.

For AtomicLong, the final result is always computed in the form value= 10+10+10=30

But for LongAdder, there’s a base variable inside, a Cell[] array. Base variable: in non-race condition, directly accumulates to this variable Cell[] array: in race condition, accumulates to each thread’s own slot Cell[I]

value = base + cell[0]+…. +cell[i]

Unsafe

The Unsafe object provides very low-level methods that manipulate memory and threads. The Unsafe object cannot be called directly, but can only be obtained through reflection.

/ * * *@Author blackcat
 * @create2021/8/2 * breasts@version: 1.0
 * @description: Retrieves the Unsafe object */
public class UnsafeAccessor {

    static Unsafe unsafe;

    static {
        try {
            Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe");
            theUnsafe.setAccessible(true);
            unsafe = (Unsafe) theUnsafe.get(null);
        } catch (NoSuchFieldException | IllegalAccessException e) {
            throw newError(e); }}static Unsafe getUnsafe(a) {
        returnunsafe; }}Copy the code

/ * * *@Author blackcat
 * @create2021/8/2 with *@version: 1.0
 * @descriptionAtomicData */
@Slf4j
public class AtomicData {

    private volatile int data;
    static final Unsafe unsafe;
    static final long DATA_OFFSET;
    static {
        unsafe = UnsafeAccessor.getUnsafe();
        try {
            // The offset of the data attribute in the DataContainer object, which is used for the Unsafe to access the property directly
            DATA_OFFSET = unsafe.objectFieldOffset(AtomicData.class.getDeclaredField("data"));
        } catch (NoSuchFieldException e) {
            throw newError(e); }}public AtomicData(int data) {
        this.data = data;
    }

    public void decrease(int amount) {
        int oldValue;
        while(true) {
            // Get the old value of the shared variable. You can add a breakpoint to this line and modify the data debugging to further understand
            oldValue = data;
            // Cas tries to change data to the old value -amount, and returns false if the old value was changed by another thread
            if (unsafe.compareAndSwapInt(this, DATA_OFFSET, oldValue, oldValue - amount)) {
                return; }}}public void increment(int amount) {
        int oldValue;
        while(true) {
            oldValue = data;
            if (unsafe.compareAndSwapInt(this, DATA_OFFSET, oldValue, oldValue + amount)) {
                return; }}}public int getData(a) {
        return data;
    }


    public static void main(String[] args) throws InterruptedException {
        AtomicData atomicData = new AtomicData(100);
        Thread t1 = new Thread(() -> {
            for (int i = 0; i < 5000; i++) {
                atomicData.decrease(1); }},"add");
        Thread t2 = new Thread(() -> {
            for (int i = 0; i < 5000; i++) {
                atomicData.increment(1); }},"sub");
        t1.start();
        t2.start();
        t1.join();
        t2.join();
        log.info("result:{}", atomicData.getData()); }}Copy the code