Concurrency and Multithreading: Accelerating Backend Workflows in Java#

Processes#

• A process is an instance of a running program.
• It has its own memory (heap, stack, etc.).
• Inter-process communication is more expensive and complex.

Threads#

• Threads exist within a process.
• Threads share the same memory space.
• Thread creation is more lightweight compared to creating a new process.
• Communication between threads is simpler, but you must manage shared data to avoid conflicts.

Table comparing processes and threads:

Creating a Thread#

In Java, you can create a thread by:

1. Extending the Thread class.
2. Implementing the Runnable interface.
3. Implementing the Callable interface (which allows returning a value and throwing checked exceptions).

Below is a basic example using the Thread class:

1
public class SimpleThread extends Thread {
2
    @Override
3
    public void run() {
4
        System.out.println("Hello from Thread: " + Thread.currentThread().getName());
5
    }
6

7
    public static void main(String[] args) {
8
        SimpleThread thread = new SimpleThread();
9
        thread.start();
10
        System.out.println("Hello from main thread: " + Thread.currentThread().getName());
11
    }
12
}

Output might look like:

1
Hello from main thread: main
2
Hello from Thread: Thread-0

Notice that we override the run() method, but we call the start() method to execute it in a separate thread. If you call run() directly, it will execute in the current thread, defeating concurrency.

Using Runnable#

Creating a thread by implementing the Runnable interface is more flexible. Classes that implement Runnable are not tied to a specific thread. Here’s an example:

1
public class RunnableExample implements Runnable {
2
    @Override
3
    public void run() {
4
        System.out.println("Running in a separate thread: " + Thread.currentThread().getName());
5
    }
6

7
    public static void main(String[] args) {
8
        Thread thread = new Thread(new RunnableExample());
9
        thread.start();
10
    }
11
}

This approach is often preferred because it separates the task logic from the threading aspect.

Using Callable#

Callable<V> is similar to Runnable, but:

• It can return a value (V).
• It can throw checked exceptions.

Typically used with an ExecutorService:

1
import java.util.concurrent.*;
2

3
public class CallableExample implements Callable<String> {
4
    @Override
5
    public String call() {
6
        return "Result from Callable";
7
    }
8

9
    public static void main(String[] args) throws Exception {
10
        ExecutorService executor = Executors.newSingleThreadExecutor();
11
        Future<String> future = executor.submit(new CallableExample());
12
        String result = future.get();  // Blocks until the task is done
13
        System.out.println(result);
14
        executor.shutdown();
15
    }
16
}

join()#

The join() method makes one thread wait for another to complete before proceeding. This can be useful in scenarios where you need a final result from a secondary thread before continuing.

1
public class JoinExample {
2
    public static void main(String[] args) throws InterruptedException {
3
        Thread t = new Thread(() -> {
4
            try {
5
                Thread.sleep(2000);
6
                System.out.println("Thread finished.");
7
            } catch (InterruptedException e) {
8
                e.printStackTrace();
9
            }
10
        });
11
        t.start();
12
        t.join();  // Wait for t to finish
13
        System.out.println("Main thread resuming.");
14
    }
15
}

sleep()#

Thread.sleep() pauses the current thread for a specified time. Use it judiciously for simulation or pacing, but be mindful it doesn’t release locks.

1
Thread.sleep(1000); // Sleeps for 1 second

interrupt()#

Threads can interrupt each other, allowing for cooperative cancellation or signaling. Always check Thread.interrupted() or handle InterruptedException for graceful shutdown.

The synchronized Keyword#

Placing synchronized on a method or block ensures mutual exclusion:

1
public class Counter {
2
    private int count = 0;
3

4
    public synchronized void increment() {
5
        count++;
6
    }
7

8
    public synchronized int getCount() {
9
        return count;
10
    }
11
}

Mutual exclusion can also be done over specific critical regions:

1
public void increment() {
2
    synchronized (this) {
3
        count++;
4
    }
5
}

Lock Objects#

The java.util.concurrent.locks package introduces Lock and ReentrantLock for more granular control, such as being able to interrupt a lock acquisition or attempt to acquire it without blocking indefinitely.

1
private final Lock lock = new ReentrantLock();
2

3
public void increment() {
4
    lock.lock();
5
    try {
6
        count++;
7
    } finally {
8
        lock.unlock();
9
    }
10
}

Volatile Keyword#

volatile ensures that a variable’s updates are immediately visible to other threads. This is different from synchronized, which ensures mutual exclusion. volatile does not protect compound actions (like incrementing an integer).

1
private volatile boolean running = true;

Why Thread Pools?#

Creating a new thread for each task can be expensive under heavy load. Thread pools:

• Reuse a fixed number of threads.
• Manage task submission and scheduling.
• Prevent resource exhaustion.

Executors#

Executors is a factory class that provides several methods to create executor services:

• newFixedThreadPool(int nThreads): A pool of exactly nThreads.
• newCachedThreadPool(): Grows or shrinks threads as needed.
• newSingleThreadExecutor(): A single-threaded executor to submit tasks sequentially.

Example:

1
import java.util.concurrent.*;
2

3
public class ThreadPoolExample {
4
    public static void main(String[] args) {
5
        ExecutorService executor = Executors.newFixedThreadPool(4);
6

7
        for (int i = 0; i < 10; i++) {
8
            executor.submit(() -> {
9
                System.out.println("Task handled by " + Thread.currentThread().getName());
10
            });
11
        }
12

13
        executor.shutdown();
14
    }
15
}

Configuring Thread Pools#

Manually creating ThreadPoolExecutor allows more granular control:

• Core pool size, maximum pool size, keep-alive time, queue type, rejection policy.

1
ThreadPoolExecutor executor = new ThreadPoolExecutor(
2
    2,
3
    4,
4
    60L, TimeUnit.SECONDS,
5
    new LinkedBlockingQueue<>(100),
6
    Executors.defaultThreadFactory(),
7
    new ThreadPoolExecutor.AbortPolicy()
8
);

Concurrent Collections#

• ConcurrentHashMap
• CopyOnWriteArrayList
• BlockingQueue implementations like LinkedBlockingQueue, ArrayBlockingQueue

These classes are optimized for concurrent usage, reducing or removing the need for explicit synchronization.

BlockingQueue#

A BlockingQueue is thread-safe and blocks on put() if the queue is full, or on take() if the queue is empty. This is crucial in producer-consumer patterns.

1
BlockingQueue<String> queue = new LinkedBlockingQueue<>(10);
2

3
Thread producer = new Thread(() -> {
4
    try {
5
        queue.put("Item");
6
    } catch (InterruptedException e) {
7
        Thread.currentThread().interrupt();
8
    }
9
});
10

11
Thread consumer = new Thread(() -> {
12
    try {
13
        String item = queue.take();
14
        System.out.println(item);
15
    } catch (InterruptedException e) {
16
        Thread.currentThread().interrupt();
17
    }
18
});

CountdownLatch#

Allows one or more threads to wait until a set of tasks completes. You initialize CountDownLatch with a count, and each task calls countDown(). Threads can await() until the count reaches zero.

1
CountDownLatch latch = new CountDownLatch(3);
2
for (int i = 0; i < 3; i++) {
3
    new Thread(() -> {
4
        // Do some work
5
        latch.countDown();
6
    }).start();
7
}
8
// Wait for all tasks
9
latch.await();
10
System.out.println("All tasks finished!");

CyclicBarrier#

Similar to CountDownLatch but reusable. It makes threads wait for each other at a common barrier point.

Phaser#

An even more flexible synchronization barrier, allowing dynamic registration of parties (threads) and multiple phases of synchronization.

Identifying Tasks#

Partition your work into tasks that can be performed independently. For instance:

• Handling HTTP requests in parallel.
• Breaking large computations into sub-tasks.

Avoiding Shared State#

Reduce or eliminate mutable shared data. Use Immutable objects or local copies. Many concurrency problems vanish when data is not shared.

Choosing the Right Concurrency Model#

• Thread per Task: Simple but can be inefficient under high loads.
• Thread Pool: Reuse a fixed or dynamic set of threads.
• Event-Driven/Asynchronous: Use non-blocking I/O and callbacks or futures.
• Reactive Programming: Employ frameworks like Project Reactor or RxJava for managing asynchronous data streams.

Handling Failures#

Threads might throw exceptions, or a rejection policy might be triggered in a thread pool. Have a plan for:

• Logging or reporting errors.
• Retrying tasks as appropriate.
• Graceful shutdown.

Atomics#

java.util.concurrent.atomic provides classes like AtomicInteger, AtomicLong, and AtomicReference which enable lock-free, thread-safe operations. For small updates, these can outperform synchronized blocks.

Example using AtomicInteger:

1
private AtomicInteger count = new AtomicInteger();
2

3
public void increment() {
4
    count.incrementAndGet();
5
}

ThreadLocal#

ThreadLocal<T> provides each thread its own copy of a variable. This is useful for contexts like request-scoped data in a web application.

1
private static ThreadLocal<SimpleDateFormat> sdf = ThreadLocal.withInitial(
2
    () -> new SimpleDateFormat("yyyy-MM-dd")
3
);
4

5
public static String formatDate(Date date) {
6
    return sdf.get().format(date);
7
}

Java’s Reactive Streams#

For highly asynchronous and non-blocking I/O, Java has a Flow API (in java.util.concurrent package). Libraries like Project Reactor, RxJava, and Akka Streams implement the Reactive Streams specification.

Virtual Threads (Project Loom)#

Project Loom (in preview for newer Java versions) introduces “virtual threads,” making it possible to have thousands (or millions) of threads with minimal overhead. This can simplify asynchronous code by letting each operation use a lightweight thread without performance penalties typically associated with thread creation and context switching.

Asynchronous File Processing#

Imagine you have to process multiple files (like log files) concurrently and store results in a database.

Pseudo-flow:

1. Main thread scans a directory of files.
2. For each file, submit a task to a thread pool to parse it.
3. Store the relevant data in the database.
4. Use a CountDownLatch to wait for all parsing tasks to complete.

Code snippet outline:

1
public class FileProcessor {
2
    public static void main(String[] args) throws InterruptedException {
3
        File folder = new File("/path/to/logs");
4
        File[] files = folder.listFiles();
5

6
        ExecutorService executor = Executors.newFixedThreadPool(5);
7
        CountDownLatch latch = new CountDownLatch(files.length);
8

9
        for (File file : files) {
10
            executor.submit(() -> {
11
                processFile(file);
12
                latch.countDown();
13
            });
14
        }
15

16
        latch.await();  // wait for all tasks
17
        System.out.println("All files processed!");
18

19
        executor.shutdown();
20
    }
21

22
    private static void processFile(File file) {
23
        // read file, parse content, store results
24
    }
25
}

Producer-Consumer System#

Imagine a system where one thread produces events (messages) and another consumes them:

1
BlockingQueue<String> queue = new LinkedBlockingQueue<>(10);
2

3
Runnable producer = () -> {
4
    try {
5
        while (!Thread.currentThread().isInterrupted()) {
6
            String msg = "Event-" + System.currentTimeMillis();
7
            queue.put(msg);
8
            System.out.println("Produced: " + msg);
9
            Thread.sleep(100);
10
        }
11
    } catch (InterruptedException e) {
12
        Thread.currentThread().interrupt();
13
    }
14
};
15

16
Runnable consumer = () -> {
17
    try {
18
        while (!Thread.currentThread().isInterrupted()) {
19
            String msg = queue.take();
20
            System.out.println("Consumed: " + msg);
21
        }
22
    } catch (InterruptedException e) {
23
        Thread.currentThread().interrupt();
24
    }
25
};
26

27
ExecutorService executor = Executors.newFixedThreadPool(2);
28
executor.submit(producer);
29
executor.submit(consumer);
30

31
// Shutdown logic ...

Context Switching#

Each time the CPU swaps between threads, a context switch occurs. Excessive context switching can degrade performance. Balancing the number of threads with CPU cores is vital.

Lock Contention#

Overuse of shared resources or overly broad synchronization blocks can cause multiple threads to queue, waiting for locks. Use lock-free structures or reduce lock granularity to mitigate this.

Scalability#

As the task count grows, ensure your concurrency design can scale:

• Monitoring thread pools’ queue sizes.
• Load testing.
• Profiling locks and hotspots.

Benchmarks#

Always measure. Tools like JMH (Java Microbenchmark Harness) help you write micro-benchmarks to measure concurrency performance accurately.