Details on thread safety (atomicity, visibility, order)

Definition: What is thread safety

A class is thread-safe if it behaves correctly when accessed by multiple threads, regardless of how the runtime environment is scheduled or how the processes are executed interchangeably, and does not require any additional synchronization or coordination in the calling code.

Two, three embodiment of thread safety

Third, thread safety: atomicity

3.1 atomicity — Atomic packages

3.1.1, AtomicInteger

AtomicInteger is an Integer class that provides atomic operations on addition and subtraction in a thread-safe manner

Here are the methods that AtomicIntege basically includes:

public final int getAndSet(int newValue)       // Set AtomicInteger to newValue and return oldValue
public final boolean compareAndSet(int expect, int update)    // If the input value equals the expected value, set and return true/false
public final int getAndIncrement(a)     // Increments the AtomicInteger atom by 1 and returns the value before the current increment
public final int getAndDecrement(a)   // Decrement the AtomicInteger atom by 1 and return the value before decrement
public final int getAndAdd(int delta)   // Add the delta value to the AtomicInteger and add back the previous value
public final int incrementAndGet(a)   // Increments the AtomicInteger atom by 1 and returns the value
public final int decrementAndGet(a)    // Subtracts the AtomicInteger atom by 1 and returns the value
public final int addAndGet(int delta)   // Adds the specified delta value to the AtomicInteger and returns the added value
Copy the code

Example:

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class AtomicIntegerExample {

    // Total requests
    public static int clientTotal = 5000;

    // The number of concurrent threads
    public static int threadTotal = 200;

    public static AtomicInteger count = new AtomicInteger(0);

    public static void main(String[] args) throws Exception {
    	// Get the thread pool
        ExecutorService executorService = Executors.newCachedThreadPool();
        // Define semaphore
        final Semaphore semaphore = new Semaphore(threadTotal);
        final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
        for (int i = 0; i < clientTotal ; i++) {
            executorService.execute(() -> {
                try {
                    semaphore.acquire();
                    add();
                    semaphore.release();
                } catch (Exception e) {
                    log.error("exception", e);
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        executorService.shutdown();
        log.info("count:{}", count.get());
    }

    private static void add(a) { count.incrementAndGet(); }}Copy the code

Here we use the total number of requests: 5000 and the number of concurrent threads executing at the same time: 200. We need to get a result of: 5000 for this result to be correct.

View the result:

13:43:26. [the main] INFO 473 com. Mmall. Concurrency. Example. The atomic. AtomicIntegerExample - count: 5000Copy the code

The result is 5000, which means thread-safe.

So what does AtomicInteger actually do for us in the underlying code? First we see AtomicInteger. IncrementAndGet () method

public class AtomicInteger extends Number implements java.io.Serializable{
/** * increments the AtomicInteger atom by 1 and returns the value *@returnUpdated value */
    public final int incrementAndGet(a) {
        return unsafe.getAndAddInt(this, valueOffset, 1) + 1; }}Copy the code

AtomicInteger calls Java’s underlying safe.getAndAddInt() method, which is key to implementing CAS.

IncrementAndGet () returns the incremented value, and getAndIncrement() returns the unincremented value, corresponding to the ++ I and I ++ operations, respectively. Likewise, decrementAndGet() and getAndDecrement() operate on — I and I –.

The Unsafe class is under the Sun.misc package and is not part of the Java standard. However, many of Java's basic libraries, including some of the most widely used high performance development libraries, were developed based on the Unsafe class, such as Netty, Cassandra, Hadoop, Kafka, etc. The Unsafe class plays an important role in making Java more efficient and enhancing the underlying operations of the Java language. The Unsafe class gave Java the ability to manipulate memory space in the same way that Pointers do in C, but it also introduced Pointers to the problem. Overusing the Unsafe class increases the likelihood of errors, so Java is not officially recommended and there is little official documentation. An Unsafe class is usually not to be used, either, unless you have a clear purpose and need to know more about it.Copy the code

Looking again at the safe.getAndAddInt() method

	/* * getIntVolatile and compareAndSwapInt are native methods * * getIntVolatile is the current expected value * * compareAndSwapInt is the CAS(compare and) If the value of the memory area has not changed, then the new value is directly assigned to the memory area */
    public final int getAndAddInt(Object var1, long var2, int var4) {
        int var5;
        do {
            var5 = this.getIntVolatile(var1, var2);
        } while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));
        return var5;
    }
    
    public native int getIntVolatile(Object var1, long var2);
    public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5);
       
Copy the code

We can see that getAndAddInt(Object var1, Long var2, int var4), the first argument passed in is the current Object, which is our: Count.incrementandget (), in getAndAddInt(), var1 is count, var2 is the current value, for example, in the current loop count is 2, var4 is incremented by 1 each time

Second, the two method calls involved in the getAndAddInt() method are defined as native, that is, methods that are native to Java’s underlying implementation

• GetIntVolatile () : Gets a reference to the main memory address that holds the current object count.
• CompareAndSwapInt () : compares the value of the current object with that of the underlying object. If the value is equal, the current object value is increased by 1. If the value is not equal, the underlying object value is obtained again.

We know that volatile is consistent, but AtomicInteger is not atomicity. AtomicInteger is consistent and atomicity. AtomicInteger’s internal implementation uses the volatile keyword, which is why the data retrieved from the underlying layer is the most recent when performing CAS (click here for CAS questions) :

If the current value is not the same as the latest value in memory, it indicates that it was modified by another thread during the process. The current value can only be updated to the latest value, and then the current value can be compared with the latest value in memory again until the two are equal, and the process of +1 is completed.Copy the code

The advantage of using AtomicInteger is that, unlike sychronized keyword or lock, atomicity is implemented in the form of a lock, which affects performance, but rather in the form of a circular comparison to improve performance.

3.1.2, AtomicLong

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public class AtomicLongExample {

    // Total requests
    public static int clientTotal = 5000;

    // The number of concurrent threads
    public static int threadTotal = 200;

    public static AtomicLong count = new AtomicLong(0);

    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        final Semaphore semaphore = new Semaphore(threadTotal);
        final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
        for (int i = 0; i < clientTotal ; i++) {
            executorService.execute(() -> {
                try {
                    semaphore.acquire();
                    add();
                    semaphore.release();
                } catch (Exception e) {
                    log.error("exception", e);
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        executorService.shutdown();
        log.info("count:{}", count.get());
    }

    private static void add(a) {
        count.incrementAndGet();
        // count.getAndIncrement();}}Copy the code

14:59:38. [the main] INFO 978 com. Mmall. Concurrency. Example. Atomic. AtomicLongExample - count: 5000Copy the code

3.1.3, AtomicBoolean

AtomicBoolean is located in Java. Util. Concurrent. Atomic package, is a Java provide can guarantee atomicity operation data of a class

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
public class AtomicBooleanExample {

    private static AtomicBoolean isHappened = new AtomicBoolean(false);

    // Total requests
    public static int clientTotal = 5000;

    // The number of concurrent threads
    public static int threadTotal = 200;

    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        final Semaphore semaphore = new Semaphore(threadTotal);
        final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
        for (int i = 0; i < clientTotal ; i++) {
            executorService.execute(() -> {
                try {
                    semaphore.acquire();
                    test();
                    semaphore.release();
                } catch (Exception e) {
                    log.error("exception", e);
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        executorService.shutdown();
        log.info("isHappened:{}", isHappened.get());
    }

    private static void test(a) {
        if (isHappened.compareAndSet(false.true)) {
            log.info("execute"); }}}Copy the code

Return result:

15:04:54. 954 / - thread pool - 1-2 INFO. Com mmall. Concurrency. Example. Atomic. AtomicBooleanExample - execute 15:04:54. 971 [main] INFO com.mmall.concurrency.example.atomic.AtomicBooleanExample - isHappened:trueCopy the code

Here we find log.info(“execute”); Is executed only once in the code, and isHappened:true is true.

This is because isHappend is set to true when the program compareAndSet() for the first time. Because of atomicity, there is no interference from other threads. By using AtomicBoolean, we make the code execute only once.

3.1.4, AtomicReference

No matter what execution timing or alternation method is used for operations between multiple threads, invariance conditions must not be destroyed. To maintain state consistency, related state variables need to be updated in a single atomic operation.Copy the code

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.atomic.AtomicReference;

@Slf4j
public class AtomicReferenceExample {

    private static AtomicReference<Integer> count = new AtomicReference<>(0);

    public static void main(String[] args) {
        count.compareAndSet(0.2); 
        count.compareAndSet(0.1);
        count.compareAndSet(1.3); 
        count.compareAndSet(2.4); 
        count.compareAndSet(3.5); 
        log.info("count:{}", count.get()); }}Copy the code

15:26:59. [the main] INFO 680 com. Mmall. Concurrency. Example. Atomic. AtomicReferenceExample - count: 4Copy the code

count.compareAndSet(0, 2); //count=0? Count.com pareAndSet(0, 1) count.compareAndSet(0, 1) //count=0? Count =2 count.compareAndSet(1, 3); //count=1? Count =2 count.compareAndSet(2, 4); //count=2? Count.com pareAndSet(3, 5); count.compareAndSet(3, 5); //count=3? If 5 is assigned, the test fails, and count=4Copy the code

3.1.5 ABA solution in CAS

CAS is not perfect and can cause ABA problems. For example, if the current memory was originally A, then changed to B and then to A by another thread, then when the current thread accesses A, it is assumed that it has not been accessed by any other thread. There is a risk of error in some scenarios. Like in a linked list. To solve this ABA problem, most optimistic locks are implemented by introducing a version number to mark the object, which changes every time the version number is changed, such as using a timestamp as the version number. The JDK provides an AtomicStampedReference class to address this problem, which maintains a stamp of type int that is updated each time data is updated.

3.2 atomicity — synchronized

3.3 atomicity — comparison

• Atomic: Performs better than Lock and can only synchronize one value
• Synchronized: can not interrupt the lock, suitable for not fierce competition, good readability
• Lock: interruptible Lock, diversified synchronization, can maintain the normal when the competition is fierce

Thread safety: visibility

4.1 Visibility — Syncronized

Two provisions of JMM on syncronization:

• The thread must flush the latest value of the shared variable to main memory before it can be unlocked
• When the thread locks, the value of the shared variable in the working memory will be cleared, so that when using the shared variable, it needs to read the latest value from the main memory (note: locking and unlocking are the same lock). Since atomicity and visibility are guaranteed by Syncronized, the variable can be used safely as long as it is modified by Syncronized

4.2 Visibility – Volatile

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;

@Slf4j
public class VolatileExample {

    // Total requests
    public static int clientTotal = 5000;

    // The number of concurrent threads
    public static int threadTotal = 200;

    public static volatile int count = 0;

    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        final Semaphore semaphore = new Semaphore(threadTotal);
        final CountDownLatch countDownLatch = new CountDownLatch(clientTotal);
        for (int i = 0; i < clientTotal ; i++) {
            executorService.execute(() -> {
                try {
                    semaphore.acquire();
                    add();
                    semaphore.release();
                } catch (Exception e) {
                    log.error("exception", e);
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
        executorService.shutdown();
        log.info("count:{}", count);
    }

    private static void add(a) { count++; }}Copy the code

16:12:01. [the main] INFO 404 com. Mmall. Concurrency. Example. Count. VolatileExample4 - count: 4986Copy the code

Five, thread safety: order

• In the JMM, the compiler and processor are allowed to reorder instructions, but the reordering process does not affect the execution of a single-threaded program, but does affect the correctness of multithreaded concurrent execution.
• Order is ensured through volatile, synchronized, and lock

5.1 happens-before principle

Thread safety: Summary

• atomicAtomic package, CAS algorithm, synchronized, Lock
  
  Atomicity is mutually exclusive, so only one thread can operate
• visibility: synchronized and volatile
  
  Changes made by a main memory thread can be observed by other threads in a timely manner
• order: happens-before principle
  
  Observe the happens-before principle. If two threads cannot be observed from the happens-before principle, their orderliness cannot be observed and the virtual machine can reorder them at will