Concurrency and Multithreading in Java

Synchronization Basics in Java

Process

A Process is a self-contained execution environment with its own memory space and resources. Two Processes do not share memory directly they communicate through inter-process communication mechanisms such as pipes, sockets, or shared memory. A Process can spawn multiple threads, each thread sharing the same memory space and resources of the parent process.

Understanding java Threads and CPU Cores

Core is a smallest unit of CPU that can independently execute instructions. When we say 8 cores CPU, it means the CPU contains 8 physical cores for executing instructions concurrently. If our program is multithreaded, this means its can simultaneously run 8 threads on the 8 cores of the CPU. But if we have more threads than cores, the OS will time-slice the CPU among the threads and context switches occur, where turns are taken by each thread to get CPU time.

So, when we say our program is concurrent it means that multiple threads of execution are making progress within the program, potentially running in parallel on multiple CPU cores.

int cores = Runtime.getRuntime().availableProcessors();
System.out.println("Available CPU cores: " + cores);

Sometime people use a formula to determine the optimal number of threads based on the number of available CPU cores. The formula often used is:
Optimal number of threads = Number of CPU cores * (1 + Wait time / Compute time)
Where:

Wait time is the time a thread spends waiting for I/O or other resources.

Compute time is the time a thread spends performing computations.

States of a Process and Thread

New: thread is created but not yet started.
Runnable: ready for CPU time but not yet running.
Running: currently executing.
Blocked/Waiting: waiting for a monitor lock or a specific condition.
Timed Waiting: waiting for a specified amount of time.
Terminated: has completed execution.

Inspecting Java Processes and Threads

Using Java diagnostic command-line tool jcmd we can inspect Java processes and their threads.

jcmd <pid> Thread.print

Example::

admin@220 ~ % jcmd 84488 Thread.print
84488:
"main" #1 prio=5 os_prio=0 tid=0x00007f8b58001000 nid=0x1e03 waiting on condition [0x00007f8b5a3ff000]
   java.lang.Thread.State: RUNNABLE

"Reference Handler" #2 prio=10 os_prio=0 tid=0x00007f8b58002000 nid=0x1e04 in Object.wait() [0x00007f8b5a4ff000]
   java.lang.Thread.State: WAITING (on object monitor)
...

Thread and Runnable

Java provides the Thread class and the Runnable interface to create and manage threads. A Thread represents a single thread of execution, while a Runnable is a task that can be executed by a thread. Threads can be extended or implemented using the Runnable interface to define their behavior.

Runnable implementation:

class RunnableImplementation implements Runnable {

    @Override
    public void run() {
        System.out.println(
                "Thread name ::  "
                        + Thread.currentThread().getName());
    }
}


public class Main {
    public static void main(String[] args) {
      Runnable task = new RunnableImplementation();
      Thread t1 = new Thread(task); 
      t1.start();
    }
}

Anonymous Runnable Implementation:

public class Main {
    public static void main(String[] args) {
        Runnable task = new Runnable() {
            @Override
            public void run() {
                System.out.println(
                        "Thread name ::  "
                                + Thread.currentThread().getName());
            }
        };
        Thread t1 = new Thread(task);
        t1.start();
    }
}

Extending Thread:

We can extend Thread class and override its run method. This approach allows us to define the thread's behavior directly within the subclass.

Example::

public class RunThreads extends Thread {
    public static void main(String[] args) {
        Thread t1 = new RunThreads();
        System.out.println(
            "Current thread name is "
            + Thread.currentThread().getName()
        );
        t1.start();
    }

    @Override
    public void run() {
        System.out.println(
                "Thread name ::  "
                        + Thread.currentThread().getName());
    }
}

Lambda expression:

public class Main {
    public static void main(String[] args) {
        Runnable task = () -> {
            System.out.println(
                    "Thread name ::  "
                            + Thread.currentThread().getName());
        };
        Thread t1 = new Thread(task);
        t1.start();
    }
}

Method reference:

public class Main {
    public static void main(String[] args) {
        Runnable task = Main::printThreadName;
        Thread t1 = new Thread(task);
        t1.start();
    }

    public static void printThreadName() {
        System.out.println(
                "Thread name ::  "
                        + Thread.currentThread().getName());
    }
}

NOTE: Threads provide run and start methods to execute Thread tasks. The run method contains the code that will be executed by the main thread, while the start method is used to begin the thread's execution in a new call stack.

t.run() → Executes the code on the current thread. t.start() → Starts a new thread, which then executes run().

Complete Example::

public class RunThreads extends Thread {

    public static void main(String[] args) {
        // Thread 1: Runnable implementation
        Runnable task = new RunnableImplementation();
        Thread t1 = new Thread(task);

        // Thread 2: Anonymous Runnable
        Thread t2 = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println(
                    "Thread name ::  "
                    + Thread.currentThread().getName()
                );
            }
        });

        // Thread 3: Extending Thread
        Thread t3 = new RunThreads();

        // Thread 4: Lambda expression
        Thread t4 = new Thread(() ->
            System.out.println(
                "Thread name ::  "
                + Thread.currentThread().getName()
            )
        );

        // Thread 5: Method reference
        Thread t5 = new Thread(RunThreads::execute);
        // if i dont give name here, the thread will have a default name like Thread-0, Thread-1, etc.

        System.out.println(
            "Current thread name is "
            + Thread.currentThread().getName()
        );

        t1.run(); // Executes on main thread

        t1.start();
        t2.start();
        t3.start();
        t4.start();
        t5.start();
    }

    public static void execute() {
        System.out.println(
            "Thread name ::  "
            + Thread.currentThread().getName()
        );
    }

    @Override
    public void run() {
        System.out.println(
                "Thread name ::  "
                        + Thread.currentThread().getName());
    }
}

class RunnableImplementation implements Runnable {

    @Override
    public void run() {
        System.out.println(
                "Thread name ::  "
                        + Thread.currentThread().getName());
    }
}

Interrupted Exception

Java provides the interrupt method to signal a thread to stop execution flow. When a thread is interrupted, it can either handle the interruption by catching the InterruptedException or by checking its interrupted status using Thread.currentThread().isInterrupted().

Example::

import java.util.concurrent.TimeUnit;

public class TestingInterrupt implements Runnable {

    @Override
    public void run() {
        try {
            System.out.println("inside run function");
            work();
            System.out.println("work functioncompleted");
        } catch (InterruptedException x) {
            System.out.println("work function was interrupted");
            return;
        }
        System.out.println("leaving run function");
    }

    public void work() throws InterruptedException {
        while (true) {
            System.out.println("working...");
            if (Thread.currentThread().isInterrupted()) {
                System.out.println("isInterrupted() = " + Thread.currentThread().isInterrupted());
                
                // this thread goes to sleep for a short time, which will throw InterruptedException if interrupted
                TimeUnit.MILLISECONDS.sleep(10);
                
                System.out.println("this line won't be executed");
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        TestingInterrupt runnableTask = new TestingInterrupt();
        Thread thread = new Thread(runnableTask);
        thread.start();

        // wait for 2 sec 
        Thread.sleep(2000);
        
        
        System.out.println("inside main function about to interrupting other thread");
        
        thread.interrupt();

        // wait for the other thread to finish before leaving main()
        thread.join();
        System.out.println("leaving main function");
    }
}

Code Execution flow:

main function starts and creates a thread with the TestingInterrupt runnable task.
- as thread.start() is called, thread class invokes overridden run method of TestingInterrupt class in the new thread.
- run method invokes the work method.
Within main method, Thread.sleep(2000) is called, which pauses the main thread for 2 seconds.
After 2 seconds, thread.interrupt() is called to signal the other thread to stop its execution.
After this, main thread waits for the other thread to finish by calling thread.join().
Inside Work method, Thread.currentThread().isInterrupted() is checked in each iteration of the loop.
- After main thread calls thread.interrupt(), the interrupted status of the other thread becomes true.
- The sleep will throw InterruptedException if the thread is interrupted, which is caught in the run method, within catch block.

Thread Scheduler, Race Condition, Daemon Threads and Thread Groups

If we have multiple threads running and CPU doesn't have enough cores to run all threads simultaneously, we might need to rely on the thread scheduler to determine which thread gets to run at any given time. Due to this, the order is not guaranteed and can lead to race conditions. But we have scheduling algorithms and locks to help manage thread execution and prevent race conditions.

Main scheduling algorithms :

Preemptive Priority-Based Scheduling
Time-Slicing (Round-Robin) Scheduling
First-Come, First-Served (FCFS) Scheduling

Preemptive Priority-Based Scheduling

In Java, each thread has a priority, which is an integer value between Thread.MIN_PRIORITY (1) and Thread.MAX_PRIORITY (10). We use Thread.setPriority(int priority) to set the priority of a thread and Thread.getPriority() to get the current priority of a thread. The scheduler uses these priorities to decide which thread should run next. Higher priority threads are more likely to be executed before lower priority threads, but this is not strictly guaranteed.

Thread thread1 = new Thread(() -> {
    System.out.println("Thread 1 running");
});
Thread thread2 = new Thread(() -> {
    System.out.println("Thread 2 running");
});

thread1.setPriority(Thread.MAX_PRIORITY); // or thread1.setPriority(10)
thread2.setPriority(Thread.MIN_PRIORITY);

thread1.start();
thread2.start();

Time-Slicing (Round-Robin) Scheduling and First-Come, First-Served (FCFS) Scheduling

Java doesn't provide method or API to control the exact scheduling of threads. The thread scheduler is part of the JVM and underlying operating system.

In Time-Slicing (Round-Robin) Scheduling, each thread gets a fixed time slice to execute, and the scheduler cycles through the threads. In First-Come, First-Served (FCFS) Scheduling, threads are executed in the order they arrive, without preemption.

we can't always predict the sequence of execution.

Thread.yield()

The Thread.yield() method is a hint to the thread scheduler that the current thread is willing to yield its current use of the CPU. The scheduler may choose to pause the current thread and allow other threads to execute. But this is not guaranteed, JVM and OS can ignore it.

public class TestingYield {

	public static void main(String[] args) {
		var t0 = new Thread(() -> {
			Thread.yield();
			System.out.println(Thread.currentThread().getName());
		});
		var t1 = new Thread(() -> System.out.println(Thread.currentThread().getName()));
		t0.start();
		t1.start();
	}
}

Daemon Threads

In Java, a daemon thread is a background thread which is detached from the main program flow and does not exit until all user threads have finished execution. Java provides the setDaemon(boolean on) method to mark a thread as a daemon.

Thread daemonThread = new Thread(() -> {
    while (true) {
        System.out.println("Daemon thread running");
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
});
daemonThread.setDaemon(true);
daemonThread.start();

Thread Groups

In Java, a thread group is a mechanism to group multiple threads together. This allows us to manage multiple threads as a single unit. We can create a thread group using the ThreadGroup class.

ThreadGroup group = new ThreadGroup("GroupName");

Thread thread1 = new Thread(group, () -> {
    System.out.println("Thread 1 running");
});
Thread thread2 = new Thread(group, () -> {
    System.out.println("Thread 2 running");
});

thread1.start();
thread2.start();


// to print details of the thread group
System.out.println("Active threads in group: " + group.activeCount());
System.out.println("Thread group name: " + group.getName());

We can also perform operations on the entire group, such as interrupting all threads:

group.interrupt();

Synchronization

Critical Sections A critical section is a group of instructions that should be executed concurrently only by one thread at a time to prevent race conditions.
Atomic Operations An atomic operation is an operation that is performed as a single unit without any interference from other threads, ensuring consistency and preventing race conditions. Java provides several classes and mechanisms to perform atomic operations, such as the AtomicInteger class and the synchronized keyword.

count++ is not atomic and can lead to race conditions when accessed by multiple threads concurrently.

Below is an Example: of using AtomicInteger to perform atomic operations:

import java.util.concurrent.atomic.AtomicInteger;
AtomicInteger count = new AtomicInteger(5);
count.incrementAndGet();

Monitors Monitor is a locking mechanism that allows threads to have both mutual exclusion and the ability to wait for certain conditions. In Java, every object can act as a monitor, and the synchronized keyword is used to acquire the monitor lock.
Mutexes A mutex (mutual exclusion) is a lock that ensures that only one thread can access a resource at a time.
Monitors vs Mutexes

Monitors are higher-level constructs that include mutexes and condition variables, while mutexes are lower-level locks used for mutual exclusion.

Synchronized method and block with the synchronized keyword

The synchronized keyword in Java is used to control access to critical sections, ensuring that only one thread can execute a block of code or method at a time.

// Synchronized block
synchronized (this) {
    // critical section code
}

// Synchronized method
public synchronized void synchronizedMethod() {
    // critical section code
}

Wait and Notify Mechanisms The wait() and notify() methods are used for inter-thread communication, allowing threads to wait for certain conditions and notify other threads when those conditions are met. These methods must be called within a synchronized context.

synchronized (this) {
    while (conditionNotMet) {
        wait();
    }
    // critical section code
    notify();
}

public class TestingWriterReader {
    static class MessageContext {
        private String message;
        private boolean available = false;

        // Writer writes a message
        public synchronized void write(String msg) {
            message = msg;
            available = true;

            System.out.println("Writer: Message written");
            notify(); // Wake up the waiting reader
        }

        // Reader reads the message
        public synchronized String read() throws InterruptedException {
            while (!available) {
                System.out.println("Reader: Waiting for message...");
                wait(); // Release lock and wait
            }

            available = false;
            return message;
        }
    }

    public static void main(String[] args) {
        MessageContext  contextObj = new MessageContext();

        Thread reader = new Thread(() -> {
            try {
                String msg = contextObj.read();
                System.out.println("Reader: Received: " + msg);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        Thread writer = new Thread(() -> {
            try {
                Thread.sleep(1000); // Simulate writing delay
                contextObj.write("Hello from Writer!");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        reader.start();
        writer.start();
    }
}

Output:

Reader: Waiting for message...
Writer: Message written
Reader: Received: Hello from Writer!

Concurrency in Java

Shared Memory

Java Memory is divided 2 main regions: heap and stack memory.

Heap memory is shared among all threads, this is used for objects with dynamic memory allocation.
Stack memory is thread-local and stores method call frames and local variables.

To safely share memory between threads, synchronization mechanisms such as synchronized blocks, volatile variables, and higher-level concurrency utilities from the java.util.concurrent package are used.

With these utilities, Java provides different ways to handle concurrency and shared memory safely.

ThreadLocal → Each thread gets its own copy of a value.

ThreadLocal<Integer> local = ThreadLocal.withInitial(() -> 0);

Runnable task = () -> {
    local.set(local.get() + 1);
    System.out.println(
        Thread.currentThread().getName() + ": " + local.get()
    );
};

new Thread(task, "A").start();
new Thread(task, "B").start();

Output:

A: 1
B: 1

InheritableThreadLocal → A child thread can inherit a value from its parent thread.

InheritableThreadLocal<Integer> inheritableThreadLocal = new InheritableThreadLocal<>();

Runnable inheritableTask = () -> {
    inheritableThreadLocal.set(inheritableThreadLocal.get() + 1);
    System.out.println(
        Thread.currentThread().getName() + ": " + inheritableThreadLocal.get()
    );
};

new Thread(inheritableTask, "Parent").start();
Thread.sleep(100); // Ensure parent thread sets the value before child starts
new Thread(inheritableTask, "Child").start();

Output:

Parent: 1
Child: 2

AtomicInteger → Multiple threads can safely update the same integer.

import java.util.concurrent.atomic.AtomicInteger;

AtomicInteger atomicInt = new AtomicInteger(0);

Runnable atomicTask = () -> {
    int newValue = atomicInt.incrementAndGet();
    System.out.println(
        Thread.currentThread().getName() + ": " + newValue
    );
};

new Thread(atomicTask, "Thread-1").start();
new Thread(atomicTask, "Thread-2").start();

Output:

Thread-1: 1
Thread-2: 2

Deadlock and Livelock

Deadlock is a state where two or more threads are blocked forever, each waiting for a resource held by the other.

Example: of deadlock:

import java.util.concurrent.TimeUnit;

public class TestingDeadlockIssue {

	private static Object resource1 = new Object();
	private static Object resource2 = new Object();
	
	public static void main(String[] args) {
		new Thread(() -> {
			synchronized (resource1) {
				System.out.println("resource1 is captured");
				try {
					TimeUnit.MILLISECONDS.sleep(500);
				} catch (InterruptedException e) {}
				
				synchronized (resource2) {
					System.out.println("This block will never be executed");
				}
			}
		}).start();
		
		new Thread(() -> {
			synchronized (resource2) {
				System.out.println("resource2 is captured");
				synchronized (resource1) {
					System.out.println("This block will never be executed");
				}
			}
		}).start();
	}
}

In this above Example:, the two threads are trying to acquire locks on resource1 and resource2 in different orders, leading to a deadlock situation. To resolve deadlock we can use techniques like Lock ordering, which is acquiring locks in a consistent global order to prevent circular wait conditions.

Example: of Lock ordering to prevent deadlock:

import java.util.concurrent.TimeUnit;

public class TestingDeadlockIssue {

	private static Object resource1 = new Object();
	private static Object resource2 = new Object();
	
	public static void main(String[] args) {
		new Thread(() -> {
			synchronized (resource1) {
				System.out.println("resource1 is captured");
				try {
					TimeUnit.MILLISECONDS.sleep(500);
				} catch (InterruptedException e) {}
				
				synchronized (resource2) {
					System.out.println("resource2 is captured");
				}
			}
		}).start();
		
		new Thread(() -> {
			synchronized (resource1) {
				System.out.println("resource1 is captured");
				synchronized (resource2) {
					System.out.println("resource2 is captured");
				}
			}
		}).start();
	}
}

Livelock is a situation where threads are not blocked but keep changing their state in response to each other, preventing progress.

Example: of livelock:

import java.util.concurrent.TimeUnit;

public class LivelockIssue {

    static class Worker {
        private final String name;
        private boolean active = true;

        Worker(String name) {
            this.name = name;
        }

        public boolean isActive() {
            return active;
        }

        public void work(Worker other) {
            while (active) {
                if (other.isActive()) {
                    System.out.println(
                        name + ": Other worker is active. Yielding..."
                    );

                    try {
                        TimeUnit.MILLISECONDS.sleep(100);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        return;
                    }

                    continue;
                }

                System.out.println(name + ": Doing work...");
                active = false;
            }
        }
    }

    public static void main(String[] args) {
        Worker workerA = new Worker("Worker A");
        Worker workerB = new Worker("Worker B");

        Thread t1 = new Thread(() -> workerA.work(workerB));
        Thread t2 = new Thread(() -> workerB.work(workerA));

        t1.start();
        t2.start();
    }
}

What happens in the above Example:: - Worker A checks if Worker B is active. - Worker B checks if Worker A is active. - Both workers yield and check again. - Neither changes its active state to false. - Both threads continue running indefinitely.

To fix this, we can use CountDownLatch for giving priority to the threads, CountDownLatch is a synchronization utility that allows one or more threads to wait until a specified number of operations or events are completed.

Example: using CountDownLatch:

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class TestingDeadlockIssue {

    static class Worker {
        private final String name;
        private final int priority;
        private volatile boolean active = true;

        Worker(String name, int priority) {
            this.name = name;
            this.priority = priority;
        }

        public boolean isActive() {
            return active;
        }

        public void work(Worker other, CountDownLatch start) {
            try {
                // due to this, both threads will wait until the start latch is counted down
                start.await();

                while (active) {
                    // change is this condition helps to deside which worker should wait based on priority
                    if (other.isActive() && priority < other.priority) {
                        System.out.println(
                            name + ": Waiting for " + other.name
                        );

                        TimeUnit.MILLISECONDS.sleep(100);
                        continue;
                    }

                    System.out.println(name + ": Doing work...");
                    active = false;
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    public static void main(String[] args) {
        Worker a = new Worker("Worker A", 2);
        Worker b = new Worker("Worker B", 1);

        CountDownLatch start = new CountDownLatch(1);

        Thread t1 = new Thread(() -> a.work(b, start));
        Thread t2 = new Thread(() -> b.work(a, start));

        t1.start();
        t2.start();
		
		// both threads are ready to start working
        start.countDown();
    }
}

Executor Services, Callable and Future

ExecutorService

ExecutorService is java's concurrency library method which allow us to manage threads lifecycle, without manual intervention. ExecutorService uses factory methods from Executors class, to create thread pools. thread pools is used to reduce the overhead of creating and destroying threads repeatedly.

Example: of creating a cached thread pool using ExecutorService:

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

public class TestinghExecutorServices {
	
	public static void main(String[] args) {
		ExecutorService es = Executors.newCachedThreadPool();
		es.execute(() -> System.out.println("thread T1: " + Thread.currentThread().getName()));
		es.execute(() -> System.out.println("thread T2: " + Thread.currentThread().getName()));
		
		es.shutdown();
	}

}

Ways to create a thread pool include:

Method	Description
`Executors.newCachedThreadPool()`	Creates a thread pool that creates new threads as needed, but will reuse previously constructed threads when available.
`Executors.newFixedThreadPool(int n)`	Creates a thread pool with a fixed number of threads.
`Executors.newSingleThreadExecutor()`	Creates a single-threaded executor that uses a single worker thread to execute tasks.
`Executors.newScheduledThreadPool(int n)`	Creates a thread pool that can schedule commands to execute periodically with or without a fixed delay.

execute() vs submit()
execute() is used to submit tasks for execution without expecting any result, execute() is Runnable only.
submit() is used to submit tasks for execution and returns a Future representing the pending result of the task, submit() can accept both Runnable and Callable.

Runnable task calls run() method and does not return a result, while Callable task calls call() method and returns a result.

Future represents the result of an asynchronous computation. It provides methods to check if the computation is complete, to wait for its completion, and to retrieve the result of the computation.

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class TestingFuture {

    public static void main(String[] args) throws Exception {

        ExecutorService executor =
                Executors.newSingleThreadExecutor();

        Future<Integer> future = executor.submit(() -> {
            System.out.println("Calculating...");
            Thread.sleep(2000);

            return 100 + 200;
        });

        System.out.println("Task submitted");

        // Wait for the result if it is not ready
        Integer result = future.get();

        System.out.println("Result: " + result);

        executor.shutdown();
    }
}

invokeAny() vs invokeAll()

invokeAll - executes multiple tasks and returns all their results. invokeAny - executes multiple tasks and returns the result of one that has completed successfully.

Example: of invokeAll()

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class TestingInvokeAll {
	
	public static void main(String[] args) throws InterruptedException {
		ExecutorService es = Executors.newCachedThreadPool();
		// callable tasks to be executed by the executor service
		List<Callable<String>> tasks = new ArrayList<>(Arrays.asList(
					() -> "task #1",
					() -> "task #2",
					() -> "task #3"
				));
		
		List<Future<String>> futures = es.invokeAll(tasks);
		// stream the futures and print their results
		// stream vs map -> stream allows processing elements in a functional style, map transforms each element.\
		// stream is part of the Java Stream API, which allows functional-style operations on collections of elements.
		// without stream, we would need to use a for loop to iterate over the futures and get their results.
		futures.stream().map(future -> {
			try {
				return future.get();
			} catch (InterruptedException | ExecutionException e) {
				e.printStackTrace();
			}
			return "";
		}).forEach(System.out::println);
		// :: is the method reference operator in Java, used to refer to methods without invoking them.
		// Shutdown the executor service
		es.shutdown();
	}
}

Example: of invokeAny()

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class TestingInvokeAny {

	public static void main(String[] args) throws InterruptedException, ExecutionException {
		ExecutorService es = Executors.newCachedThreadPool();
		List<Callable<String>> tasks = new ArrayList<>(Arrays.asList(
					() -> "task #1",
					() -> "task #2",
					() -> "task #3"
				));

		// invokeAny returns the result of one successfully completed task
		String result = es.invokeAny(tasks);
		System.out.println("Result: " + result);

		es.shutdown();
	}
}

awaitTermination

The awaitTermination method is used to block the current thread until either all tasks have completed execution after a shutdown request, or the timeout occurs, or the current thread is interrupted, whichever happens first.

Example: of awaitTermination()

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class TestingAwaitTermination {

	public static void main(String[] args) throws InterruptedException {
		ExecutorService es = Executors.newCachedThreadPool();
		es.submit(() -> {
			Thread.sleep(2000);
			System.out.println("Task completed");
			return null;
		});
		es.shutdown();
		if (es.awaitTermination(5, TimeUnit.SECONDS)) {
			System.out.println("All tasks completed");
		} else {
			System.out.println("Timeout occurred before termination");
		}
	}
}

More on Multithreading

Fork/Join

ForkJoinPool

ForkJoinPool is a thread pool that allows Java to break tasks into smaller subtasks and then combine them to produce a final result.

           +-------------+
           |  Root task  |
           +------+------+ 
                  | fork
        +---------+---------+
        v                   v
    +-------------+     +-------------+
    |  Subtask 1  |     |  Subtask 2  |
    +------+------+     +------+------+
        | join              | join
        +---------+---------+
                  v
           +-------------+
           |   Result    |
           +-------------+

import java.util.concurrent.*;

public class TestingForkJoin {
    static class SumTask extends RecursiveTask<Integer> {
        private final int start;
        private final int end;
        SumTask(int start, int end) {
            this.start = start;
            this.end = end;
        }
        @Override
        protected Integer compute() {
            // Base case: small enough to calculate directly
            if (end - start <= 2) {
                int sum = 0;

                for (int i = start; i <= end; i++) {
                    sum += i;
                }

                return sum;
            }

            // Divide the task into two subtasks
            int mid = (start + end) / 2;

            SumTask left = new SumTask(start, mid);
            SumTask right = new SumTask(mid + 1, end);

            // Execute subtasks in parallel
            left.fork();

            // Compute one subtask in the current worker
            int rightResult = right.compute();

            // Wait for the other subtask
            int leftResult = left.join();

            // Combine results
            return leftResult + rightResult;
        }
    }

    public static void main(String[] args) {
            // Create a ForkJoinPool with 4 threads
            ForkJoinPool pool = new ForkJoinPool(4);
        try {
            // Submit the SumTask to the pool and get the result
            int result = pool.invoke(new SumTask(1, 10));
            System.out.println("Sum = " + result);
        } finally {
            pool.shutdown();
        }
    }
}

RecursiveTask: RecursiveTask is a class in the Fork/Join framework that represents a task that returns a result. It is used when you want to break a task into smaller subtasks and then combine their results.
RecursiveAction: RecursiveAction is a class in the Fork/Join framework that represents a task that does not return a result. It is used when you want to break a task into smaller subtasks and execute them in parallel without combining results.

import java.util.concurrent.RecursiveAction;
import java.util.concurrent.RecursiveTask;

class TestingRecursiveTask extends RecursiveTask<Integer> {
    private final int n;
    TestingRecursiveTask(int n) {
        this.n = n;
    }
    @Override
    protected Integer compute() {
        if (n <= 1) return n;
        TestingRecursiveTask f1 = new TestingRecursiveTask(n - 1);
        TestingRecursiveTask f2 = new TestingRecursiveTask(n - 2);
        f1.fork();
        int result2 = f2.compute();
        int result1 = f1.join();
        return result1 + result2;
    }
}

class TestingRecursiveAction extends RecursiveAction {
    private final int n;
    TestingRecursiveAction(int n) {
        this.n = n;
    }
    @Override
    protected void compute() {
        if (n <= 1) return;
        TestingRecursiveAction f1 = new TestingRecursiveAction(n - 1);
        TestingRecursiveAction f2 = new TestingRecursiveAction(n - 2);
        f1.fork();
        f2.compute();
        f1.join();
    }
}

what is work-stealing ? Work-stealing is a scheduling strategy used by the Fork/Join framework where idle worker threads can "steal" tasks from the queues of busy threads to keep all threads busy and improve parallelism.

ForkJoinPool vs ExecutorService

Feature	ExecutorService	ForkJoinPool
Primary purpose	General task execution	Parallel divide-and-conquer tasks
Work distribution	Depends on executor implementation	Work-stealing
Typical tasks	Background jobs, API calls	Recursive computation, parallel algorithms
Main task interfaces	Runnable, Callable	RecursiveTask, RecursiveAction
Result handling	Future	ForkJoinTask, join()

ForkJoinPool extends AbstractExecutorService which extends ExecutorService, providing additional support for work-stealing and parallel divide-and-conquer tasks.

CompletableFuture ( was introduced in Java 8)

runAsync :runAsync is a method in CompletableFuture that allows you to run a task asynchronously without blocking the main thread. It takes a Runnable and returns a CompletableFuture<Void>.

Example:

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class TestingRunAsync {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		// runAsync using the default ForkJoinPool, it allow program to spawn asynchronous tasks without explicitly providing an ExecutorService
		// mean code will not block the main thread while the asynchronous task is running
		CompletableFuture.runAsync(() -> {
			System.out.println("Thread name: " + Thread.currentThread().getName());
			try {
				TimeUnit.SECONDS.sleep(10);
				System.out.println("Slept for 10 seconds for " + Thread.currentThread().getName());
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		});
		TimeUnit.SECONDS.sleep(2);
		ExecutorService es = Executors.newCachedThreadPool();
		// below code is not using the default ForkJoinPool, it uses the provided ExecutorService instead
		// this allows us to control the thread pool used for asynchronous execution
		CompletableFuture.runAsync(() -> {
			System.out.println("Thread name: " + Thread.currentThread().getName());
			try {
				TimeUnit.SECONDS.sleep(10);
				System.out.println("Slept for 10 seconds for " + Thread.currentThread().getName());
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}, es);
		es.shutdown();
	}
}

Output:

Thread name: ForkJoinPool.commonPool-worker-1
Thread name: pool-1-thread-1
Slept for 10 seconds for ForkJoinPool.commonPool-worker-1
Slept for 10 seconds for pool-1-thread-1

supplyAsync: supplyAsync is a method in CompletableFuture that allows you to run a task asynchronously and return a result. It takes a Supplier and returns a CompletableFuture containing the result of the computation.

Example:

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class TestingSupplyAsync {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		// supplyAsync using the default ForkJoinPool
		CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
			System.out.println("Thread name: " + Thread.currentThread().getName());
			try {
				TimeUnit.SECONDS.sleep(5);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			return "Result from default ForkJoinPool";
		});

		ExecutorService es = Executors.newCachedThreadPool();
		// supplyAsync using the provided ExecutorService
		CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
			System.out.println("Thread name: " + Thread.currentThread().getName());
			try {
				TimeUnit.SECONDS.sleep(5);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			return "Result from custom ExecutorService";
		}, es);

		System.out.println(future1.get());
		System.out.println(future2.get());
		es.shutdown();
	}
}

complete() and default values: complete() is a method in CompletableFuture that allows you to manually complete a CompletableFuture with a given value. If the CompletableFuture is already completed, this method has no effect. You can also provide a default value to be used if the computation has not completed yet.

CompletableFuture provide functional such as:

Executing asynchronous tasks
- Methods like supplyAsync(), complete() and runAsync() allow you to execute tasks asynchronously.
Transforming results of asynchronous computations (Callback methods)
- Methods like thenApply() and thenApplyAsync() allow you to transform the result of a completed future.
Combining multiple asynchronous computations (Dependent versus independent tasks)
- Methods like thenCombine(), thenCompose() and allOf() allow you to combine results from multiple futures.
Handling exceptions in asynchronous computations
- Methods like exceptionally() and handle() allow you to handle exceptions that occur during asynchronous computations.
Getting results
- Methods like get() and join() allow you to retrieve the result of a completed future, blocking if necessary until the computation is done.
Cancelling asynchronous computations
- Methods like cancel() allow you to cancel an ongoing computation if it has not yet completed.

Lock API

The Lock interface in Java provides more extensive locking operations than can be obtained using synchronized methods and statements and includes the ability to try to acquire the lock without blocking, and the ability to interrupt threads waiting to acquire a lock.

ReentrantLock

The ReentrantLock class provides reentrant mutual exclusion. A thread can acquire the lock multiple times without causing a deadlock. Fairness option is also available in ReentrantLock to ensure that the longest-waiting thread gets the lock first.

Example: usage:

import java.util.concurrent.locks.ReentrantLock;

public class ReentrantLockExample {
    private final ReentrantLock lock = new ReentrantLock();

    public void performTask() {
        lock.lock();
        try {
            // Critical section
            System.out.println("Thread " + Thread.currentThread().getName() + " is executing the task.");
        } finally {
            lock.unlock();
        }
    }
}

Comparing a ReentrantLock with a synchronized method

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.IntStream;


public class TestingReentrantLock {

	private static int counter;
	
	private Lock lock = new ReentrantLock();
	// this can also be private ReentrantLock instead of Lock

	public static void main(String[] args) throws InterruptedException {
		var thisInstance = new TestingReentrantLock();
		var es = Executors.newFixedThreadPool(4);
		IntStream.range(0, 10000).forEach((i) -> es.execute(thisInstance::incrementWithLock));
		terminateExecutorService(es);
		System.out.println(counter);
		counter = 0;
		var es1 = Executors.newFixedThreadPool(4);
		IntStream.range(0, 10000).forEach((i) -> es1.execute(thisInstance::incrementSynchronized));
		terminateExecutorService(es1);
		System.out.println(counter);
		counter = 0;
		var es2 = Executors.newFixedThreadPool(4);
		IntStream.range(0, 10000).forEach((i) -> es2.execute(thisInstance::increment));
		terminateExecutorService(es2);
		System.out.println(counter);
	}

	public void incrementWithLock() {
		try {
			lock.lock();
			counter++;
		} finally {
			lock.unlock();
		}
	}
	public void incrementSynchronized() {
		synchronized(this) {
			counter++;
		}
	}
	public void increment() {
		counter++;
	}
	private static void terminateExecutorService(ExecutorService es) throws InterruptedException {
	    // closing pool to accept no new tasks and finish existing ones
		es.shutdown();
		es.awaitTermination(3, TimeUnit.SECONDS);
		es.shutdownNow();
	}
}

Output of the program

$ javac TestingReentrantLock.java && java TestingReentrantLock
10000
10000
9906

lock.unlock() should always be called in a finally block to ensure that the lock is released even if an exception occurs within the critical section.

Reentrancy with nested methods

Example::

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class TestingReentrantLock {

	private Lock lock = new ReentrantLock();

	public static void main(String[] args) throws InterruptedException {
		var demo = new TestingReentrantLock();
		var es = Executors.newFixedThreadPool(4);
		
		es.execute(() -> System.out.println(demo.calculate("+", 2, 3)));
		es.execute(() -> System.out.println(demo.calculate("+", 7, 12)));
		es.execute(() -> System.out.println(demo.calculate("+", 87, 4)));
		
		es.execute(() -> System.out.println(demo.calculate("-", 2, 3)));
		es.execute(() -> System.out.println(demo.calculate("-", 57, 12)));
		es.execute(() -> System.out.println(demo.calculate("-", 110, 7)));
		
		terminateExecutorService(es);
		
	}

	public double add(double value1, double value2) {
		try {
			lock.lock();
			System.out.println("lock is acquired");
			
			return value1 + value2;
		} finally {
			lock.unlock();
			System.out.println("lock is released");
		}
	}

	public double subtract(double value1, double value2) {
		try {
			lock.lock();
			System.out.println("lock is acquired");
			
			return value1 - value2;
		} finally {
			lock.unlock();
			System.out.println("lock is released");
		}
	}

	public double calculate(String operation, double operand1, double operand2) {
		try {
			lock.lock();
			System.out.println("lock is acquired");

			switch (operation) {
			case "+":
				return add(operand1, operand2);
			case "-":
				return subtract(operand1, operand2);
			default:
				System.out.println("Calculate works only with + and - operators and two values only");
				return 0;
			}

		} finally {
			lock.unlock();
			System.out.println("lock is released");
		}
	}
	
	private static void terminateExecutorService(ExecutorService es) throws InterruptedException {
		es.shutdown();
		es.awaitTermination(3, TimeUnit.SECONDS);
		es.shutdownNow();
	}
}

In above Example: calculate method acquires the lock and then calls either add or subtract methods, which also attempt to acquire the same lock. Since ReentrantLock is reentrant, the same thread can acquire the lock multiple times without causing a deadlock. This demonstrates the reentrancy feature of ReentrantLock.

If a non-reentrant lock such as Semaphore were used, this code would result in a deadlock because the same thread would not be able to acquire the lock it already holds.

Fair vs unfair locks Fairness can reduce starvation risk, but it can also reduce throughput. It does not guarantee fair CPU scheduling, and untimed tryLock() can still barge even on a fair lock.
- Unfair lock: Waiting thread is not guaranteed to acquire the lock.
- Fair lock: Threads waiting to acquire the lock are generally served in arrival order.

Example::

ReentrantLock fairLock = new ReentrantLock(true); // fair lock
ReentrantLock unfairLock = new ReentrantLock(false); // unfair lock

Timed and interruptible locking

ReentrantLock provides methods to acquire the lock with a timeout and to respond to interrupts:

tryLock(long timeout, TimeUnit unit): Attempts to acquire the lock within the specified time.
lockInterruptibly(): Acquires the lock unless the current thread is interrupted.

Example: of tryLock with timeout:

import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class TestingReentrantLock {
	
	private static ReentrantLock lock = new ReentrantLock();
	
	public static void main(String[] args) {
		var es = Executors.newFixedThreadPool(2);
		
		es.submit(TestingReentrantLock::LockingTask);
		es.submit(TestingReentrantLock::LockingTask);
		
		es.shutdown();
	}

	private static void LockingTask() {
		try {
			System.out.println(Thread.currentThread().getName() + " is trying to acquire the lock");
			// Try to acquire the lock within 2 seconds
			// how this works is that tryLock will attempt to acquire the lock within the specified time
			// if it fails to acquire the lock within 2 seconds, it will return false
			if (lock.tryLock(2, TimeUnit.SECONDS)) {
				System.out.println("lock is acquired: " + lock.isLocked());
				TimeUnit.SECONDS.sleep(3);
			} else {
				System.out.println("Thread didn't acquired a lock: " + Thread.currentThread().getName());
				System.out.println("lock is currently held: " + lock.isLocked() );
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			if (lock.isHeldByCurrentThread()) {
				lock.unlock();
			}
		}
	}
}

Output:

$ javac TestingReentrantLock.java && java TestingReentrantLock
pool-1-thread-1 is trying to acquire the lock
pool-1-thread-2 is trying to acquire the lock
lock is acquired: true
Thread didn't acquired a lock: pool-1-thread-2
lock is currently held: true

Example: of lockInterruptibly():

import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class TestingReentrantLock {

	private static ReentrantLock lock = new ReentrantLock();

	public static void main(String[] args) {
		var es = Executors.newFixedThreadPool(2);

		es.submit(TestingReentrantLock::LockingTask);
		es.submit(TestingReentrantLock::LockingTask);

		es.shutdown();
	}

	private static void LockingTask() {
		try {
			System.out.println(Thread.currentThread().getName() + " is trying to acquire the lock");
			// Try to acquire the lock within 2 seconds
			// how this works is that tryLock will attempt to acquire the lock within the
			// specified time
			// if it fails to acquire the lock within 2 seconds, it will return false
			if (lock.tryLock(2, TimeUnit.SECONDS)) {
				System.out.println("lock is acquired: " + lock.isLocked() + " by thread: " + Thread.currentThread().getName());
				TimeUnit.SECONDS.sleep(3);
			} else {
				// trying lockInterruptibly
				System.out.println("Trying lockInterruptibly...");
				lock.lockInterruptibly();
				System.out.println("Thread acquired a lock: " + Thread.currentThread().getName());
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			if (lock.isHeldByCurrentThread()) {
				lock.unlock();
			}
		}
	}
}

Output:

pool-1-thread-1 is trying to acquire the lock
pool-1-thread-2 is trying to acquire the lock
lock is acquired: true by thread: pool-1-thread-1
Trying lockInterruptibly...
Thread acquired a lock: pool-1-thread-2

lockInterruptibly() does not automatically interrupt a thread. It makes waiting responsive to an interrupt request.

Condition

A Condition Interfaceallows threads to wait for a specific state change while coordinating through a Lock.

Important methods of the Condition interface include:

Method	Description
`await()`	Waits until signalled or interrupted.
`awaitUninterruptibly()`	Waits without terminating its wait because of interruption.
`await(long, TimeUnit)`	Waits for a limited time.
`awaitNanos(long)`	Waits for a specified duration in nanoseconds.
`awaitUntil(Date)`	Waits until a deadline.
`signal()`	Wakes one waiting thread.
`signalAll()`	Wakes all waiting threads.

import java.util.Deque;
import java.util.LinkedList;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class TestingCondition {
	public static void printStack(TestingCondition instance) {
		instance.lock.lock();
		try {
			System.out.println("Current stack: " + instance.stack);
		} finally {
			instance.lock.unlock();
		}
	}

	private static final int CAPACITY = 5;

	private Deque<String> stack = new LinkedList<>();
	private Lock lock = new ReentrantLock();
	private Condition stackEmptyCondition = lock.newCondition();
	private Condition stackFullCondition = lock.newCondition();

	public static void main(String[] args) {
		var TestingConditionInstance = new TestingCondition();
		var es = Executors.newFixedThreadPool(4);

		es.submit(() -> System.out.println(TestingConditionInstance.popFromStack()));
		es.submit(() -> TestingConditionInstance.pushToStack("Test String #1"));
		// print stack after pushing Test String #1
		printStack(TestingConditionInstance); 
		es.submit(() -> TestingConditionInstance.pushToStack("Test String #2"));
		es.submit(() -> TestingConditionInstance.pushToStack("Test String #3"));
		es.submit(() -> System.out.println(TestingConditionInstance.popFromStack()));
		es.submit(() -> System.out.println(TestingConditionInstance.popFromStack()));
		

		es.shutdown();
	}

	public void pushToStack(String item) {
		try {
			lock.lock();
			while (stack.size() == CAPACITY) {
				stackFullCondition.await();
			}
			stack.push(item);
			System.out.println("Pushed item: " + item);
			// Signal any waiting threads that the stack is no longer empty
			stackEmptyCondition.signalAll();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			lock.unlock();
		}
	}

	public String popFromStack() {
		try {
			lock.lock();
			while (stack.size() == 0) {
				// Wait until the stack is no longer empty
				stackEmptyCondition.await();
			}
			var item = stack.pop();
			System.out.println("Popped item: " + item);
			return item;
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			// Signal any waiting threads that the stack is no longer full
			stackFullCondition.signalAll();
			lock.unlock();
		}
		return "";
	}
}

Output:

$ javac TestingCondition.java && java TestingCondition
Current stack: []
Pushed item: Test String #1
Popped item: Test String #1
Test String #1
Pushed item: Test String #2
Pushed item: Test String #3
Popped item: Test String #3
Test String #3
Popped item: Test String #2
Test String #2

Critical rules for using conditions in Java:

lock should be acquired before calling await(), signal(), or signalAll().

await() releases the associated lock while waiting and reacquires it before returning.

Use a loop to recheck the condition after waking.

Signal the condition corresponding to the state change.

Always release the lock in finally.

ReentrantReadWriteLock

A read-write lock separates access into two modes:

Read lock: Multiple threads can read concurrently.
Write lock: Exclusive access for modifying shared state.

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class TestingReentrantReadWriteLock {

	private Map<String, String> syncHashMap = new HashMap<>();
    private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
    private Lock writeLock = readWriteLock.writeLock();
    private Lock readLock = readWriteLock.readLock();

    public void put(String key, String value) {
        try {
            writeLock.lock();
            syncHashMap.put(key, value);
        } finally {
            writeLock.unlock();
        }
    }

    public String remove(String key){
        try {
            writeLock.lock();
            return syncHashMap.remove(key);
        } finally {
            writeLock.unlock();
        }
    }

    public String get(String key){
        try {
            readLock.lock();
            return syncHashMap.get(key);
        } finally {
            readLock.unlock();
        }
    }

    public boolean containsKey(String key) {
        try {
            readLock.lock();
            return syncHashMap.containsKey(key);
        } finally {
            readLock.unlock();
        }
    }
    
    public static void main(String[] args) {
        var instance = new TestingReentrantReadWriteLock();
        var es = Executors.newFixedThreadPool(4);
        es.submit(() -> instance.put("key1", "value1"));
        es.submit(() -> System.out.println(instance.get("key1")));
        es.submit(() -> instance.put("key3", "value3"));
        es.submit(() -> instance.put("key2", "value2"));
        es.submit(() -> System.out.println(instance.get("key2")));
        es.submit(() -> System.out.println(instance.get("key3")));
        es.shutdown();
    }
}

Output:

value1
value2
value3

Synchronizers

Synchronizers are utilities that help coordinate the execution of multiple threads. They provide mechanisms to control the order and timing of thread execution, ensuring proper synchronization and avoiding race conditions. Common synchronizers classesin Java include CyclicBarrier, CountDownLatch, Semaphore, and Exchanger.

CyclicBarrier

CyclicBarrier is a class which implements synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point. It is useful in scenarios where you want multiple threads to wait until all threads have reached a certain point before proceeding.

Method	Description
`await()`	Causes the current thread to wait until all parties have reached the barrier.
`getParties()`	Returns the number of parties required to trip the barrier.
`getNumberWaiting()`	Returns the number of parties currently waiting at the barrier.
`reset()`	Resets the barrier to its initial state.

Example:

import java.util.List;
import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.Executors;
import java.util.stream.IntStream;

public class TestingCyclicBarrier {
	
	private static final int NUMBER_OF_THREADS = 4;
	private static final int NUMBERS_TO_GENERATE = 3;
	private static final int MAXIMUM_RANDOM_NUMBER = 100;
	
	private static CyclicBarrier barrier;
	private List<Integer> generatedNumbers = new CopyOnWriteArrayList<>();
	private Random r = new Random();
	
	public static void main(String[] args) {
		var  TestingCyclicBarrierInstance = new TestingCyclicBarrier();
		barrier = new CyclicBarrier(NUMBER_OF_THREADS, TestingCyclicBarrierInstance::numberGenerationCallback);
		
		var ex = Executors.newFixedThreadPool(NUMBER_OF_THREADS);
		IntStream.range(0, NUMBER_OF_THREADS).forEach(i -> ex.submit(TestingCyclicBarrierInstance::generateNumbers));
		System.out.println("getParties(): " + barrier.getParties());
		
		ex.shutdown();
		
	}
	
	private void numberGenerationCallback() {
		int max = generatedNumbers.stream()
						.mapToInt(i -> i)
						.max().getAsInt();
		System.out.println("All generated numbers: " + generatedNumbers);
		System.out.println("Maximum: " + max);
						
	}

	private void generateNumbers() {
		for (int i = 0; i < NUMBERS_TO_GENERATE; i++) {
			generatedNumbers.add(r.nextInt(MAXIMUM_RANDOM_NUMBER));
		}
		try {
			System.out.println("await(): " + barrier.await());
		} catch (InterruptedException | BrokenBarrierException e) {
			e.printStackTrace();
		}
	}
}

$ javac TestingCyclicBarrier.java && java TestingCyclicBarrier
All generated numbers: [17, 75, 85, 34, 84, 55, 13, 55, 87, 0, 39, 61]
Maximum: 87
getParties(): 4
await(): 0
await(): 3
await(): 1
await(): 2

CountDownLatch

countDownLatch is a class that allows one or more threads to wait until a set of operations being performed in other threads completes. It is initialized with a given count, and threads can call await() to wait until the count reaches zero. Other threads call countDown() to decrement the count.

Method	Description
countDown()	Decrements the count of the latch by one.
await()	Causes the current thread to wait until the latch has counted down to zero.
getCount()	Returns the current count of the latch.

Example:

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.stream.IntStream;
import java.util.concurrent.TimeUnit;

public class TestingCountDownLatch {
	private static final int COUNT = 4;
	public static void main(String[] args) throws InterruptedException {
		CountDownLatch countDownLatch = new CountDownLatch(COUNT);
		var es = Executors.newFixedThreadPool(2);
		IntStream.range(0, COUNT).forEach(i -> es.submit(new Worker(countDownLatch)));
		countDownLatch.await();
		System.out.println("Latch is released");
		es.shutdown();
	}
	private static class Worker implements Runnable {
		private CountDownLatch countDownLatch;
		public Worker(CountDownLatch countDownLatch) {
			this.countDownLatch = countDownLatch;
		}
		
		public void run() {
			System.out.println("doing some work...");
			System.out.println(countDownLatch.getCount());
			countDownLatch.countDown();
		}
	}
}

$ javac TestingCountDownLatch.java && java TestingCountDownLatch
doing some work...
4
doing some work...
3
doing some work...
2
doing some work...
1
Latch is released

Semaphore

semaphore is a class that uses permits to control access to a shared resource. Threads can acquire permits before accessing the resource and release them after use. It also initialized with a given number of permits.

Method	Description
acquire()	Acquires a permit, blocking if necessary until one is available.
release()	Releases a permit, returning it to the semaphore.
availablePermits()	Returns the current number of available permits.
tryAcquire()	Attempts to acquire a permit immediately, returning true if successful, false otherwise.

Example:

import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

public class TestingSemaphore {
	public static void main(String[] args) throws InterruptedException {
		int loginPermits = 4;
		var es = Executors.newFixedThreadPool(4);
		var TestingSemaphoreInstance = new TestingSemaphore(loginPermits);
		IntStream.range(0, loginPermits).forEach(user -> es.submit(TestingSemaphoreInstance::tryLogin));
		es.shutdown();
		es.awaitTermination(2, TimeUnit.SECONDS);
		// permits are not automatically released until logout is called
		System.out.println("availablePermitsForLogin:\t" + TestingSemaphoreInstance.availablePermitsForLogin());
		TestingSemaphoreInstance.tryLogin();
		
		TestingSemaphoreInstance.logout();
		System.out.println("availablePermitsForLogin:\t" + TestingSemaphoreInstance.availablePermitsForLogin());
	}

	private Semaphore semaphore;

	public TestingSemaphore(int slotLimit) {
		semaphore = new Semaphore(slotLimit);
	}

	private void tryLogin() {
		if (semaphore.tryAcquire()) {
			System.out.println("Congrats! You are logged in!");
		} else {
			System.out.println("No permits for login available. Try later.");
		}
	}

	private void logout() {
		semaphore.release();
	}

	private int availablePermitsForLogin() {
		return semaphore.availablePermits();
	}
}

$ javac TestingSemaphore.java && java TestingSemaphore
Congrats! You are logged in!
Congrats! You are logged in!
Congrats! You are logged in!
Congrats! You are logged in!
availablePermitsForLogin:	0
No permits for login available. Try later.
availablePermitsForLogin:	1

Exchanger

Exchanger is a class which allows to exchange data between two threads.

Example:

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Exchanger;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class TestingExchanger {
	
	public static void main(String[] args) throws InterruptedException {
		Exchanger<String> exchanger = new Exchanger<>();

	    Runnable task1 = () -> {
	        try {
	            String msg = exchanger.exchange("Message from Task #1");
	            System.out.println("Received from another thread in task #1: " + msg);
	        } catch (InterruptedException e) {
	            e.printStackTrace();
	        }
	    };

	    Runnable task2 = () -> {
	        try {
	            String msg = exchanger.exchange("Message from Task #2");
	            System.out.println("Received from another thread in task #2: " + msg);
	        } catch (InterruptedException e) {
	        	 e.printStackTrace();
	        }
	    };
	    var es = Executors.newCachedThreadPool();
	    es.submit(task1);
	    es.submit(task2);
	    es.shutdown();
	}
}

Output:

$ javac TestingExchanger.java && java TestingExchanger
Received from another thread in task #1: Message from Task #2
Received from another thread in task #2: Message from Task #1

Phaser

Phaser is a class which allows threads to wait for each other at a common barrier point, similar to CyclicBarrier, but more flexible.

In phaser we have arriveAndAwaitAdvance method which can be used by a thread to signal its arrival at the barrier and wait for other threads to arrive as well.

Example:

import java.util.concurrent.Executors;
import java.util.concurrent.Phaser;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

public class TestingPhaser {
	private static final int NUMBER_OF_THREADS = 2;
	private static Phaser phaser = new Phaser();
	public static void main(String[] args) throws InterruptedException {
		var es = Executors.newFixedThreadPool(NUMBER_OF_THREADS);
		System.out.println("Phase before tasks execution: " + phaser.getPhase());
		IntStream.range(0, NUMBER_OF_THREADS).forEach(i -> es.submit(new Task(phaser)));
		es.shutdown();
		
		es.awaitTermination(1, TimeUnit.SECONDS);
		System.out.println("Parties after all threads are de-registered: " 
				+ phaser.getRegisteredParties());
		
	}

	private static class Task implements Runnable {
		private Phaser phaser;
		public Task(Phaser phaser) {
			this.phaser = phaser;
			this.phaser.register();
		}

		@Override
		public void run() {
			System.out.println("Arrived in thread: " + Thread.currentThread().getName());
			// Arrive at the phaser and wait for other threads to reach this point
			System.out.println("Arrival phase number: " + phaser.arriveAndAwaitAdvance());
			try {
				TimeUnit.MILLISECONDS.sleep(100);
				System.out.println("getPhase(): " + phaser.getPhase());
				System.out.println("Arrived in thread: " + Thread.currentThread().getName());
				System.out.println("Arrival phase number: " + phaser.arriveAndAwaitAdvance());
				TimeUnit.MILLISECONDS.sleep(100);
				System.out.println("getPhase(): " + phaser.getPhase());
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			phaser.arriveAndDeregister();
		}
	}
}

Output:

$ javac TestingPhaser.java && java TestingPhaser
Phase before tasks execution: 0
Arrived in thread: pool-1-thread-2
Arrived in thread: pool-1-thread-1
Arrival phase number: 1
Arrival phase number: 1
getPhase(): 1
Arrived in thread: pool-1-thread-1
getPhase(): 1
Arrived in thread: pool-1-thread-2
Arrival phase number: 2
Arrival phase number: 2
getPhase(): 2
getPhase(): 2
Parties after all threads are de-registered: 0

Virtual Threads ( Java 21 )

A virtual thread is a lightweight thread that is managed by the JRE rather than the operating system. Thread created and managed by OS as known as platform threads they are heavier and have more overhead compared to virtual threads. Virtual threads allow for a large number of concurrent threads with minimal resource consumption, making them suitable for high-concurrency applications.

Example:

import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.IntStream;

public class TestingVirtualThread {

    public static void main(String[] args) throws InterruptedException {

        // create a virtual thread using Thread.startVirtualThread()
        Thread vThread = Thread.startVirtualThread(() -> {
            System.out.println("within virtual thread (Thread.startVirtualThread)");
        });

        // Wait for the virtual thread to finish
        vThread.join();

        // create a virtual thread using Executors.newVirtualThreadPerTaskExecutor()
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            // Submit a simple task
            Future<String> result = executor.submit(() -> {
                System.out.println("within virtual thread in ExecutorService!");
                return "Task completed";
            });

            // Wait and print result
            System.out.println("Result: " + result.get());
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        // run multiple tasks in parallel using virtual threads
        int numberOfTasks = 4;

        System.out.println("\nRunning " + numberOfTasks + " tasks in parallel using virtual threads...");
        long startTime = System.nanoTime();

        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
        	List<Future<?>> futures = IntStream.range(0, numberOfTasks)
        	        .<Future<?>>mapToObj(i -> executor.submit(() -> {
        	            System.out.println("Task #" + i + " is running on " + Thread.currentThread());
        	            // Simulate some work
        	            try {
        	                Thread.sleep(100);           // Simulate work
        	            } catch (InterruptedException e) {
        	                Thread.currentThread().interrupt();
        	            }
        	        }))
        	        .toList();
            
            for (Future<?> future : futures) {
                try {
                    future.get(); // block until task completes
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }

        long endTime = System.nanoTime();
        double virtualThreadsTime = (endTime - startTime) / 1_000_000.0;
		System.out.printf("Finished %d tasks in %.2f ms%n",
                numberOfTasks, virtualThreadsTime);
        
        startTime = System.nanoTime();
        int cpus = Runtime.getRuntime().availableProcessors();
        try (ExecutorService executor = Executors.newFixedThreadPool(cpus)) {
            List<Future<?>> futures = IntStream.range(0, numberOfTasks)
                    .<Future<?>>mapToObj(i -> executor.submit(() -> {
                        System.out.println("Task #" + i + " is running on " + Thread.currentThread());
                        // Simulate some work
                        try {
                            Thread.sleep(100);           
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                        }
                    }))
                    .toList();
            
            for (Future<?> future : futures) {
                try {
                    future.get(); // block until task completes
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
        endTime = System.nanoTime();
        double platformThreadsTime = (endTime - startTime) / 1_000_000.0;
		System.out.printf("Finished %d tasks in %.2f ms%n",
                numberOfTasks, platformThreadsTime);
		System.out.println("=== Result ===");
		System.out.println("Virtual Threads Time: " + virtualThreadsTime);
		System.out.println("Platform Threads Time: " + platformThreadsTime);

        Runtime runtime = Runtime.getRuntime();
        long memoryBefore = runtime.totalMemory() - runtime.freeMemory();
        System.out.println("Memory usage before tasks: " + memoryBefore);
        // Suggest running garbage collection before measuring memory after tasks
        System.gc();
        long memoryAfter = runtime.totalMemory() - runtime.freeMemory();
        System.out.println("Memory usage after tasks: " + memoryAfter);
    }
}

Output:

within virtual thread (Thread.startVirtualThread)
within virtual thread in ExecutorService!
Result: Task completed

Running 4 tasks in parallel using virtual threads...
Task #0 is running on VirtualThread[#18]/runnable@ForkJoinPool-1-worker-1
Task #1 is running on VirtualThread[#19]/runnable@ForkJoinPool-1-worker-2
Task #2 is running on VirtualThread[#20]/runnable@ForkJoinPool-1-worker-2
Task #3 is running on VirtualThread[#21]/runnable@ForkJoinPool-1-worker-2
Finished 4 tasks in 108.27 ms
Task #0 is running on Thread[#23,pool-1-thread-1,5,main]
Task #1 is running on Thread[#23,pool-1-thread-1,5,main]
Task #2 is running on Thread[#23,pool-1-thread-1,5,main]
Task #3 is running on Thread[#23,pool-1-thread-1,5,main]
Finished 4 tasks in 404.18 ms
=== Result ===
Virtual Threads Time: 108.270899
Platform Threads Time: 404.176887
Memory usage before tasks: 8838440
Memory usage after tasks: 4702336

Synchronization Basics in Java

Process

Understanding java Threads and CPU Cores

States of a Process and Thread

Inspecting Java Processes and Threads

Thread and Runnable

Interrupted Exception

Thread Scheduler, Race Condition, Daemon Threads and Thread Groups

Preemptive Priority-Based Scheduling

Time-Slicing (Round-Robin) Scheduling and First-Come, First-Served (FCFS) Scheduling

Thread.yield()

Daemon Threads

Thread Groups

Synchronization

Concurrency in Java

Shared Memory

Deadlock and Livelock

Executor Services, Callable and Future

More on Multithreading

Fork/Join

ForkJoinPool

ForkJoinPool vs ExecutorService

CompletableFuture ( was introduced in Java 8)

CompletableFuture provide functional such as:

Lock API

ReentrantLock

Condition

ReentrantReadWriteLock

Synchronizers

CyclicBarrier

CountDownLatch

Semaphore

Exchanger

Phaser

Virtual Threads ( Java 21 )

About the author

Comments

More from this blog

Getting Started with Clojure

Understanding VectorDB and Vector Embedding

The Bcrypt Algorithm for Secure Password Hashing

Getting started with RabbitMq using NodeJs