[Java] Multi-thread processing --Exclusive control

Be careful of data sharing in multi-thread processing

• In multithread processing ** Be aware of whether data is shared between threads **
• If multiple threads process shared data at the same time, ** data inconsistency will occur! ** **

NG example

• Example of incrementing the value field prepared in the main thread in parallel by 100,000 threads (the result is expected to be 100,000)

//NG example
public class SynchronizedNotUse {
//Data shared by multiple threads
  private int value = 0;

//Execution of 100,000 threads
  public static void main(String[] args) {
    final int TASK_NUM = 100000;
    var th = new Thread[TASK_NUM];
    var tb = new SynchronizedNotUse();
    //Thread creation and execution
    for (var i = 0; i < TASK_NUM; i++) {
      th[i] = new Thread(() -> {
        tb.increment();
      });
      th[i].start();
    }
    //Wait until the thread ends
    for (var i = 0; i < TASK_NUM; i++) {
      try {
        th[i].join();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }

    System.out.println(tb.value); //Different every time
  }
  //Increment the value field
  void increment() {
    this.value++;
  }
}

• I always expect 100,000 results, but I get different values every time.
• The problem is ** the increment method of the SynchronizedNotUse class **
• In the increment method, add the value of the instance field with the ++ operator.
• At first glance, it seems that other threads cannot interrupt, but the ++ operator is internally
• ** Get the current location of the variable, add the value, reassign the operation result **
• If an interrupt from another thread occurs during processing, the result will not be reflected correctly!
• → ** Use the synchronized instruction **!

Exclusive control with synchronized block

• Processes enclosed in synchronized blocks ** will no longer be called by multiple threads at the same time **
• Even if it is almost at the same time, the one called later will be in the ** waiting state ** until the pre-processing (acquiring the lock) is completed.
• ** Lock **: Occupy a specific process
• ** Exclusive control (synchronous processing) **: Lock to prevent data inconsistency due to simultaneous execution
• Synchronous processing works correctly ** only when the locked objects are the same instance **
• Cannot sync between different lock objects
• Automatically unlocks when exiting the block

public class SynchronizedUse {
  private int value = 0;

  public static void main(String[] args) {
    final int TASK_NUM = 100000;
    var th = new Thread[TASK_NUM];
    var tb = new SynchronizedUse();
    for (var i = 0; i < TASK_NUM; i++) {
      th[i] = new Thread(() -> {
        tb.increment();
      });
      th[i].start();
    }
    for (var i = 0; i < TASK_NUM; i++) {
      try {
        th[i].join();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }

    System.out.println(tb.value);
  }
   //Specify the object to be locked as the current instance
  void increment() {
    synchronized(this) {
      this.value++;
    }
  }
}

synchronized modifier

• Used when you want to synchronize the entire method
• While one thread is executing synchronized method 1, another thread cannot execute synchronized method 2.
• Automatically unlocks when exiting method
• When the compiler generates cached code (** optimization **), synchronized ** is read from the original memory, so consistency is guaranteed **
• cf: The volatile modifier can also be used for variable assignment / retrieval only

synchronized void increment() {
    this.value++;
}

Explicit lock

• ** Explicit lock **: Explicit lock acquisition / release required
• Implicit lock: You don't have to be aware of lock acquisition / release (synchronized)
• ** ReentrantLock class **
• If you get an explicit lock, ** immediately after that, wrap it in a try block **
• Finally block guarantees unlocking

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class LockBasic {
  private int value = 0;
  private final Lock lock = new ReentrantLock();

  public static void main(String[] args) {
    final int TASK_NUM = 100000;
    var th = new Thread[TASK_NUM];
    var tb = new LockBasic();
    for (var i = 0; i < TASK_NUM; i++) {
      th[i] = new Thread(() -> {
        tb.increment();
      });
      th[i].start();
    }
    for (var i = 0; i < TASK_NUM; i++) {
      try {
        th[i].join();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }

    System.out.println(tb.value);
  }


  void increment() {
    //Rock acquisition
    lock.lock();
    try {
      this.value++;
    } finally {
      //unlock
      lock.unlock();
    }
  }
}

** tryLock method **

• ** You can check in advance if you can get a lock **
• Skip if you don't need to wait for lock

  if (lock.tryLock(10,TimeUnit.SECONDS)) {
    try {
    //Processing to be exclusively controlled
    } finally {
      lock.unlock();
    }
  }else{
  //What to do if you can't get a lock
}

Lock-free exclusive control

• ** Atomic assignment, addition and subtraction to a single variable at the hardware level **
• cf: synchronized and java.util.concurrent.locks are exclusive control assuming lock
• ** Atomic **: Guaranteed to have no interrupts
• In Java, reading and writing to variables other than ** long / double type ** is guaranteed by atomic
• Since long / double types are not necessarily atomic, there is a possibility of interrupts in a multithreaded environment (data corruption).
• You can implement the ++ operator atomically ** by using the ** getAndIncrement method of the AtomicInteger class!

import java.util.concurrent.atomic.AtomicInteger;

public class AtomicBasic {
  private AtomicInteger value = new AtomicInteger();

  public static void main(String[] args) {
    final int TASK_NUM = 100000;
    var th = new Thread[TASK_NUM];
    var tb = new AtomicBasic();
    for (var i = 0; i < TASK_NUM; i++) {
      th[i] = new Thread(() -> {
        tb.increment();
      });
      th[i].start();
    }
    for (var i = 0; i < TASK_NUM; i++) {
      try {
        th[i].join();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
    System.out.println(tb.value);
  }

  void increment() {
    value.getAndIncrement();
  }
}

Thread pool

• ** Prepare and pool threads used by the app in advance **
• Remove pooled threads as needed and return to pool when finished
• ** Save overhead on spawn / discard by reusing threads! ** **
• ** Executors class **: Thread pool generation
• ** execute method **: Execute the specified process via thread
• ** shutdown method **:
• Thread pool ** eliminates the need to be aware of thread creation! ** **

public class ThreadPool implements Runnable {
  @Override
  public void run() {
    for (var i = 0; i < 30; i++){
      System.out.println(Thread.currentThread().getName() + ":" + i);
    }
  }
}

import java.util.concurrent.Executors;

public class ThreadPoolBasic {

  public static void main(String[] args) {
    //Create a thread pool with 10 threads
    var es = Executors.newFixedThreadPool(10);
    //Call and execute with execute method
    es.execute(new ThreadPool());
    es.execute(new ThreadPool());
    es.execute(new ThreadPool());
    es.shutdown();
  }
}

Schedule execution

• ** ScheduledThreadService object **
• Generated by ** newScheduledThreadPool ** / ** newSingleScheduledExecutor ** method of Executors class
• ** schedule method **: Process is executed only once after the specified time has elapsed.
• ** scheduleAtFixedRate / scheduleWithFixedDelay method **: Process execution at specified time intervals

import java.time.LocalDateTime;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ThreadSchedule {
  public static void main(String[] args) {
  //Thread pool preparation
    var sche = Executors.newScheduledThreadPool(2);
  //Schedule execution registration
    sche.scheduleAtFixedRate(() -> {
      System.out.println(LocalDateTime.now());
    }, 0, 5, TimeUnit.SECONDS);

    //Pause the main thread after waiting for the schedule to execute
    try {
      Thread.sleep(10000);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }

    //Shut down the thread pool
    sche.shutdown();
  }
}

Receive the processing result of the thread

• ** ExecutorService ** / ** ScheduledExecutorService ** + ** Callable interface **
• ** Callable **: Assign T type to return type
• ** call method **: Code to execute on the Callable interface
• ** submit method **: execute the process of the defined Callable object
• Return value is ** Future type **: Asynchronous processing result
• ** get method **: Get the actual return value

//Find a random number in another thread and pause for a few milliseconds
//Display that value in the main thread
import java.util.Random;
import java.util.concurrent.Callable;

//Callable<Integer>Assign Integer type with
public class ThreadCallable implements Callable<Integer> {
  @Override
  //Code to execute in the Callable interface
  public Integer call() throws Exception {
    var rnd = new Random();
    var num = rnd.nextInt(1000);
    Thread.sleep(num);
    return num;
  }
}

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;

public class ThreadCallableBasic {

  public static void main(String[] args) {
  //Thread execution
    var exe = Executors.newSingleThreadExecutor();
    var r1 = exe.submit(new ThreadCallable());
    var r2 = exe.submit(new ThreadCallable());
    var r3 = exe.submit(new ThreadCallable());

  //Thread result display
    //The return value of the submit method is Future<Integer>
    try {
      System.out.println("r1: " + r1.get());
      System.out.println("r2: " + r2.get());
      System.out.println("r3: " + r3.get());
    } catch (InterruptedException | ExecutionException e) {
      e.printStackTrace();
    }
    exe.shutdown();
  }
}

Post-processing of thread processing

• ** CompletableFuture class **
• Concisely write post-processing after receiving asynchronous processing results
• Asynchronous processing is executed in series
• Asynchronous processing is executed in parallel
• Pass the asynchronous process to be executed to ** supplyAsync method **

//Find a random number in another thread and pause for a few milliseconds
//Show its value

import java.util.Random;
import java.util.concurrent.CompletableFuture;

public class FutureBasic {

  public static void main(String[] args) {
    //Asynchronous processing execution
    CompletableFuture.supplyAsync(() -> {
      var r = new Random();
      var num = r.nextInt(1000);
      heavy(num);
      return num;
    })
      //Processing after completion
      .thenAcceptAsync((result) -> {
        System.out.println(result);
      });
    System.out.println("...Any post-processing...");
    heavy(7000);
  }
  //Dummy processing (heavy) suspends processing only for the specified time
  public static void heavy(int num) {
    try {
      Thread.sleep(num);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}

Sort success / error handling

• ** whenCompleteAsync method **

import java.util.Random;
import java.util.concurrent.CompletableFuture;

public class FutureComplete {

  public static void main(String[] args) {
    CompletableFuture.supplyAsync(() -> {
      var r = new Random();
      var num = r.nextInt(1000);
      heavy(num);
      return num;
    })
      .whenCompleteAsync((result, ex) -> {
        //success
        if (ex == null) {
          System.out.println(result);
        } else {
        //Failure
          System.out.println(ex.getMessage());
        }
      });
    heavy(7000);
  }

  public static void heavy(int num) {
    try {
      Thread.sleep(num);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}

Execute multiple asynchronous processes in series

• ** Demonstrate the true value of Complete Future **
• ** supplyAsync **: Supplier (with return value without arguments)
• Result value generation
• ** thenApplyAsync **: Function (with arguments and return value)
• Value processing and conversion
• ** thenAcceptAsync **: Consumer (with arguments, no return value)
• Result receiving process

import java.util.Random;
import java.util.concurrent.CompletableFuture;

public class FutureSeq {

  public static void main(String[] args) {
  //Process 1(Random number generation)
    CompletableFuture.supplyAsync(() -> {
      var r = new Random();
      var num = r.nextInt(5000);
      heavy(2000);
      System.out.printf("Process 1: %d\n", num);
      return num;
    })
  //Process 2(Random number times)
    .thenApplyAsync((data) -> {
      var result = data * 2;
      heavy(2000);
      System.out.printf("Process 2: %d\n", result);
      return result;
    })
  //Process 3(Random number double)
    .thenAcceptAsync((data) -> {
      var num = data * 2;
      heavy(2000);
      System.out.printf("Process 3: %d\n", num);
    });
    heavy(7000);
  }

  public static void heavy(int num) {
    try {
      Thread.sleep(num);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }
}