Introduction to Multithreading
Program, Process, and Thread
A program is a static set of instructions that stores on a disk. A process is an instance of a program that is currently being executed by the operating system. A thread is a small unit of process that can be scheduled and executed by the operating system. In simpler terms, a program is the code, a process is the excution of the code, a thread is a unit of execution within a process. Their relationships are illustrated in the following image:
Thread Life Cycle Within Concurrency
A thread goes through various states in its life cycle, from creation to its termination. The life cycle of a thread typically includes five states:
New,Runnable,Running,Blocked/Waiting, andTerminated.
New: This refers to the state where a thread instance is created but has not yet started running. In Java, this is commonly achieved by usingnew Thread().Runnable: This refers to the state where a thread instance is ready to run but is waiting for CPU time. In Java, this is commonly achieved by ivokingstart()on the thread instance.Running: This is refers to the state where the thread instance is actively executing its task.Blocked/Waiting: This refers to the state where a thread instance is waiting for other threads to complete their tasks, or it is waiting for a particular condition to be met.Terminated: This refers to the state where the thread instance has completed its task and has exited.
Fork/Join Model
The Fork/Join Model is an effective strategy commonly used in scenarios where a task can be recursively divided into multiple smaller, independent subtasks. This strategy typically involves the following steps:
-
The process begins with a single master thread. When it encounters a task that can be broken down into smaller subtasks, the master thread forks (spawns) multiple subsidiary threads, each of which takes on a portion of the task.
-
Once the subsidiary threads have completed their respective tasks, they join back with the master thread. Joining involves terminating the subsidiary threads and consolidating their results into the master thread. Afterward, the master thread continues its execution.
SHOW CODE
import java.util.Arrays;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;
public class Solution {
// RecursiveTask to compute the sum of a part of an array
static class SumTask extends RecursiveTask {
private final int[] array;
private final int start;
private final int end;
// Threshold for splitting the task into smaller subtasks
private static final int THRESHOLD = 1000;
public SumTask(int[] array, int start, int end) {
this.array = array;
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
// If the task is small enough, calculate the sum directly
if (end - start <= THRESHOLD) {
long sum = 0;
for (int i = start; i < end; i++) {
sum += array[i];
}
return sum;
}
// Otherwise, split the task into two subtasks
int middle = (start + end) / 2;
SumTask leftTask = new SumTask(array, start, middle);
SumTask rightTask = new SumTask(array, middle, end);
// Fork the subtasks
leftTask.fork();
rightTask.fork();
// Join the results of the subtasks
long leftResult = leftTask.join();
long rightResult = rightTask.join();
// Combine the results and return
return leftResult + rightResult;
}
}
public static void main(String[] args) {
int[] array = new int[10000];
// Initialize all elements to 1
Arrays.fill(array, 1);
// Create a ForkJoinPool to execute the tasks
ForkJoinPool pool = new ForkJoinPool();
// Submit the task to the ForkJoinPool
SumTask task = new SumTask(array, 0, array.length);
long result = pool.invoke(task);
System.out.println("Sum: " + result); // Output the result
}
}
SHOW OUTPUT
Sum: 10000
Critical Section & Race Condition
Critical Section refers to a portion of a program where shared resources are accessed and modified. A Race Condition occurs when the program’s behavior depends on the non-deterministic execution order of threads. Since multiple threads can access a critical section concurrently, it is essential to synchronize their access to avoid data inconsistency.
For example, suppose there are four threads trying to read a value of 20 from memory. Each thread increments the value by 1 and writes it back to memory. Without synchronization mechanisms, after all threads complete their writes, the final value might be 21 instead of the expected 24. This leads to an unexpected outcome, as illustrated below.
To resolve this issue, access to the critical section must be exclusive. This is commonly achieved by using mutexes, read/write locks, semaphores, conditional variables or barriers to ensure that only one thread can modify the shared resource at a time.
Synchronization Mechanisms
Synchronization Mechanisms are rules used to coordinate the execution of threads in a program, ensuring safe access to shared resources. These mechanisms can be primarily classified into five types: Mutexes, Read/Write Locks, Semaphores, Conditional Variables, and Barriers.
Mutex Mechanism
The mutex mechanism ensures that only one thread can access the critical section at a time. Other threads will be blocked until the mutex is released. The key logic behind the mutex mechanism is as follows:
Lock lock = new ReentrantLock();
lock.lock(); // Acquire the lock
try {
// Critical section code
} finally {
lock.unlock(); // Release the lock
}
SHOW CODE
public class Mutex {
private static int counter = 0;
private static final Object lock = new Object();
public static void runExperiment(String experimentName, Runnable task) {
counter = 0;
Thread t1 = new Thread(task);
Thread t2 = new Thread(task);
t1.start();
t2.start();
try {
t1.join();
t2.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Final counter value " + experimentName + ": " + counter + "");
}
public static void incrementCounterWithMutex() {
for (int i = 0; i < 100; i++) {
synchronized (lock) {
int temp = counter;
try {
Thread.sleep(1); // Sleep for 1 millisecond
} catch (InterruptedException e) {
e.printStackTrace();
}
counter = temp + 1;
}
}
}
public static void incrementCounterNoMutex() {
for (int i = 0; i < 100; i++) {
int temp = counter;
try {
Thread.sleep(1); // Sleep for 1 millisecond
} catch (InterruptedException e) {
e.printStackTrace();
}
counter = temp + 1;
}
}
public static void main(String[] args) {
runExperiment("With Mutex Experiment", Mutex::incrementCounterWithMutex);
runExperiment("No Mutex Experiment", Mutex::incrementCounterNoMutex);
}
}
SHOW OUTPUT
Final counter value With Mutex Experiment: 200
Final counter value No Mutex Experiment: 100
Read/Write Lock Mechanism
The Read/Write Lock Mechanism allows threads read the shared resources concurrently, but ensures exclusive access for write operations. This mechanism is useful in scenarios where there are many read operations and fewer write operations, such as caching systems or databases. The key logic behind the read/write lock mechanism is as follows:
ReadWriteLock rwLock = new ReentrantReadWriteLock();
rwLock.readLock().lock(); // Acquire read lock
try {
// Read operation
} finally {
rwLock.readLock().unlock(); // Release read lock
}
rwLock.writeLock().lock(); // Acquire write lock
try {
// Write operation
} finally {
rwLock.writeLock().unlock(); // Release write lock
}
Under the Read/Write Lock mechanism, multiple threads can access the critical section simultaneously without blocking each other. When a writer thread is accessing the critical section, other writer threads and reader threads are blocked. Similarly, when a reader thread is accessing the critical section, the writer thread is blocked.
SHOW CODE
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.locks.ReentrantReadWriteLock;
public class ReadWriteLock {
private static volatile int counter = 0;
private static final int TARGET_VALUE = 1000;
private static final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
public static int incrementValue() {
lock.writeLock().lock();
try {
Thread.sleep(1);
if (counter < TARGET_VALUE) {
counter++;
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.writeLock().unlock();
}
return counter;
}
public static int readValue() {
lock.readLock().lock();
try {
Thread.sleep(1);
return counter;
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.readLock().unlock();
}
return 0;
}
public static void main(String[] args) {
long start = System.currentTimeMillis();
List readers = new ArrayList<>();
for (int i = 0; i < 8; i++) {
readers.add(new Thread(() -> {
while (readValue() < TARGET_VALUE) {
try {
Thread.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}));
}
List writers = new ArrayList<>();
for (int i = 0; i < 2; i++) {
writers.add(new Thread(() -> {
while (incrementValue() < TARGET_VALUE) {
try {
Thread.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}));
}
readers.forEach(Thread::start);
writers.forEach(Thread::start);
readers.forEach(t -> {
try {
t.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
});
writers.forEach(t -> {
try {
t.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
});
long end = System.currentTimeMillis();
System.out.println("Time taken: " + (end - start) / 1000.0 + " seconds");
}
}
SHOW OUTPUT
Time taken: 1.787 seconds
In the above code, the
readValuemethod uses a read lock, allowing multiple reader threads to access the counter (critical section) simultaneously without blocking each other. If the method used a write lock instead of a read lock, the execution time would be higher. This is because when a writer thread is accessing the counter, all other threads would be blocked.
public static int readValue() {
// lock.readLock().lock();
lock.writeLock().lock();
try {
Thread.sleep(1);
return counter;
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
// lock.readLock().unlock();
lock.writeLock().unlock();
}
return 0;
}
Time taken: 6.793 seconds # Use write lock
Semaphore Mechanism
The Semaphore mechanism uses an integer to manage a set of permits, allowing multiple threads to access shared resources simultaneously, but only up to a predetermined limit. There are two types of semaphores: Binary Semaphore and Counting Semaphore.
- Binary Semaphore: Similar to a mutex, it operates with two states (
0and1), ensuring mutual exclusion in critical sections. - Counting Semaphore: Allows a count greater than 1, permitting multiple threads to access shared resources concurrently, as long as the permit limit is not exceeded.
Semaphores are useful in controlling access to a pool of resources, such as limiting the number of threads that can access a database connection or a server. The key logic behind the semaphore mechanism is as follows:
Semaphore semaphore = new Semaphore(3); // Allow 3 threads at once
semaphore.acquire(); // Acquire a permit
try {
// Critical section code
} finally {
semaphore.release(); // Release a permit
}
SHOW CODE
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
public class SemaphoreTest {
// Global shared resource
// AtomicInteger allows multiple threads to read/write value of counter without requiring synchronization
private static final AtomicInteger counter = new AtomicInteger(0);
// Semaphore with a count of 5
private static final Semaphore semaphore = new Semaphore(5);
private static final int TARGET_VALUE = 5000;
public static void main(String[] args) {
long startTime = System.currentTimeMillis();
Thread[] workers = new Thread[10];
for (int i = 0; i < workers.length; i++) {
workers[i] = new Thread(SemaphoreTest::worker);
workers[i].start();
}
for (Thread worker : workers) {
try {
worker.join();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
System.out.println("Thread was interrupted");
}
}
long endTime = System.currentTimeMillis();
System.out.println("Time taken: " + (endTime - startTime) / 1000.0 + " seconds");
}
private static void worker() {
while (true) {
try {
semaphore.acquire(); // Acquire the semaphore
if (counter.get() >= TARGET_VALUE) {
break;
}
counter.incrementAndGet(); // Atomically increments the counter
Thread.sleep(1); // Simulate work
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
System.out.println("Thread was interrupted");
} finally {
semaphore.release(); // Release the semaphore
}
}
}
}
Conditional Variable Mechanism
The Conditional Variable Mechanism allows a thread to suspend its execution until other threads signal that a certain condition has been met. It is often used in conjunction with mutexes to implement synchronization patterns, such as the producer-consumer pattern.
In this mechanism, a thread can wait on a condition variable until it is notified by other threads that the condition has been satisfied. When the condition is fulfilled, one or more threads can be signaled to wake up and proceed. Condition variables are commonly used in scenarios like the producer-consumer problem, where one thread produces data and another consumes it, with both threads needing to wait for each other under specific conditions.
The key logic behind the Conditional Variable Mechanism is as follows:
synchronized (lock) {
while (!condition) {
lock.wait(); // Wait until condition is true
}
// Proceed with critical section
}
synchronized (lock) {
condition = true;
lock.notify(); // Notify waiting threads
}
SHOW CODE
public class ConditionalVariable {
private static final Object mutex = new Object();
private static int sharedNumber;
private static boolean ready = false;
public static void producer() {
synchronized (mutex) {
sharedNumber = 9; // Producing a number
ready = true;
System.out.println("Producer has produced the number: " + sharedNumber);
mutex.notify(); // Notify the consumer
}
}
public static void consumer() {
synchronized (mutex) {
while (!ready) {
try {
mutex.wait(); // Wait until the number is ready
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
System.out.println("Consumer thread was interrupted.");
}
}
System.out.println("Consumer has consumed the number: " + sharedNumber);
}
}
public static void main(String[] args) {
Thread producerThread = new Thread(ConditionalVariable::producer);
Thread consumerThread = new Thread(ConditionalVariable::consumer);
producerThread.start();
consumerThread.start();
try {
producerThread.join();
consumerThread.join();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
System.out.println("Main thread was interrupted.");
}
}
}
public class Solution {
private static final Object mutex = new Object();
private static int sharedNumber;
private static boolean ready = false;
private static void producer() {
synchronized (mutex) {
sharedNumber = 9; // Producing a number
ready = true;
System.out.println("Producer has produced the number: " + sharedNumber);
}
}
private static void consumer() {
// Busy waiting loop
while (true) {
synchronized (mutex) {
if (ready) {
System.out.println("Consumer has consumed the number: " + sharedNumber);
break;
}
}
try {
Thread.sleep(1); // Sleep for a short time
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
System.out.println("Thread was interrupted");
}
}
}
public static void main(String[] args) {
Thread producerThread = new Thread(Solution::producer);
Thread consumerThread = new Thread(Solution::consumer);
producerThread.start();
consumerThread.start();
try {
producerThread.join();
consumerThread.join();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
System.out.println("Main thread was interrupted");
}
}
}
Barrier Mechanism
The Barrier Mechanism allows threads to wait for each other at a specified point in their execution. It ensures that threads cannot proceed to the subsequent code until all threads have reached the barrier, as illustrated in the following image.
Barrier Mechanism are useful in parallel algorithms where multiple threads need to work in phases, and the next phase cannot begin until all threads complete the current one.
The key logic behind this mechanism is as follows:
CyclicBarrier barrier = new CyclicBarrier(4, new Runnable() {
public void run() {
System.out.println("All threads reached the barrier!");
}
});
// Threads:
barrier.await(); // Wait at the barrier
SHOW CODE
import java.util.concurrent.CyclicBarrier;
class Barriers {
private static final CyclicBarrier barrier = new CyclicBarrier(2, () ->
System.out.println("All threads have reached the barrier. Continue execution."));
public static void main(String[] args) {
Thread t1 = new Thread(Barriers::work);
Thread t2 = new Thread(Barriers::work);
// Start both threads.
t1.start();
t2.start();
}
private static void work() {
System.out.println("Thread " + Thread.currentThread().getName()
+ " is waiting at the barrier");
try {
// Wait for the specified number of threads (2 in this case) to reach the barrier.
barrier.await();
System.out.println("Thread " + Thread.currentThread().getName() + " is released");
} catch (Exception e) {
// Handle exceptions, if any.
}
}
}
SHOW OUTPUT
Thread Thread-0 is waiting at the barrier
Thread Thread-1 is waiting at the barrier
All threads have reached the barrier. Continue execution.
Thread Thread-1 is released
Thread Thread-0 is released
Thread Creation In Java
Extending Thread Class
SHOW CODE
class MyThread extends Thread {
@Override
public void run() {
System.out.println(this.getName() + " is running.");
}
}
public class Main {
public static void main(String[] args) {
MyThread myThread = new MyThread();
myThread.setName("MYTHREAD-0");
myThread.start();
}
}
Implementing Runnable Interface
SHOW CODE
public class Main {
public static void main(String[] args) {
Thread thread = new Thread(() ->
System.out.println(Thread.currentThread().getName() + " is running."));
thread.start();
}
}
//class MyRunnable implements Runnable {
// @Override
// public void run() {
// System.out.println("Thread is running");
// }
//}
//
//public class Main {
// public static void main(String[] args) {
// Thread thread = new Thread(new MyRunnable());
// thread.start(); // Starts a new thread
// }
//}
Using the Callable Interface
SHOW CODE
import java.util.concurrent.*;
public class Main {
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newSingleThreadExecutor();
Future<String> future = executor.submit(() ->
Thread.currentThread().getName() + " is running.");
System.out.println(future.get());
executor.shutdown();
}
}
class MyCallable implements Callable<String> {
// @Override
// public String call() {
// return "Thread executed";
// }
//}
//
//public class Main {
// public static void main(String[] args) throws Exception {
// ExecutorService executor = Executors.newSingleThreadExecutor();
// Future<String> future = executor.submit(new MyCallable());
// System.out.println(future.get()); // Retrieves the result
// executor.shutdown();
// }
//}
Using Executors
SHOW CODE
import java.util.concurrent.*;
public class Main {
public static void main(String[] args) {
ExecutorService executor = Executors.newFixedThreadPool(2);
Runnable task1 = () -> System.out.println(Thread.currentThread().getName() + " executed");
Runnable task2 = () -> System.out.println(Thread.currentThread().getName() + " executed");
executor.execute(task1);
executor.execute(task2);
executor.shutdown();
}
}
Thread Termination In Java
Using a flag
SHOW CODE
class Task implements Runnable {
// Use volatile to ensure visibility between thread
private volatile boolean running = true;
@Override
public void run() {
while (running) {
System.out.println("Thread is running.");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
System.out.println("Thread is stopping.");
}
public void stop() {
running = false;
}
}
public class Main {
public static void main(String[] args) throws InterruptedException {
Task task = new Task();
Thread thread = new Thread(task);
thread.start();
Thread.sleep(3000); // Allow thread to run for a while
task.stop(); // Signal thread to stop
}
}
Interrupting a Thread
SHOW CODE
public class Main {
public static void main(String[] args) throws InterruptedException {
Thread thread = new Thread(() -> {
try {
while (!Thread.currentThread().isInterrupted()) {
System.out.println("Thread is running");
Thread.sleep(1000);
}
} catch (InterruptedException e) {
System.out.println("Thread is interrupted");
}
});
thread.start();
Thread.sleep(3000); // Allow thread to run for a while
thread.interrupt(); // Interrupt the thread
}
}
Daemon Threads
A daemon thread in Java is a low-priority thread that runs in the background, typically handling tasks such as garbage collection or other housekeeping operations. The JVM terminates when all non-daemon threads have completed execution, even if daemon threads are still active. To create a daemon thread, the
setDaemon(true)method is used on aThreadobject before it is started.
SHOW CODE
public class Main {
public static void main(String[] args) {
Thread daemonThread = new Thread(() -> {
while (true) {
System.out.println("Daemon thread is running");
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
});
daemonThread.setDaemon(true); // Mark as daemon
daemonThread.start();
System.out.println("Main thread is ending");
}
}
SHOW OUTPUT
Main thread is ending
Daemon thread is running
Performance Optimization
When it comes to multithreading, performance optimization typically focuses on two key areas: latency reduction and throughput improvement.
Latency reduction refers to minimizing the time it takes to complete a single task, measured in
time units. This is often achieved by breaking a task into smaller subtasks that can be executed concurrently by multiple threads.
Throughput, on the other hand, measures the number of tasks completed within a specific time frame, typically expressed as
tasks / time unit. Throughput improvement is commonly achieved by utilizingthread pooling, which reduces the overhead of creating and destroying threads for each task.
SHOW CODE: Latency Reduction
import javax.imageio.ImageIO;
import java.awt.image.BufferedImage;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
public class Main {
private static final String SOURCE_FILE = "src/main/resources/1-flower.jpg";
private static final String DESTINATION_FILE = "./out/1-flower.jpg";
public static void main(String[] args) throws IOException {
BufferedImage originalImage = ImageIO.read(new File(SOURCE_FILE));
BufferedImage resultImage = new BufferedImage(originalImage.getWidth(), originalImage.getHeight(), BufferedImage.TYPE_INT_RGB);
long startTime = System.currentTimeMillis();
// recolorSingleThreaded(originalImage, resultImage);
int numberOfThreads = 1;
recolorMultithreaded(originalImage, resultImage, numberOfThreads);
long endTime = System.currentTimeMillis();
long duration = endTime - startTime;
File outputFile = new File(DESTINATION_FILE);
File parentDir = outputFile.getParentFile();
if (parentDir != null && !parentDir.exists()) {
parentDir.mkdirs(); // Create the directory if it doesn't exist
}
ImageIO.write(resultImage, "jpeg", outputFile);
System.out.println("duration = " + duration);
}
public static void recolorMultithreaded(BufferedImage originalImage, BufferedImage resultImage, int numberOfThreads) {
List<Thread> threads = new ArrayList<>();
int width = originalImage.getWidth();
int height = originalImage.getHeight() / numberOfThreads;
for(int i = 0; i < numberOfThreads ; i++) {
final int threadMultiplier = i;
Thread thread = new Thread(() -> {
int xOrigin = 0 ;
int yOrigin = height * threadMultiplier;
recolorImage(originalImage, resultImage, xOrigin, yOrigin, width, height);
});
threads.add(thread);
}
for(Thread thread : threads) {
thread.start();
}
for(Thread thread : threads) {
try {
thread.join();
} catch (InterruptedException e) {
}
}
}
private static void recolorSingleThreaded(BufferedImage originalImage, BufferedImage resultImage) {
recolorImage(originalImage, resultImage, 0, 0, originalImage.getWidth(), originalImage.getHeight());
}
private static void recolorImage(BufferedImage originalImage, BufferedImage resultImage, int leftCorner, int topCorner, int width, int height) {
for (int x = leftCorner; x < leftCorner + width && x < originalImage.getWidth(); x++) {
for (int y = topCorner; y < topCorner + height && y < originalImage.getHeight(); y++) {
recolorPixel(originalImage, resultImage, x, y);
}
}
}
private static void recolorPixel(BufferedImage originalImage, BufferedImage resultImage, int x, int y) {
int rgb = originalImage.getRGB(x, y);
int red = getRed(rgb);
int green = getGreen(rgb);
int blue = getBlue(rgb);
int newRed;
int newGreen;
int newBlue;
if (isShadeOfGray(red, green, blue)) {
newRed = Math.min(255, red + 10);
newGreen = Math.max(0, green - 80);
newBlue = Math.max(0, blue - 20);
} else {
newRed = red;
newGreen = green;
newBlue = blue;
}
int newRGB = createRGBFromColors(newRed, newGreen, newBlue);
setRGB(resultImage, x, y, newRGB);
}
public static void setRGB(BufferedImage image, int x, int y, int rgb) {
image.getRaster().setDataElements(x, y, image.getColorModel().getDataElements(rgb, null));
}
public static boolean isShadeOfGray(int red, int green, int blue) {
return Math.abs(red - green) < 30 && Math.abs(red - blue) < 30 && Math.abs(green - blue) < 30;
}
public static int createRGBFromColors(int red, int green, int blue) {
int rgb = 0;
rgb |= blue;
rgb |= green << 8;
rgb |= red << 16;
rgb |= 0xFF000000;
return rgb;
}
public static int getRed(int rgb) {
return (rgb & 0x00FF0000) >> 16;
}
public static int getGreen(int rgb) {
return (rgb & 0x0000FF00) >> 8;
}
public static int getBlue(int rgb) {
return rgb & 0x000000FF;
}
}
SHOW IMAGE: BEFORE PROCESSING
SHOW IMAGE: AFTER PROCESSING
Single Threaded VS. Multithreaded
SHOW CODE: Throughput vs. Threads
import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpHandler;
import com.sun.net.httpserver.HttpServer;
import java.io.IOException;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
public class ThroughputHttpServer {
private static final String INPUT_FILE = "./resources/war_and_peace.txt";
private static final int NUMBER_OF_THREADS = 8;
public static void main(String[] args) throws IOException {
String text = new String(Files.readAllBytes(Paths.get(INPUT_FILE)));
startServer(text);
}
public static void startServer(String text) throws IOException {
HttpServer server = HttpServer.create(new InetSocketAddress(8000), 0);
server.createContext("/search", new WordCountHandler(text));
Executor executor = Executors.newFixedThreadPool(NUMBER_OF_THREADS);
server.setExecutor(executor);
server.start();
}
private static class WordCountHandler implements HttpHandler {
private String text;
public WordCountHandler(String text) {
this.text = text;
}
@Override
public void handle(HttpExchange httpExchange) throws IOException {
String query = httpExchange.getRequestURI().getQuery();
String[] keyValue = query.split("=");
String action = keyValue[0];
String word = keyValue[1];
if (!action.equals("word")) {
httpExchange.sendResponseHeaders(400, 0);
return;
}
long count = countWord(word);
byte[] response = Long.toString(count).getBytes();
httpExchange.sendResponseHeaders(200, response.length);
OutputStream outputStream = httpExchange.getResponseBody();
outputStream.write(response);
outputStream.close();
}
private long countWord(String word) {
long count = 0;
int index = 0;
while (index >= 0) {
index = text.indexOf(word, index);
if (index >= 0) {
count++;
index++;
}
}
return count;
}
}
}
SHOW IMAGE: Throughput VS. Threads
Problems
Linear Search with Finding One Occurrence
Problem Statement: The input consists of a large array (or list) of elements and a target value to search for. The goal is to utilize multiple threads to divide the search space and concurrently search for the target value, ultimately returning the index of the first occurrence of the target or Requirements: Constraints:SHOW PROBLEM
-1 indicating that the target is not present.
SHOW CODE
public class Solution {
private static final int SIZE = 280000;
private static final int NUM_THREADS = 4;
private static final Object mtx = new Object(); // Mutex for controlling access to foundIndex
private static volatile int foundIndex = -1;
private static void linearSearch(int threadId, int[] arr, int key) {
int chunkSize = arr.length / NUM_THREADS;
int start = threadId * chunkSize;
int end = (threadId == NUM_THREADS - 1) ? arr.length : start + chunkSize;
for (int i = start; i < end; ++i) {
// Early exit if foundIndex is set by another thread
synchronized (mtx) {
if (foundIndex != -1) {
break;
}
}
if (arr[i] == key) {
synchronized (mtx) {
if (foundIndex == -1) {
foundIndex = i;
break; // Exit after setting foundIndex
}
}
}
}
}
public static void main(String[] args) {
// Fill array with random numbers between 0-99
int[] arr = new int[SIZE];
for (int i = 0; i < SIZE; ++i) {
arr[i] = (int) (Math.random() * 100);
}
Thread[] threads = new Thread[NUM_THREADS];
int key = 9;
for (int i = 0; i < NUM_THREADS; ++i) {
final int threadId = i;
threads[i] = new Thread(() -> linearSearch(threadId, arr, key));
threads[i].start();
}
for (Thread thread : threads) {
try {
thread.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
if (foundIndex == -1) {
System.out.println("Element not found in the array.");
} else {
System.out.println("Element found at index: " + foundIndex);
}
}
}
Linear Search for All Occurrences
Problem Statement: The input consists of a large array (or list) of elements and a target value to search for. The goal is to utilize multiple threads to divide the search space and concurrently search for the target value, ultimately returning the index of all occurrences of the target or an empty list indicating that the target is not present. Requirements: Constraints:SHOW PROBLEM
SHOW CODE
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
public class Solution {
private static final int SIZE = 4000;
private static final int NUM_THREADS = 4;
// Mutex for controlling access to foundPlaces
private static final Object lockObj = new Object();
private static List<Integer> foundPlaces = new ArrayList<>();
private static void linearSearch(int threadId, int[] arr, int key) {
int chunkSize = arr.length / NUM_THREADS;
int start = threadId * chunkSize;
int end = (threadId == NUM_THREADS - 1) ? arr.length : start + chunkSize;
for (int i = start; i < end; ++i) {
if (arr[i] == key) {
synchronized (lockObj) { // Lock when modifying foundPlaces
foundPlaces.add(i); // Append the index to foundPlaces
}
}
}
}
public static void main(String[] args) {
// Create an array and fill it with random numbers between 0 and 99
int[] arr = new int[SIZE];
Random random = new Random();
for (int i = 0; i < SIZE; ++i) {
arr[i] = random.nextInt(100);
}
Thread[] threads = new Thread[NUM_THREADS]; // List to hold the threads
int key = 9; // Element to find
// Start the threads
for (int i = 0; i < NUM_THREADS; ++i) {
final int threadId = i;
threads[i] = new Thread(() -> linearSearch(threadId, arr, key));
threads[i].start();
}
// Join the threads with the main thread
for (Thread thread : threads) {
try {
thread.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// Display the result
if (foundPlaces.isEmpty()) {
System.out.println("Element not found in the array.");
} else {
System.out.print("Element found at indices: ");
synchronized (lockObj) { // Lock when reading from foundPlaces
for (int index : foundPlaces) {
System.out.print(index + " ");
}
}
System.out.println();
}
}
}
The synchonization block can be removed when reading values from SHOW NOTES
foundPlaces, as no threads modify the foundPlaces after the join() method is called.// Display the result
if (foundPlaces.isEmpty()) {
System.out.println("Element not found in the array.");
} else {
System.out.print("Element found at indices: ");
// No need for synchronization here, as no threads are modifying foundPlaces now
// synchronized (lockObj) { // Lock when reading from foundPlaces
for (int index : foundPlaces) {
System.out.print(index + " ");
}
// }
System.out.println();
}
Linear Search with Indices and Occurrences
Problem Statement: You are given a large array (or list) of elements and a target value to search for. Your task is to implement a linear search that finds all occurrences of the target value in the array using multiple threads. Each thread should search a specific segment of the array, and the results (indices where the target is found) should be stored in a shared collection (e.g., a list). Additionally, a shared variable should keep track of the count of occurrences of the target value. Requirements: Constraints:SHOW PROBLEM
SHOW CODE
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
public class Solution {
private static final int SIZE = 4000;
private static final int NUM_THREADS = 4;
private static final List<Integer> foundIndices = new ArrayList<>(); // Shared list to store the indices of all occurrences
private static int occurrencesCount = 0; // Shared variable to store the count of occurrences
private static final Object indicesLock = new Object(); // Lock for synchronizing access to foundIndices
private static final Object countLock = new Object(); // Lock for synchronizing access to occurrencesCount
// Function executed by each thread to search for indices and occurrences
private static void searchIndicesOccurrences(int threadId, int[] arr, int key) {
int chunkSize = arr.length / NUM_THREADS;
int start = threadId * chunkSize;
int end = (threadId == NUM_THREADS - 1) ? arr.length : start + chunkSize;
List<Integer> localIndices = new ArrayList<>();
int localCount = 0;
for (int i = start; i < end; ++i) {
if (arr[i] == key) {
localIndices.add(i);
localCount++;
}
}
if (!localIndices.isEmpty()) {
synchronized (indicesLock) {
foundIndices.addAll(localIndices);
}
}
if (localCount > 0) {
synchronized (countLock) {
occurrencesCount += localCount;
}
}
}
public static void main(String[] args) {
// Create an array and fill it with random numbers between 0 and 99
int[] arr = new int[SIZE];
Random random = new Random();
for (int i = 0; i < SIZE; ++i) {
arr[i] = random.nextInt(100);
}
Thread[] threads = new Thread[NUM_THREADS]; // Array of threads
int key = 9; // Element to find
// Start the threads
for (int i = 0; i < NUM_THREADS; ++i) {
int threadId = i;
threads[i] = new Thread(() -> searchIndicesOccurrences(threadId, arr, key));
threads[i].start();
}
// Wait for all threads to complete
for (Thread thread : threads) {
try {
thread.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
// Output the results
if (foundIndices.isEmpty()) {
System.out.println("Element not found in the array.");
} else {
System.out.print("Element found " + occurrencesCount + " times at indices: ");
for (int index : foundIndices) {
System.out.print(index + " ");
}
System.out.println();
}
}
}
Min/Max/Sum
Problem Statement: The input consists of a large array (or list) of elements. The goal is to efficiently calculate the minimum, maximum, and sum of elements in the array using multithreading. The array is divided into multiple segments, with each thread processing a specific segment. The threads will then return partial results (minimum, maximum, and sum for their respective segments), which will be combined to compute the final values for the entire array. Requirements: Multithreading: Divide the array into multiple segments, with each thread processing a specific portion of the array. Each thread should independently compute the minimum, maximum, and sum for its assigned segment. Combine Results: Once each thread has finished processing, their partial results (min, max, and sum for the segment) should be combined to compute the final minimum, maximum, and sum for the entire array. Efficiency: The program should use multithreading to optimize the processing time, especially when dealing with large arrays containing millions of elements. The number of threads should be adjustable for optimal performance. Constraints: The array can be large (millions of elements). The number of threads used should be adjustable based on the size of the array and the system’s capabilities. The array can contain both positive and negative numbers, and the minimum, maximum, and sum should be computed accurately, including for arrays with all negative values.SHOW PROBLEM
SHOW CODE
import java.util.Random;
public class Solution {
private static final int DATA_SIZE = 100;
private static final int NUMBER_OF_THREADS = 4;
private static int[] data = new int[DATA_SIZE];
private static int[] threadResultsSum = new int[NUMBER_OF_THREADS];
private static int[] threadResultsMin = new int[NUMBER_OF_THREADS];
private static int[] threadResultsMax = new int[NUMBER_OF_THREADS];
public static void main(String[] args) throws InterruptedException {
// Initialize data array
Random random = new Random();
for (int i = 0; i < DATA_SIZE; i++) {
data[i] = random.nextInt(500);
}
Thread[] threads = new Thread[NUMBER_OF_THREADS * 3];
// Start threads for sum, min, and max calculations
for (int i = 0; i < NUMBER_OF_THREADS; i++) {
final int threadId = i;
final int start = threadId * (DATA_SIZE / NUMBER_OF_THREADS);
final int end = (threadId + 1) * (DATA_SIZE / NUMBER_OF_THREADS);
threads[threadId] = new Thread(() -> threadedSum(threadId, start, end));
threads[threadId + NUMBER_OF_THREADS] = new Thread(() -> threadedMin(threadId, start, end));
threads[threadId + NUMBER_OF_THREADS * 2] = new Thread(() -> threadedMax(threadId, start, end));
threads[threadId].start();
threads[threadId + NUMBER_OF_THREADS].start();
threads[threadId + NUMBER_OF_THREADS * 2].start();
}
// Wait for threads to finish
for (Thread thread : threads) {
thread.join();
}
// Aggregate results from threads
int totalSum = 0;
for (int sum : threadResultsSum) {
totalSum += sum;
}
int min = Integer.MAX_VALUE;
for (int minResult : threadResultsMin) {
min = Math.min(min, minResult);
}
int max = Integer.MIN_VALUE;
for (int maxResult : threadResultsMax) {
max = Math.max(max, maxResult);
}
System.out.println("Sum is " + totalSum);
System.out.println("Min is " + min);
System.out.println("Max is " + max);
}
private static void threadedSum(int threadId, int start, int end) {
int sum = 0;
for (int i = start; i < end; i++) {
sum += data[i];
}
threadResultsSum[threadId] = sum;
}
private static void threadedMin(int threadId, int start, int end) {
int min = Integer.MAX_VALUE;
for (int i = start; i < end; i++) {
min = Math.min(min, data[i]);
}
threadResultsMin[threadId] = min;
}
private static void threadedMax(int threadId, int start, int end) {
int max = Integer.MIN_VALUE;
for (int i = start; i < end; i++) {
max = Math.max(max, data[i]);
}
threadResultsMax[threadId] = max;
}
}
Pi Calculation
Problem Statement The problem is to estimate the value of Pi using the Monte Carlo simulation method. In this method, we randomly generate points within a square and check how many of these points fall inside a circle inscribed within the square. The ratio of points inside the circle to the total points generated provides an approximation of Pi. Requirements: Random Point Generation: Circle Inside the Square: Monte Carlo Estimation of Pi: Multithreading: Output: Constraints:SHOW PROBLEM
SHOW CODE
import java.util.Random;
public class Solution {
private static final int NUM_THREADS = 10;
private static final int NUMBER_OF_TOSSES = 100000000;
private static int[] results = new int[NUM_THREADS];
public static void main(String[] args) throws InterruptedException {
Thread[] threads = new Thread[NUM_THREADS];
for (int i = 0; i < NUM_THREADS; i++) {
final int threadId = i;
threads[i] = new Thread(() -> {
Random rand = new Random();
int start = threadId * NUMBER_OF_TOSSES / NUM_THREADS;
int end = (threadId + 1) * NUMBER_OF_TOSSES / NUM_THREADS;
int count_in_circle = 0;
for (int j = start; j < end; j++) {
double x = rand.nextDouble() * 2 - 1; // Random x in range [-1, 1]
double y = rand.nextDouble() * 2 - 1; // Random y in range [-1, 1]
if (x * x + y * y <= 1) { // If point is inside the circle
count_in_circle++;
}
}
results[threadId] = count_in_circle;
});
threads[i].start();
}
for (Thread thread : threads) {
thread.join();
}
// Compute final estimate of Pi
int total_inside = 0;
for (int result : results) {
total_inside += result;
}
double pi_estimate = 4.0 * total_inside / NUMBER_OF_TOSSES;
System.out.println("PI = " + pi_estimate);
}
}
Introduction to Monte Carlo Estimation of $\pi$ Suppose a square with a circle inscribed inside it, as shown in the image below. The ratio of the area of the circle to the area of the square is
$$
\frac{A_{\text{circle}}}{A_{\text{square}}} = \frac{\pi r^2}{4 r^2} = \frac{\pi}{4}
$$ Therefore, the value of $\pi$ can be estimated by the formula: $$
\pi \approx 4 \times \frac{\text{Number of points inside the circle}}{\text{Total number of points inside the square}}
$$ Visulization:SHOW NOTES
The circle has a radius of $r$, and the square has side length $2r$. The area of the circle is $A_{\text{circle}} = \pi r^2$, and the area of the square is $A_{\text{square}} = (2r)^2 = 4r^2$
JUC
CompletableFuture
Future Interface
The Future interface in Java defines several methods for managing asynchronous tasks, such as retriving the execution result, canceling a task, and checking if a task has been canceled or completed. It includes four commonly used methods:
get(),isDone(),cancel(), andisCancelled().
- The
get()mthod retrives the result of computation. Note: If the computation is not yet complete, this method will block until the result is available. - The
isDone()method checks whether the computation has finished executing. - The
cancel()method attempts to cancel the task execution. - The
isCancelled()method checks if the task has been canceled before completion.
The FutureTask class is a concrete implementation of the Future interface. It implements both the Runnable and Future interfaces, allowing it to be executed by a thread while also returning a result. Its constructor takes a
Callable<V>, enabling the task to return a value. The class diagram is shown below:![FutureTask Class Diagram]()
The Future combined with a thread pool enhances execution performance efficiently. However, its drarback lies in the
get()method, which is blocking, and theisDone()method, whose polling behavior consumes CPU resources.
SHOW CODE
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
class Main {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executor = Executors.newSingleThreadExecutor();
Future<String> future = executor.submit(() -> {
Thread.sleep(3000); // Simulate a long running task
return "Task Completed";
});
System.out.println("Waiting for result...");
// Block the main thread until the result is available
String result = future.get();
System.out.println("Result: " + result);
executor.shutdown();
}
}
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
class Main {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executor = Executors.newSingleThreadExecutor();
Future<String> future = executor.submit(() -> {
Thread.sleep(3000); // Simulate a long-running task
return "Task Completed";
});
// Busy-waiting, consuming CPU cycles
while (!future.isDone()) {
System.out.println("Checking if task is done...");
Thread.sleep(100); // Reducing CPU usage slightly, but still inefficient
}
System.out.println("Result: " + future.get());
executor.shutdown();
}
}
CompletableFuture
The CompletableFuture implements both the Future and CompletionStage interfaces. The CompletionStage represents a computation step that can be completed asynchronously. Unlike Future, CompletableFuture provides advanced mechanisms for handling asynchronous tasks, including non-blocking execution, method chaining, exception handling, combining futures, and thread pool customization.
- Non-blocking Execution: Unlike
Future.get(), whcih blocks until the result is available, CompletableFuture enables asynchronous execution without blocking the main thread. - Chaining Supports method chaining with
thenApply,thenCompose, andthenRunto process results sequently. - Exception Handling: Provides
exceptionallyandhandlemethods to gracefully handle errors. - Combining Futures: Supports combining multiple futures using
thenCombine,allOf, andanyOffor parallel execution. - Thread Pool Customization: Allows executing tasks in a custom thread pool for better resource management.
SHOW CODE
import java.util.concurrent.*;
class Main {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
try {
Thread.sleep(2000);
System.out.println("Task executed in a separate thread");
} catch (InterruptedException e) {
e.printStackTrace();
}
});
System.out.println("Main thread is not blocked.");
future.join();
}
}
import java.util.concurrent.*;
class Main {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello")
.thenApply(result -> result + ", World!")
.thenApply(String::toUpperCase);
System.out.println(future.join()); // Output: HELLO, WORLD!
}
}
import java.util.concurrent.*;
class Main {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
if (true) {
throw new RuntimeException("Something went wrong!");
}
return "Success";
}).exceptionally(ex -> "Recovered from: " + ex.getMessage());
// Recovered from: java.lang.RuntimeException: Something went wrong!
System.out.println(future.join());
}
}
import java.util.concurrent.*;
class Main {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Task 1");
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "Task 2");
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> "Task 3");
CompletableFuture allFutures = CompletableFuture.allOf(future1, future2, future3);
// Retrieve results from individual futures
String result1 = future1.get();
String result2 = future2.get();
String result3 = future3.get();
System.out.println(result1);
System.out.println(result2);
System.out.println(result3);
allFutures.join(); // Wait for all tasks to complete
System.out.println("All tasks completed!");
}
}
The advantage of
CompletableFutureis that when an asynchronous task completes, it automatically invokes the callback method. Once the main thread sets up the callback, it no longer needs to monitor the asynchronous task, and allows execution to proceed in sequence. Additionally, if the asynchronous task fails, it automatically triggers the corresponding error-handling method.
By default,
CompletableFuture.supplyAsync()andrunAsync()utilize theForkJoinPool.commonPool(), a shared thread pool based on the ForkJoin framework. To use a custom thread pool, you need to provide a customExecutorwhen calling these methods.
SHOW CODE
import java.util.concurrent.*;
class Main {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture.runAsync(() -> {
// Running in: ForkJoinPool.commonPool-worker-1
System.out.println("Running in: " + Thread.currentThread().getName());
});
// Sleep to allow async task to execute
try {
Thread.sleep(100);
} catch (InterruptedException e) {
}
}
}
import java.util.concurrent.*;
class Main {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executor = Executors.newFixedThreadPool(5);
CompletableFuture.runAsync(() ->
// Custom pool: pool-1-thread-1
System.out.println("Custom pool: "
+ Thread.currentThread().getName()), executor);
}
}
The main difference between
CompletableFuture.get()andCompletableFuture.join()is exception handling.get()throws checked exceptions (ExecutionException,InterruptedException), requiring explicit handling, whilejoin()throws an uncheckedCompletionException, eliminating the need for explicit exception handling.
The main differences between
thenRun(),thenAccept(), andthenApplyare
thenRun()runs a task after previous stage completes, but does not take its result as input.thenAccept()runs a task after previous stage completes and consumes its result.thenApply()runs a task after previous stage completes and transform its result.
CompletableFuture.supplyAsync(() -> "Task Completed")
.thenRun(() -> System.out.println("Follow-up action executed!"));
CompletableFuture.supplyAsync(() -> "Hello, World!")
.thenAccept(result -> System.out.println("Result: " + result));
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello")
.thenApply(result -> result + ", World!");
System.out.println(future.join()); // Output: Hello, World!
| Method | Takes Previous Result? | Returns a Value? | Use Case |
|---|---|---|---|
thenRun |
❌ No | ❌ No (Void) |
Just run a task after completion |
thenAccept |
✅ Yes | ❌ No (Void) |
Consume the result without returning a new one |
thenApply |
✅ Yes | ✅ Yes (R) |
Transform the result and return a new value |
Java Locks
Optimistic Lock vs. Pessimistic Lock
Optimistic Lock assumes minimal contention, allowing multiple threads to read and write data concurrently. Instead of using locks, it relies on versioning or CAS (Compare-And-Swap) to detect conflicts during updates. If a conflict is detected (i.e., another thread has modified the data), the operation is retried. It is particularly effective in high-read, low-write scenarios.
SHOW CODE
# Use versioning in SQL
UPDATE products
SET stock = stock - 1, version = version + 1
WHERE id = 1 AND version = 5;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
class Main {
public static void main(String[] args) throws ExecutionException, InterruptedException {
AtomicInteger count = new AtomicInteger(0);
int expectedValue, newValue;
do {
expectedValue = count.get();
newValue = expectedValue + 1;
} while (!count.compareAndSet(expectedValue, newValue));
System.out.println("Updated Count: " + count.get());
}
}
Pessimistic Lock assumes high contention and prevents concurrent modification by blocking the resource. Other threads must wait until the lock is released. It is suitable for high-write, low-read scenerios. In Java, the
synchronizedandReentrantLockare implementations of pessimistic lock.
SHOW CODE
class PessimisticLockExample {
private int count = 0;
public synchronized void increment() {
count++;
}
}
import java.util.concurrent.locks.ReentrantLock;
public class PessimisticLockExample {
private int count = 0;
private final ReentrantLock lock = new ReentrantLock();
public void increment() {
lock.lock();
try {
count++;
} finally {
lock.unlock();
}
}
}
synchronized Related Problems
SHOW CODE
class Phone {
public synchronized void sendEmail() {
System.out.println("send email");
}
public synchronized void sendSMS() {
System.out.println("send SMS");
}
}
class Main {
public static void main(String[] args) {
Phone phone = new Phone();
new Thread(() -> {
phone.sendEmail();
}, "A").start();
// Ensure thread A starts first
try {
TimeUnit.MILLISECONDS.sleep(300);
} catch (InterruptedException e) {
e.printStackTrace();
}
new Thread(() -> {
phone.sendSMS();
}, "B").start();
}
}
The above code first outputs
send email, followed bysend SMS. This happens because bothsendEmail()andsendSMS()are synchronized methods that lock the same object. As a result, the thread that acquires the lock first executes first, while other threads must wait until the lock is released.
SHOW CODE
class Phone {
public synchronized void sendEmail() {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("send email");
}
public synchronized void sendSMS() {
System.out.println("send SMS");
}
}
class Main {
public static void main(String[] args) {
Phone phone = new Phone();
new Thread(() -> {
phone.sendEmail();
}, "A").start();
// Ensure thread A starts first
try {
TimeUnit.MILLISECONDS.sleep(300);
} catch (InterruptedException e) {
e.printStackTrace();
}
new Thread(() -> {
phone.sendSMS();
}, "B").start();
}
}
The above code first outputs
send email, followed bysend SMS. This happens because bothsendEmail()andsendSMS()are synchronized methods that lock the same object and executessleep()method doesn’t release a lock. Therefore, the thread that acquires the lock first executes first, while other threads must wait until the lock is released.
SHOW CODE
class Phone {
public synchronized void sendEmail() {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("send email");
}
public synchronized void sendSMS() {
System.out.println("send SMS");
}
public void hello() {
System.out.println("hello");
}
}
class Main {
public static void main(String[] args) {
Phone phone = new Phone();
new Thread(phone::sendEmail, "A").start();
// Ensure thread A starts first
try {
TimeUnit.MILLISECONDS.sleep(300);
} catch (InterruptedException e) {
e.printStackTrace();
}
new Thread(() -> {
phone.hello();
}, "B").start();
}
}
The above code outputs
helloimmediately, followed bysend email. This is because thehello()method is not synchronized and does not require a lock, allowing it to execute without waiting.
SHOW CODE
import java.util.concurrent.TimeUnit;
class Phone {
public synchronized void sendEmail() {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("send email");
}
public synchronized void sendSMS() {
System.out.println("send SMS");
}
}
class Main {
public static void main(String[] args) {
Phone phone1 = new Phone();
Phone phone2 = new Phone();
new Thread(() -> {
phone1.sendEmail();
}, "A").start();
// Ensure thread A starts first
try {
TimeUnit.MILLISECONDS.sleep(300);
} catch (InterruptedException e) {
e.printStackTrace();
}
new Thread(() -> {
phone2.sendSMS();
}, "B").start();
}
}
The above code outputs
send SMSfirst, followed bysend email. This happens because thesynchronizedkeyword locks the object instances (phone1andphone2), not the class itself. Sincephone1andphone2are different objects, their locks do not interfere with each other.
In this case,
Thread Astarts first and acquires the lock onphone1, but sleep for 3 seconds inside thesendEmail()method. Meanwhile,Thread Bstarts shortly after and acquires the lock onphone2. SinceThread Bdoes not need to wait forThread Ato release its lock (as they operate on different objects), it executessendSMS()immediately. As a result,Thread Boutputssend SMSfirst, whileThread Aoutputssend emailafter its sleep period ends.
SHOW CODE
import java.util.concurrent.TimeUnit;
class Phone {
public static synchronized void sendEmail() {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("send email");
}
public static synchronized void sendSMS() {
System.out.println("send SMS");
}
}
class Main {
public static void main(String[] args) {
Phone phone = new Phone();
new Thread(() -> {
phone.sendEmail();
}, "A").start();
// Ensure thread A starts first
try {
TimeUnit.MILLISECONDS.sleep(300);
} catch (InterruptedException e) {
e.printStackTrace();
}
new Thread(() -> {
phone.sendSMS();
}, "B").start();
}
}
The above code outputs
send emailfirst, followed bysend SMS. This happens because thestatic synchronizedkeyword locks thePhone.classobject, not individual instances of the class. Since both methods (sendEmailandsend SMS) arestatic synchronized, they share the same lock on thePhone.classobject. As a result, their execution is sequential.
SHOW CODE
import java.util.concurrent.TimeUnit;
class Phone {
public static synchronized void sendEmail() {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("send email");
}
public static synchronized void sendSMS() {
System.out.println("send SMS");
}
}
class Main {
public static void main(String[] args) {
Phone phone1 = new Phone();
Phone phone2 = new Phone();
new Thread(() -> {
phone1.sendEmail();
}, "A").start();
// Ensure thread A starts first
try {
TimeUnit.MILLISECONDS.sleep(300);
} catch (InterruptedException e) {
e.printStackTrace();
}
new Thread(() -> {
phone2.sendSMS();
}, "B").start();
}
}
The above code outputs
send emailfirst, followed bysend SMS. Even though there are two different objects (phone1andphone2), thestatic synchronizedkeyword locks thePhone.classobject, not the individual instances. This means that bothsendEmail()andsendSMS()share the same lock at the class level, ensuring sequential execution.
SHOW CODE
import java.util.concurrent.TimeUnit;
class Phone {
public static synchronized void sendEmail() {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("send email");
}
public synchronized void sendSMS() {
System.out.println("send SMS");
}
}
class Main {
public static void main(String[] args) {
Phone phone = new Phone();
new Thread(() -> {
phone.sendEmail();
}, "A").start();
// Ensure thread A starts first
try {
TimeUnit.MILLISECONDS.sleep(300);
} catch (InterruptedException e) {
e.printStackTrace();
}
new Thread(() -> {
phone.sendSMS();
}, "B").start();
}
}
The above code outputs
send SMSfirst, followed bysend email. This happens because thesynchronizedkeyword locks individual object instance, whilestatic synchronizedlocks the class-level object. In this case,sendSMS()locks thephoneobject, andsendEmail()locks thePhone.classobject. Since these locks are independent,Thread Bdoes not need to wait forThread Ato release its lock. As a result,send SMSis printed first, whilesend emailis printed after the 3-second sleep.
SHOW CODE
import java.util.concurrent.TimeUnit;
class Phone {
public static synchronized void sendEmail() {
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("send email");
}
public static synchronized void sendSMS() {
System.out.println("send SMS");
}
public void hello() {
System.out.println("hello");
}
}
class Main {
public static void main(String[] args) {
Phone phone1 = new Phone();
Phone phone2 = new Phone();
new Thread(() -> {
phone1.sendEmail();
}, "A").start();
// Ensure thread A starts first
try {
TimeUnit.MILLISECONDS.sleep(300);
} catch (InterruptedException e) {
e.printStackTrace();
}
new Thread(() -> {
phone2.hello();
phone2.sendSMS();
}, "B").start();
}
}
The above code outputs
hellofirst, followed bysend emailandsend SMS. This is becausesendEmail()andsendSMS()are bothstatic synchronized, meaning they lock thePhone.classobject, enforcing sequential execution. In contrast, thehello()method is a regular method that does not require a lock, allowing it to execute immediately.
synchronized Byte Code Analysis
SHOW CODE
class Main {
private final Object mtx = new Object();
public void f() {
synchronized (mtx) {
System.out.println("hello...");
}
}
}
public void f();
Code:
0: aload_0
1: getfield #7 // Field mtx:Ljava/lang/Object;
4: dup
5: astore_1
🌟 6: monitorenter
7: getstatic #13 // Field java/lang/System.out:Ljava/io/PrintStream;
10: ldc #19 // String hello...
12: invokevirtual #21 // Method java/io/PrintStream.println:(Ljava/lang/String;)V
15: aload_1
🌟 16: monitorexit
17: goto 25
20: astore_2
21: aload_1
🌟 22: monitorexit
23: aload_2
24: athrow
25: return
From the above bytecode, the
synchronizedblock translates intomonitorenterandmonitorexitinstructions. The lock is acquired before entering the block and released before exiting. The secondmonitorexitensures that the lock is properly released if an exception occurs, as shown in the following code:
SHOW CODE
class Main {
private final Object mtx = new Object();
public void f() {
synchronized (mtx) {
System.out.println("hello...");
throw new RuntimeException("Exception occurred");
}
}
}
public void f();
Code:
0: aload_0
1: getfield #7 // Field mtx:Ljava/lang/Object;
4: dup
5: astore_1
🌟 6: monitorenter
7: getstatic #13 // Field java/lang/System.out:Ljava/io/PrintStream;
10: ldc #19 // String hello...
12: invokevirtual #21 // Method java/io/PrintStream.println:(Ljava/lang/String;)V
15: new #27 // class java/lang/RuntimeException
18: dup
19: ldc #29 // String Exception occurred
21: invokespecial #31 // Method java/lang/RuntimeException."<init>":(Ljava/lang/String;)V
24: athrow
25: astore_2
26: aload_1
🌟 27: monitorexit
28: aload_2
29: athrow
SHOW CODE
class Main {
public synchronized void f() {
System.out.println("hello...");
}
public static synchronized void m() {
System.out.println("HELLO...");
}
}
public synchronized void f();
descriptor: ()V
flags: (0x0021) ACC_PUBLIC, 🌟ACC_SYNCHRONIZED🌟
Code:
stack=2, locals=1, args_size=1
0: getstatic #7 // Field java/lang/System.out:Ljava/io/PrintStream;
3: ldc #13 // String hello...
5: invokevirtual #15 // Method java/io/PrintStream.println:(Ljava/lang/String;)V
8: return
LineNumberTable:
line 3: 0
line 4: 8
public static synchronized void m();
descriptor: ()V
flags: (0x0029) ACC_PUBLIC, 🌟ACC_STATIC, ACC_SYNCHRONIZED🌟
Code:
stack=2, locals=0, args_size=0
0: getstatic #7 // Field java/lang/System.out:Ljava/io/PrintStream;
3: ldc #21 // String HELLO...
5: invokevirtual #15 // Method java/io/PrintStream.println:(Ljava/lang/String;)V
8: return
LineNumberTable:
line 7: 0
line 8: 8
From the above bytecode, the JVM checks if the
ACC_SYNCHRONIZEDflag is set. If the flag is present, the thread acquires the appropriate monitor lock before executing the method. For instance methods, the lock is acquired on the object instance (this), while for static methods, the lock is acquired on the class object (e.g.,Main.class). After the method completes execution, the thread releases the lock. TheACC_STATICflag is used to determine whether the method is static, which in turn determines whether the lock is applied at the class level or the instance level.
Why any object in Java can be a lock?
Any object in Java can be a lock because every object has an intrinsic monitor lock managed by the JVM. The
ObjectMonitorstructure is part of the JVM’s implementation of this mechanism, tracking details like the lock owner, EntryList, etc. For example, when a thread callssynchronized(obj), the JVM checks theObjectMonitorassociated withobj. If the lock is available, the thread acquires it and sets itself as theowner. If the lock is held by another thread, the current thread waits in theEntryList. When the lock is released, the JVM wakes up a waiting thread (if any) to acquire the lock.
ObjectMonitor() {
_header = NULL;
_count = 0;
_waiters = 0;
_recursions = 0;
_object = NULL;
_owner = NULL;
_MailSet = NULL;
_MailSetLock = 0;
_Responsible = NULL;
_succ = NULL;
_cxq = NULL;
FreeNext = NULL;
_EntryList = NULL;
_SpinFreq = 0;
_SpinClock = 0;
OwnerIsThread = 0;
_previous_owner_tid = 0;
}
Fair Lock vs. Unfair Lock
In Java, locks can be calssified into fair locks and unfair locks based on their approach to thread scheduling. A fair lock ensures that threads acquire the lock in the exact order they requested it, following a first-come, first-served priciple. While this prevents thread starvation, it incurs additional overhead to maintain the order of threads. On the other hand, an unfair lock does not guarantee any specific order for thread access. A thread can acquire the lock even if it arrived later than others. This eliminates the need to maintain a strict order, improving efficiency, but it may result in thread starvation, where some threads could wait indefinitely for access.
SHOW CODE
import java.util.concurrent.locks.ReentrantLock;
public class Main {
private static final ReentrantLock unfairLock = new ReentrantLock(); // Unfair lock
private static final ReentrantLock fairLock = new ReentrantLock(true); // Fair lock
public static void main(String[] args) {
System.out.print("Testing Unfair Lock: ");
testLock(unfairLock);
System.out.print("Testing Fair Lock: ");
testLock(fairLock);
}
private static void testLock(ReentrantLock lock) {
long startTime = System.currentTimeMillis();
Runnable task = () -> {
for (int i = 0; i < 10000; i++) {
lock.lock();
try {
// Simulate work
} finally {
lock.unlock();
}
}
};
Thread[] threads = new Thread[10];
for (int i = 0; i < threads.length; i++) {
threads[i] = new Thread(task);
threads[i].start();
}
for (Thread thread : threads) {
try {
thread.join();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
long endTime = System.currentTimeMillis();
System.out.println("Time taken: " + (endTime - startTime) + " ms");
}
}
SHOW OUTPUT
Testing Unfair Lock: Time taken: 11 ms
Testing Fair Lock: Time taken: 560 ms
Recursive Lock
A recursive lock in Java allows a thread to acquire the same lock multiple times without causing a deadlock. It is particularly useful in scenarios where a thread holding a lock calls another method or enters a synchronized block that requires the same lock. By default, the
synchronizedkeyword andReentrantLockclass support recursive locking mechanism.
The recursive lock maintains a hold count, which tracks how many times a thread has acquired the lock. Each time the thread acquires the lock, the hold count is incremented, and it is decremented when the lock is released. The lock is fully released only when the hold count reaches zero.
Under the hood, when the
monitorentercommand is executed, the JVM checks theMonitorObject’s_countfield. If_countis zero, it means the lock is not held by any other thread. In this case, the JVM sets the_ownerfield to the current thread and increments the_count. If_countis not zero and the_owneris the current thread, the JVM increments the_count(indicating reentrancy). Otherwise, if the lock is held by another thread, the current thread must wait until the lock is released (i.e.,_countreaches zero).
When the
monitorexitcommand is executed, the JVM decrements the_count. Once_countreaches zero, the lock is considered fully released, and the_ownerfield is cleared, allowing other threads to acquire the lock.
SHOW CODE
import java.util.concurrent.locks.ReentrantLock;
public class Main {
private static final ReentrantLock lock = new ReentrantLock();
public static void main(String[] args) {
recursiveMethod(3);
}
public static void recursiveMethod(int count) {
lock.lock(); // Acquire the lock
try {
System.out.println(Thread.currentThread().getName() + " acquired the lock. Hold count: " + lock.getHoldCount());
if (count > 0) {
recursiveMethod(count - 1); // Recursive call
}
} finally {
lock.unlock(); // Release the lock
System.out.println(Thread.currentThread().getName() + " released the lock. Hold count: " + lock.getHoldCount());
}
}
}
SHOW OUTPUT
main acquired the lock. Hold count: 1
main acquired the lock. Hold count: 2
main acquired the lock. Hold count: 3
main acquired the lock. Hold count: 4
main released the lock. Hold count: 3
main released the lock. Hold count: 2
main released the lock. Hold count: 1
main released the lock. Hold count: 0
Deadlock
A deadlock is a situation where two or more threads are permanently blcoked, each waiting for a resource held by another, preventing further progress. For a deadlock to occur, four conditions must be met simultaneously: mutual exclusion, hold and wait, no preemption, and circular wait.
- Mutual Exclusion: A resource must be non-sharable, meaning only one thread can use it at a time.
- Hold and Wait: A thread holding at least one resource must be waiting to acquire additional resources held by other threads.
- No Preemption: Resources cannot be forcibly taken from a thread; they must be released voluntarily.
- Circular Wait: A circular chain of two or more threads must exist, where each thread is waiting for a resource held by the next in the chain.
To detect a deadlock in Java, use the
jpscommand to get the process ID, then runjstack process_idto analyze thread states and identify deadlocks.
SHOW CODE
public class Main {
private static final Object lock1 = new Object();
private static final Object lock2 = new Object();
public static void main(String[] args) {
Thread thread1 = new Thread(() -> {
synchronized (lock1) {
System.out.println("Thread 1: Holding lock 1...");
try {
Thread.sleep(100); // Simulate work
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Thread 1: Waiting for lock 2...");
synchronized (lock2) {
System.out.println("Thread 1: Acquired lock 2!");
}
}
});
Thread thread2 = new Thread(() -> {
synchronized (lock2) {
System.out.println("Thread 2: Holding lock 2...");
try {
Thread.sleep(100); // Simulate work
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Thread 2: Waiting for lock 1...");
synchronized (lock1) {
System.out.println("Thread 2: Acquired lock 1!");
}
}
});
thread1.start();
thread2.start();
}
}
$ jps
76995 Launcher
76996 Main
77127 Jps
2327 Main
--------------
$ jstack 76996
"Thread-0":
at dev.signalyu.warmup.Main.lambda$main$0(Main.java:18)
🌟 - waiting to lock <0x000000070ffce280> (a java.lang.Object)
🌟 - locked <0x000000070ffce270> (a java.lang.Object)
at dev.signalyu.warmup.Main$$Lambda$14/0x000000012d001208.run(Unknown Source)
at java.lang.Thread.run(java.base@17.0.9/Thread.java:842)
"Thread-1":
at dev.signalyu.warmup.Main.lambda$main$1(Main.java:33)
🌟 - waiting to lock <0x000000070ffce270> (a java.lang.Object)
🌟 - locked <0x000000070ffce280> (a java.lang.Object)
at dev.signalyu.warmup.Main$$Lambda$15/0x000000012d001428.run(Unknown Source)
at java.lang.Thread.run(java.base@17.0.9/Thread.java:842)
Found 1 deadlock. 🌟
Thread Interruption
In Java, thread interruption is a mechanism that allows one thread to signal another to stop its execution. The three common approaches to interrupt a thread are using a
volatileflag, anAtomicBoolean, or theinterrupt()method.
SHOW CODE
import java.util.concurrent.TimeUnit;
public class Main {
private static volatile boolean isStop = false;
public static void main(String[] args) {
new Thread(() -> {
while (!isStop) {
try {
TimeUnit.MILLISECONDS.sleep(500);
System.out.println("isStop: " + isStop);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
System.out.println("isStop --> " + isStop);
}, "t1").start();
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
// Signal threads to stop
new Thread(() -> {
isStop = true;
}, "t2").start();
}
}
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
public class Main {
private static final AtomicBoolean atomicBoolean = new AtomicBoolean(false);
public static void main(String[] args) {
new Thread(() -> {
while (!atomicBoolean.get()) {
try {
TimeUnit.MILLISECONDS.sleep(500);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
System.out.println("Atomic Boolean: " + atomicBoolean);
}
System.out.println("Atomic Boolean --> " + atomicBoolean);
}, "t1").start();
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
// Signal threads to stop
new Thread(() -> {
atomicBoolean.set(true);
}, "t2").start();
}
}
public class Main {
private static volatile boolean isStop = false;
public static void main(String[] args) {
Thread t1 = new Thread(() -> {
while (!Thread.currentThread().isInterrupted()) {
System.out.println("Is interrupted: " + Thread.currentThread().isInterrupted());
}
System.out.println("Is interrupted --> " + Thread.currentThread().isInterrupted());
}, "t1");
t1.start();
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
// Signal threads to stop
new Thread(() -> {
t1.interrupt();
}, "t2").start();
}
}
SHOW OUTPUT
isStop: false
isStop: true
isStop --> true
Atomic Boolean: false
Atomic Boolean: true
Atomic Boolean --> true
......
Is interrupted: false
Is interrupted: false
Is interrupted: false
Is interrupted --> true
The common methods for managing thread interruption in Java include
interrupt(),interrupted(), andisInterrupted():
interrupt(): An instance method that sets the thread’s interrupted status totrue. It does not forcibly stop the thread. If the thread is blocked (e.g., insleep(),wait(), orjoin()), it throws anInterruptedException.interrupted(): A static method that checks if the current thread has been interrupted. If it has, the method resets the interrupted status tofalse.isInterrupted(): An instance method that checks if the thread has been interrupted without resetting its interrupted status.
What’s the following code output?
SHOW CODE (Wrong Interruption)
import java.util.concurrent.TimeUnit;
public class Main {
public static void main(String[] args) {
Thread t1 = new Thread(() -> {
while (true) {
if (Thread.currentThread().isInterrupted()) {
System.out.println("isInterrupted --> true");
break;
}
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("INTERRUPTED...");
}
}, "t1");
t1.start();
try {
TimeUnit.SECONDS.sleep(3); // Sleep 3 seconds
} catch (InterruptedException e) {
e.printStackTrace();
}
new Thread(() -> {
t1.interrupt(); // Set interrupted status of thread 't1' to true
}, "t2").start();
}
}
SHOW OUTPUT (Wrong Interruption)
INTERRUPTED...
INTERRUPTED...
java.lang.InterruptedException: sleep interrupted
at java.base/java.lang.Thread.sleep(Native Method)
at dev.signalyu.warmup.Main.lambda$main$0(Main.java:14)
at java.base/java.lang.Thread.run(Thread.java:842)
INTERRUPTED...
INTERRUPTED...
......
When a thread is interrupted while blocked, the blocking method throws an
InterruptedExceptionand clears the interrupted status.
SHOW CODE (Correct Interruption)
import java.util.concurrent.TimeUnit;
public class Main {
public static void main(String[] args) {
Thread t1 = new Thread(() -> {
while (true) {
if (Thread.currentThread().isInterrupted()) {
System.out.println("isInterrupted --> true");
break;
}
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt(); // 🌟🌟🌟 Restore its interrupted status 🌟🌟🌟
e.printStackTrace();
}
System.out.println("INTERRUPTED...");
}
}, "t1");
t1.start();
try {
TimeUnit.SECONDS.sleep(3); // Sleep 3 seconds
} catch (InterruptedException e) {
e.printStackTrace();
}
new Thread(() -> {
t1.interrupt(); // Set interrupted status of thread 't1' to true
}, "t2").start();
}
}
SHOW OUTPUT (Correct Interruption)
INTERRUPTED...
INTERRUPTED...
java.lang.InterruptedException: sleep interrupted
at java.base/java.lang.Thread.sleep(Native Method)
at dev.signalyu.warmup.Main.lambda$main$0(Main.java:14)
at java.base/java.lang.Thread.run(Thread.java:842)
INTERRUPTED...
isInterrupted --> true
After restoring the thread’s interrupted status by adding
Thread.currentThread().interrupt();, it exits the while loop gracefully.
LockSupport
wait()/nofity() related problems
There are two common issues when using
wait()andnotify()in Java:
- Synchronization Requirement: Both
wait()andnotify()must be called within a synchronized block or method. This is because these methods rely on the object’s intrinsic lock (monitor). If they are called outside a synchronized context, the program will throw anIllegalMonitorStateException. - Missed Notifications: If
notify()is called beforewait(), the notification will be missed, and the thread callingwait()may block indefinitely. This happens becausenotify()does not have any effect if no thread is currently waiting on the object’s monitor.
SHOW CODE
public class Main {
private static final Object lock = new Object();
public static void main(String[] args) {
Thread t1 = new Thread(() -> {
// synchronized (lock) {
try {
System.out.println("Thread 1 is waiting...");
lock.wait(); // Thread 1 waits
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Thread 1 is resumed.");
// }
});
Thread t2 = new Thread(() -> {
// synchronized (lock) {
System.out.println("Thread 2 is notifying...");
lock.notify(); // Thread 2 notifies
// }
});
t1.start();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
t2.start();
}
}
public class Main {
private static final Object lock = new Object();
public static void main(String[] args) {
Thread t1 = new Thread(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// Acquire the lock after 1 second, notify() executes first
synchronized (lock) {
try {
System.out.println("Thread 1 is waiting...");
lock.wait(); // Thread 1 waits
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Thread 1 is resumed.");
}
});
Thread t2 = new Thread(() -> {
synchronized (lock) {
System.out.println("Thread 2 is notifying...");
lock.notify(); // Thread 2 notifies
}
});
t1.start();
t2.start();
}
}
SHOW OUTPUT
Thread 1 is waiting...
Exception in thread "Thread-0" java.lang.IllegalMonitorStateException: current thread is not owner
at java.base/java.lang.Object.wait(Native Method)
at java.base/java.lang.Object.wait(Object.java:338)
at dev.signalyu.warmup.Main.lambda$main$0(Main.java:11)
at java.base/java.lang.Thread.run(Thread.java:842)
Thread 2 is notifying...
Exception in thread "Thread-1" java.lang.IllegalMonitorStateException: current thread is not owner
at java.base/java.lang.Object.notify(Native Method)
at dev.signalyu.warmup.Main.lambda$main$1(Main.java:22)
at java.base/java.lang.Thread.run(Thread.java:842)
Thread 2 is notifying...
Thread 1 is waiting...
// WAIT INDEFINITELY
await()/signal() related problems
Similar to
wait()/notify(), when usingawait()/signal()from theConditioninterface, the thread must hold the associated lock. If the lock is not held, the program will throw anIllegalMonitorStateException. Additionally, ifsignal()is called beforeawait(), the notification will be missed, and the thread callingawait()may wait indefinitely. To avoid this, always use a loop to recheck the condition after waking up fromawait(). This ensures that thread only proceeds when the condition is truly met, even if a spurious wakeup occurs or notification is missed.
SHOW CODE
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public class Main {
private static final ReentrantLock lock = new ReentrantLock();
private static final Condition condition = lock.newCondition();
public static void main(String[] args) {
Thread t1 = new Thread(() -> {
// lock.lock();
try {
System.out.println("Thread 1 is waiting...");
condition.await(); // Thread 1 waits
System.out.println("Thread 1 is resumed.");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
// lock.unlock();
}
});
Thread t2 = new Thread(() -> {
// lock.lock();
try {
System.out.println("Thread 2 is signaling...");
condition.signal(); // Thread 2 signals
} finally {
// lock.unlock();
}
});
t1.start();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
t2.start();
}
}
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public class Main {
private static final ReentrantLock lock = new ReentrantLock();
private static final Condition condition = lock.newCondition();
public static void main(String[] args) {
Thread t1 = new Thread(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// Acquire the lock after 1 second, signal() executes first
lock.lock();
try {
System.out.println("Thread 1 is waiting...");
condition.await(); // Thread 1 waits
System.out.println("Thread 1 is resumed.");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
});
Thread t2 = new Thread(() -> {
lock.lock();
try {
System.out.println("Thread 2 is signaling...");
condition.signal(); // Thread 2 signals
} finally {
lock.unlock();
}
});
t1.start();
t2.start();
}
}
SHOW OUTPUT
Thread 1 is waiting...
Exception in thread "Thread-0" java.lang.IllegalMonitorStateException
at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.enableWait(AbstractQueuedSynchronizer.java:1516)
at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:1611)
at dev.signalyu.warmup.Main.lambda$main$0(Main.java:15)
at java.base/java.lang.Thread.run(Thread.java:842)
Thread 2 is signaling...
Exception in thread "Thread-1" java.lang.IllegalMonitorStateException
at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.signal(AbstractQueuedSynchronizer.java:1473)
at dev.signalyu.warmup.Main.lambda$main$1(Main.java:28)
at java.base/java.lang.Thread.run(Thread.java:842)
Thread 2 is signaling...
Thread 1 is waiting...
// WAIT INDEFINITELY
LockSupport: park()/unpark(Thread)
LockSupport is a utility class in Java that provides a low-level mechanism for thread synchronization. Unlike
synchronizedblocks or explicitLockobjects,LockSupportdoes not require explicit locking and unblocking. Instead, it operates using a permit mechanism, which makes it more flexible and lightweight. The most commonly used methods inLockSupportarepark()andunpark(Thread). Thepark()method blocks the current thread if the permit is unavailable, while theunpark(Thread)method makes the permit available for a specific thread, unblocking if it was parked.
The key points about
LockSupportinclude:
- The permit is a binary semaphore, meaning it can either be available or unavailable. If a permit is available when
park()is called, the thread continues execution without blocking. If no permit is available, the thread blocks until one is made available viaunpark(). - Unlike
synchronizedandLock,LockSupportdoes not suffer from spurious wakeups. This means that a thread will only unblock when explicitly unparked, providing more predictable behavior compared towait()/notify()orawait()/signal(). - The
unpark(Thread)method targets a specific thread, making it more precise thannotify()orsignal(), which wake up any waiting thread.
SHOW CODE
import java.util.concurrent.locks.LockSupport;
public class Main {
public static void main(String[] args) {
Thread t1 = new Thread(() -> {
System.out.println("Thread 1 is parking...");
LockSupport.park(); // Thread 1 parks
System.out.println("Thread 1 is unparked.");
});
Thread t2 = new Thread(() -> {
System.out.println("Thread 2 is unparking Thread 1...");
LockSupport.unpark(t1); // Thread 2 unparks Thread 1
});
t1.start();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
t2.start();
}
}
public class Main {
private static final Object lock = new Object();
public static void main(String[] args) {
Thread t1 = new Thread(() -> {
synchronized (lock) {
try {
System.out.println("Thread 1 is waiting...");
lock.wait(); // Thread 1 waits
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Thread 1 is resumed.");
}
});
Thread t2 = new Thread(() -> {
synchronized (lock) {
System.out.println("Thread 2 is notifying...");
lock.notify(); // Thread 2 notifies
}
});
t1.start();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
t2.start();
}
}
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public class Main {
private static final ReentrantLock lock = new ReentrantLock();
private static final Condition condition = lock.newCondition();
public static void main(String[] args) {
Thread t1 = new Thread(() -> {
lock.lock();
try {
System.out.println("Thread 1 is waiting...");
condition.await(); // Thread 1 waits
System.out.println("Thread 1 is resumed.");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
});
Thread t2 = new Thread(() -> {
lock.lock();
try {
System.out.println("Thread 2 is signaling...");
condition.signal(); // Thread 2 signals
} finally {
lock.unlock();
}
});
t1.start();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
t2.start();
}
}
SHOW OUTPUT
Thread 1 is parking...
Thread 2 is unparking Thread 1...
Thread 1 is unparked.
Thread 1 is waiting...
Thread 2 is notifying...
Thread 1 is resumed.
Thread 1 is waiting...
Thread 2 is signaling...
Thread 1 is resumed.
Java Memory Model
The Java Memory Model (JMM) is a specification that governs how threads interact with memory in a multi-threaded environment. It establishes rules that determine how and when change made by one thread to shared variables become visible to other threads, ensuring consistency and predictability in concurrent execution.
JMM classifies memory into two types: Main Memory and Working Memory. Main Memory is the central storage where all variables, including instance fields, static fields, and array elements, reside. Working Memory, on the other hand, is a thread’s local memory that holds copies of variables it accesses. Threads do not interact with main memory directly; instead, they load variables into their working memory and write back updates when necessary.
A fundamental concept in JMM is the happens-before relationship, which defines the order of operations between threads to maintain memory consistency. If one action happens-before another, the effects of the first action are guaranteed to be visible to the second. Examples include:
- A write to a
volatilevariable happens-before every subsequent read of that variable. - Unlocking a monitor happens-before the next lock on the same monitor.
The key characteristics of JMM include Atomicity, Visibility, and Ordering.
- Atomicity: Certain operations, such as reads and writes to
volatilevariables, are atomic, meaning they execute as indivisible units. - Visibility: Changes made by one thread to shared variables are guaranteed to be seen by other threads under proper synchronization mechanisms.
- Ordering: JMM defines strict rules for instruction ordering, but without synchronization, the JVM may reorder instructions for optimization, potentially causing unpredictable behavior in a multi-threaded environment.
happens-before
The happens-before relationship in the Java Memory Model (JMM) establishes rules that dictate the ordering and visibility of actions in multithreaded programs. It ensures that certain operations performed in one thread become predictably visible to another, preventing issues like instruction reordering and inconsistent data access. Below are the eight key priciples of the happens-before relationship:
- Program Order Rule: Within a single thread, actions occur in program order. For example, earlier statements happen-before later ones.
int a = 10; // Happens-before
int b = a + 5; // This sees the updated value of `a`
- Monitor Lock Rule: An unlock (
unlock()) on a monitor lock happens-before any subsequent lock (lock()) on the same monitor.
synchronized (lock) {
sharedVar = 10; // Happens-before the next lock
}
synchronized (lock) {
System.out.println(sharedVar); // Guaranteed to see updated value
}
- Volatile Variable Rule: A write to a
volatilevariable happens-before any subsequent read of that same variable.
private volatile boolean flag = false;
flag = true; // Happens-before the next read
System.out.println(flag); // Guaranteed to see true
- Thread Start Rule: A call to
Thread.start()on a new thread happens-before any actions in that thread.
Thread t = new Thread(() -> System.out.println("Running"));
t.start(); // Happens-before the thread executes the print statement
- Thread Termination Rule: All actions in a thread happen-before other threads detecting its termination. In other words, after a thread terminates, all its actions are visible to the thread that joins it.
Thread t = new Thread(() -> counter = 100);
t.start();
t.join(); // Ensures all writes in t happen-before main thread continues
System.out.println(counter); // Guaranteed to see 100
- A call to
Thread.interrupt()happens-before the interrupted thread detects the interruption viaisInterrupted()orInterruptedException.
Thread t = new Thread(() -> {
while (!Thread.currentThread().isInterrupted()) {
// Looping until interrupted
}
});
t.start();
t.interrupt(); // Happens-before the thread detects the interruption
- Final Field Rule: Writes to
finalfields in a constructor happen-before the object is seen by any other thread.
class Example {
final int x;
Example() {
x = 42; // Happens-before any other thread sees this object
}
}
- If action A happens-before action B, and action B happens-before action C, then action A happens-before action C.
volatile int a = 0;
a = 1; // A happens-before B (write to volatile)
int b = a; // B happens-before C (read from volatile)