Concurrent Programming Patterns in Java

Concurrent programming patterns are essential for building efficient, scalable, and thread-safe applications. Here are the most important patterns with detailed examples.

1. Thread Pool Pattern

ExecutorService with Thread Pool

import java.util.concurrent.*; import java.util.*; public class ThreadPoolPattern { private final ExecutorService executor; public ThreadPoolPattern(int poolSize) { this.executor = Executors.newFixedThreadPool(poolSize); } public Future<String> submitTask(String taskName, int duration) { return executor.submit(() -> { System.out.println(Thread.currentThread().getName() + " starting task: " + taskName); Thread.sleep(duration); return "Task " + taskName + " completed in " + duration + "ms"; }); } public void executeParallelTasks() throws Exception { List<Callable<String>> tasks = Arrays.asList( () -> { Thread.sleep(1000); return "Task 1"; }, () -> { Thread.sleep(2000); return "Task 2"; }, () -> { Thread.sleep(1500); return "Task 3"; }, () -> { Thread.sleep(500); return "Task 4"; } ); // Execute all tasks and wait for completion List<Future<String>> futures = executor.invokeAll(tasks); for (Future<String> future : futures) { System.out.println("Result: " + future.get()); } } public void shutdown() { executor.shutdown(); try { if (!executor.awaitTermination(60, TimeUnit.SECONDS)) { executor.shutdownNow(); } } catch (InterruptedException e) { executor.shutdownNow(); Thread.currentThread().interrupt(); } } public static void main(String[] args) throws Exception { ThreadPoolPattern pool = new ThreadPoolPattern(4); // Submit individual tasks Future<String> future1 = pool.submitTask("Database Query", 2000); Future<String> future2 = pool.submitTask("File Processing", 1500); System.out.println("Task 1 result: " + future1.get()); System.out.println("Task 2 result: " + future2.get()); // Execute batch tasks pool.executeParallelTasks(); pool.shutdown(); } }

Custom Thread Pool with Monitoring

import java.util.concurrent.*; import java.util.concurrent.atomic.*; public class MonitoringThreadPool { private final ThreadPoolExecutor executor; private final AtomicInteger activeTasks = new AtomicInteger(0); private final AtomicLong totalTasks = new AtomicLong(0); public MonitoringThreadPool(int corePoolSize, int maxPoolSize, int queueSize) { this.executor = new ThreadPoolExecutor( corePoolSize, maxPoolSize, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(queueSize), new MonitoringThreadFactory(), new ThreadPoolExecutor.CallerRunsPolicy() ); } public Future<?> submitTask(Runnable task) { totalTasks.incrementAndGet(); return executor.submit(() -> { activeTasks.incrementAndGet(); try { task.run(); } finally { activeTasks.decrementAndGet(); } }); } public void printStats() { System.out.println("=== Thread Pool Stats ==="); System.out.println("Active Tasks: " + activeTasks.get()); System.out.println("Total Tasks: " + totalTasks.get()); System.out.println("Pool Size: " + executor.getPoolSize()); System.out.println("Active Count: " + executor.getActiveCount()); System.out.println("Completed Tasks: " + executor.getCompletedTaskCount()); System.out.println("Queue Size: " + executor.getQueue().size()); } public void shutdown() throws InterruptedException { executor.shutdown(); executor.awaitTermination(30, TimeUnit.SECONDS); } static class MonitoringThreadFactory implements ThreadFactory { private final AtomicInteger threadNumber = new AtomicInteger(1); private final String namePrefix = "monitored-pool-"; @Override public Thread newThread(Runnable r) { Thread t = new Thread(r, namePrefix + threadNumber.getAndIncrement()); t.setDaemon(true); t.setPriority(Thread.NORM_PRIORITY); return t; } } public static void main(String[] args) throws Exception { MonitoringThreadPool pool = new MonitoringThreadPool(2, 4, 10); // Submit multiple tasks for (int i = 0; i < 8; i++) { final int taskId = i; pool.submitTask(() -> { System.out.println("Executing task " + taskId + " on " + Thread.currentThread().getName()); try { Thread.sleep(2000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }); } // Monitor pool Thread.sleep(1000); pool.printStats(); pool.shutdown(); } }

2. Producer-Consumer Pattern

BlockingQueue Implementation

import java.util.concurrent.*; import java.util.*; public class ProducerConsumerPattern { private final BlockingQueue<String> queue; private final List<Producer> producers; private final List<Consumer> consumers; private volatile boolean running = true; public ProducerConsumerPattern(int queueSize, int producerCount, int consumerCount) { this.queue = new LinkedBlockingQueue<>(queueSize); this.producers = new ArrayList<>(); this.consumers = new ArrayList<>(); // Create producers for (int i = 0; i < producerCount; i++) { producers.add(new Producer("Producer-" + i, queue)); } // Create consumers for (int i = 0; i < consumerCount; i++) { consumers.add(new Consumer("Consumer-" + i, queue)); } } public void start() { producers.forEach(Thread::start); consumers.forEach(Thread::start); } public void stop() { running = false; producers.forEach(Producer::stop); consumers.forEach(Consumer::stop); // Interrupt all threads producers.forEach(Thread::interrupt); consumers.forEach(Thread::interrupt); } public int getQueueSize() { return queue.size(); } static class Producer extends Thread { private final BlockingQueue<String> queue; private volatile boolean producing = true; private int itemCount = 0; public Producer(String name, BlockingQueue<String> queue) { super(name); this.queue = queue; } @Override public void run() { while (producing && !Thread.currentThread().isInterrupted()) { try { String item = "Item-" + (++itemCount) + " from " + getName(); queue.put(item); // Blocks if queue is full System.out.println(getName() + " produced: " + item); Thread.sleep(100 + (int)(Math.random() * 400)); // Simulate work } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } } System.out.println(getName() + " stopped producing"); } public void stop() { producing = false; } } static class Consumer extends Thread { private final BlockingQueue<String> queue; private volatile boolean consuming = true; public Consumer(String name, BlockingQueue<String> queue) { super(name); this.queue = queue; } @Override public void run() { while (consuming && !Thread.currentThread().isInterrupted()) { try { String item = queue.take(); // Blocks if queue is empty System.out.println(getName() + " consumed: " + item); Thread.sleep(200 + (int)(Math.random() * 300)); // Simulate processing } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } } System.out.println(getName() + " stopped consuming"); } public void stop() { consuming = false; } } public static void main(String[] args) throws Exception { ProducerConsumerPattern pc = new ProducerConsumerPattern(10, 3, 2); pc.start(); // Let it run for 10 seconds Thread.sleep(10000); System.out.println("Final queue size: " + pc.getQueueSize()); pc.stop(); } }

Advanced Producer-Consumer with Poison Pill

import java.util.concurrent.*; import java.util.*; public class PoisonPillProducerConsumer { private final BlockingQueue<Message> queue; private final List<Worker> workers; private final String POISON_PILL = "POISON_PILL"; public PoisonPillProducerConsumer(int workerCount) { this.queue = new LinkedBlockingQueue<>(); this.workers = new ArrayList<>(); for (int i = 0; i < workerCount; i++) { workers.add(new Worker("Worker-" + i, queue, POISON_PILL)); } } public void startWorkers() { workers.forEach(Thread::start); } public void submitMessage(Message message) throws InterruptedException { queue.put(message); } public void shutdown() throws InterruptedException { // Send poison pill to each worker for (int i = 0; i < workers.size(); i++) { queue.put(new Message(POISON_PILL, null)); } // Wait for all workers to finish for (Worker worker : workers) { worker.join(); } } static class Message { private final String type; private final Object data; public Message(String type, Object data) { this.type = type; this.data = data; } public String getType() { return type; } public Object getData() { return data; } } static class Worker extends Thread { private final BlockingQueue<Message> queue; private final String poisonPill; public Worker(String name, BlockingQueue<Message> queue, String poisonPill) { super(name); this.queue = queue; this.poisonPill = poisonPill; } @Override public void run() { try { while (true) { Message message = queue.take(); if (poisonPill.equals(message.getType())) { System.out.println(getName() + " received poison pill. Shutting down."); break; } processMessage(message); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } private void processMessage(Message message) { System.out.println(getName() + " processing: " + message.getType()); // Simulate message processing try { Thread.sleep(100 + (int)(Math.random() * 200)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } public static void main(String[] args) throws Exception { PoisonPillProducerConsumer pc = new PoisonPillProducerConsumer(3); pc.startWorkers(); // Submit some messages for (int i = 0; i < 10; i++) { pc.submitMessage(new Message("TASK-" + i, "Data for task " + i)); } // Shutdown gracefully pc.shutdown(); System.out.println("All workers shut down gracefully"); } }

3. Future and CompletableFuture Patterns

CompletableFuture for Asynchronous Programming

import java.util.concurrent.*; import java.util.*; public class CompletableFuturePatterns { // Basic CompletableFuture usage public CompletableFuture<String> fetchUserDataAsync(String userId) { return CompletableFuture.supplyAsync(() -> { System.out.println("Fetching user data for: " + userId + " on " + Thread.currentThread().getName()); try { Thread.sleep(1000); // Simulate API call } catch (InterruptedException e) { throw new RuntimeException(e); } return "UserData-" + userId; }); } // Chaining CompletableFutures public CompletableFuture<String> processUserData(String userId) { return fetchUserDataAsync(userId) .thenApply(userData -> { System.out.println("Transforming user data: " + userData); return userData.toUpperCase(); }) .thenApply(transformedData -> { System.out.println("Enriching user data: " + transformedData); return transformedData + "-ENRICHED"; }); } // Combining multiple CompletableFutures public CompletableFuture<String> combineUserAndProfile(String userId) { CompletableFuture<String> userFuture = fetchUserDataAsync(userId); CompletableFuture<String> profileFuture = fetchUserProfileAsync(userId); return userFuture.thenCombine(profileFuture, (userData, profileData) -> { return "Combined: " + userData + " + " + profileData; }); } // Handling multiple CompletableFutures public CompletableFuture<Void> executeMultipleTasks(List<String> userIds) { List<CompletableFuture<String>> futures = userIds.stream() .map(this::fetchUserDataAsync) .toList(); return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])) .thenRun(() -> System.out.println("All user data fetched")); } // Exception handling in CompletableFuture public CompletableFuture<String> fetchWithErrorHandling(String userId) { return CompletableFuture.supplyAsync(() -> { if ("invalid".equals(userId)) { throw new IllegalArgumentException("Invalid user ID"); } return "Data for " + userId; }).exceptionally(throwable -> { System.err.println("Error fetching data: " + throwable.getMessage()); return "Default Data"; }); } // Timeout handling public CompletableFuture<String> fetchWithTimeout(String userId, long timeout, TimeUnit unit) { return fetchUserDataAsync(userId) .orTimeout(timeout, unit) .exceptionally(throwable -> { if (throwable instanceof TimeoutException) { return "Timeout fetching data for " + userId; } return "Error: " + throwable.getMessage(); }); } private CompletableFuture<String> fetchUserProfileAsync(String userId) { return CompletableFuture.supplyAsync(() -> { try { Thread.sleep(800); } catch (InterruptedException e) { throw new RuntimeException(e); } return "Profile-" + userId; }); } public static void main(String[] args) throws Exception { CompletableFuturePatterns patterns = new CompletableFuturePatterns(); // Basic usage CompletableFuture<String> future = patterns.fetchUserDataAsync("123"); System.out.println("Result: " + future.get()); // Chaining CompletableFuture<String> chainedFuture = patterns.processUserData("456"); System.out.println("Chained result: " + chainedFuture.get()); // Combining CompletableFuture<String> combinedFuture = patterns.combineUserAndProfile("789"); System.out.println("Combined result: " + combinedFuture.get()); // Multiple tasks List<String> userIds = Arrays.asList("1", "2", "3", "4"); patterns.executeMultipleTasks(userIds).get(); // Error handling CompletableFuture<String> errorFuture = patterns.fetchWithErrorHandling("invalid"); System.out.println("Error handling result: " + errorFuture.get()); // Timeout CompletableFuture<String> timeoutFuture = patterns.fetchWithTimeout("999", 500, TimeUnit.MILLISECONDS); System.out.println("Timeout result: " + timeoutFuture.get()); } }

4. Read-Write Lock Pattern

ReentrantReadWriteLock Implementation

import java.util.concurrent.locks.*; import java.util.*; public class ReadWriteLockPattern<K, V> { private final Map<K, V> cache = new HashMap<>(); private final ReadWriteLock lock = new ReentrantReadWriteLock(); private final Lock readLock = lock.readLock(); private final Lock writeLock = lock.writeLock(); public V get(K key) { readLock.lock(); try { System.out.println(Thread.currentThread().getName() + " reading key: " + key); Thread.sleep(100); // Simulate read operation return cache.get(key); } catch (InterruptedException e) { Thread.currentThread().interrupt(); return null; } finally { readLock.unlock(); } } public void put(K key, V value) { writeLock.lock(); try { System.out.println(Thread.currentThread().getName() + " writing key: " + key); Thread.sleep(500); // Simulate write operation cache.put(key, value); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { writeLock.unlock(); } } public V putIfAbsent(K key, V value) { writeLock.lock(); try { if (!cache.containsKey(key)) { cache.put(key, value); return value; } return cache.get(key); } finally { writeLock.unlock(); } } public boolean remove(K key) { writeLock.lock(); try { return cache.remove(key) != null; } finally { writeLock.unlock(); } } public int size() { readLock.lock(); try { return cache.size(); } finally { readLock.unlock(); } } public static void main(String[] args) throws Exception { ReadWriteLockPattern<String, String> cache = new ReadWriteLockPattern<>(); // Writer thread Thread writer = new Thread(() -> { for (int i = 0; i < 5; i++) { cache.put("key" + i, "value" + i); } }, "Writer-1"); // Reader threads List<Thread> readers = new ArrayList<>(); for (int i = 0; i < 3; i++) { Thread reader = new Thread(() -> { for (int j = 0; j < 5; j++) { cache.get("key" + j); } }, "Reader-" + i); readers.add(reader); } writer.start(); Thread.sleep(100); // Let writer start first readers.forEach(Thread::start); writer.join(); for (Thread reader : readers) { reader.join(); } System.out.println("Final cache size: " + cache.size()); } }

StampedLock for Optimistic Reading

import java.util.concurrent.locks.*; import java.util.*; public class StampedLockPattern<K, V> { private final Map<K, V> map = new HashMap<>(); private final StampedLock lock = new StampedLock(); public V get(K key) { // Try optimistic read first long stamp = lock.tryOptimisticRead(); V value = map.get(key); if (!lock.validate(stamp)) { // Optimistic read failed, acquire read lock stamp = lock.readLock(); try { value = map.get(key); } finally { lock.unlockRead(stamp); } } return value; } public void put(K key, V value) { long stamp = lock.writeLock(); try { map.put(key, value); } finally { lock.unlockWrite(stamp); } } public V readWithOptimisticLock(K key) { // Optimistic read pattern long stamp = lock.tryOptimisticRead(); V currentValue = map.get(key); // Check if the data was modified during read if (!lock.validate(stamp)) { // Data was modified, retry with read lock stamp = lock.readLock(); try { currentValue = map.get(key); } finally { lock.unlockRead(stamp); } } return currentValue; } public V computeIfAbsent(K key, java.util.function.Function<K, V> mappingFunction) { // Try optimistic read first long stamp = lock.readLock(); try { V value = map.get(key); if (value != null) { return value; } } finally { lock.unlockRead(stamp); } // Upgrade to write lock stamp = lock.writeLock(); try { // Double-check after acquiring write lock V value = map.get(key); if (value == null) { value = mappingFunction.apply(key); map.put(key, value); } return value; } finally { lock.unlockWrite(stamp); } } }

5. Actor Pattern with Akka-like Implementation

Simple Actor System

import java.util.concurrent.*; import java.util.*; public class ActorPattern { private final ExecutorService executor; private final Map<String, Actor> actors; public ActorPattern() { this.executor = Executors.newCachedThreadPool(); this.actors = new ConcurrentHashMap<>(); } public interface Actor { void onMessage(Object message); String getName(); } public void sendMessage(String actorName, Object message) { Actor actor = actors.get(actorName); if (actor != null) { executor.submit(() -> actor.onMessage(message)); } } public void registerActor(Actor actor) { actors.put(actor.getName(), actor); } public void shutdown() { executor.shutdown(); } // Example Actor implementations static class PrinterActor implements Actor { private final String name; public PrinterActor(String name) { this.name = name; } @Override public void onMessage(Object message) { System.out.println("[" + name + "] Received: " + message); } @Override public String getName() { return name; } } static class CounterActor implements Actor { private final String name; private int count = 0; public CounterActor(String name) { this.name = name; } @Override public void onMessage(Object message) { if (message instanceof Increment) { count++; System.out.println("[" + name + "] Count: " + count); } else if (message instanceof GetCount) { System.out.println("[" + name + "] Current count: " + count); } } @Override public String getName() { return name; } } static class Increment {} static class GetCount {} public static void main(String[] args) throws Exception { ActorPattern system = new ActorPattern(); // Register actors system.registerActor(new PrinterActor("printer")); system.registerActor(new CounterActor("counter")); // Send messages system.sendMessage("printer", "Hello World"); system.sendMessage("printer", "Another message"); system.sendMessage("counter", new Increment()); system.sendMessage("counter", new Increment()); system.sendMessage("counter", new GetCount()); Thread.sleep(1000); system.shutdown(); } }

6. Barrier and Latch Patterns

CyclicBarrier for Coordinated Processing

import java.util.concurrent.*; import java.util.*; public class BarrierPattern { private final CyclicBarrier barrier; private final ExecutorService executor; private final List<ComputationTask> tasks; public BarrierPattern(int numberOfTasks) { this.barrier = new CyclicBarrier(numberOfTasks, () -> { System.out.println("All tasks reached barrier. Merging results..."); mergeResults(); }); this.executor = Executors.newFixedThreadPool(numberOfTasks); this.tasks = new ArrayList<>(); for (int i = 0; i < numberOfTasks; i++) { tasks.add(new ComputationTask("Task-" + i, barrier)); } } public void startComputation() { tasks.forEach(executor::submit); } public void shutdown() throws InterruptedException { executor.shutdown(); executor.awaitTermination(10, TimeUnit.SECONDS); } private void mergeResults() { int total = tasks.stream().mapToInt(ComputationTask::getResult).sum(); System.out.println("Total result: " + total); } static class ComputationTask implements Runnable { private final String name; private final CyclicBarrier barrier; private int result; public ComputationTask(String name, CyclicBarrier barrier) { this.name = name; this.barrier = barrier; } @Override public void run() { try { // Phase 1: Compute partial result System.out.println(name + " starting computation..."); Thread.sleep(1000 + (int)(Math.random() * 2000)); result = (int)(Math.random() * 100); System.out.println(name + " computed: " + result); // Wait for all tasks to complete phase 1 barrier.await(); // Phase 2: Process merged result (if needed) System.out.println(name + " proceeding after barrier"); } catch (InterruptedException | BrokenBarrierException e) { Thread.currentThread().interrupt(); } } public int getResult() { return result; } } public static void main(String[] args) throws Exception { BarrierPattern pattern = new BarrierPattern(4); pattern.startComputation(); pattern.shutdown(); } }

CountDownLatch for One-time Coordination

import java.util.concurrent.*; import java.util.*; public class CountDownLatchPattern { public static void main(String[] args) throws Exception { int numberOfServices = 3; CountDownLatch latch = new CountDownLatch(numberOfServices); ExecutorService executor = Executors.newFixedThreadPool(numberOfServices); // Service initialization tasks for (int i = 0; i < numberOfServices; i++) { final int serviceId = i; executor.submit(() -> { try { System.out.println("Initializing Service " + serviceId); Thread.sleep(1000 + serviceId * 500); // Simulate initialization System.out.println("Service " + serviceId + " initialized"); latch.countDown(); // Signal completion } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }); } // Main thread waits for all services to initialize System.out.println("Main thread waiting for services to initialize..."); latch.await(); System.out.println("All services initialized. Starting main application..."); executor.shutdown(); } }

7. Thread-Local Pattern

ThreadLocal for Per-Thread Storage

import java.util.*; import java.util.concurrent.*; public class ThreadLocalPattern { // ThreadLocal for request context private static final ThreadLocal<RequestContext> requestContext = new ThreadLocal<>(); // ThreadLocal with initial value private static final ThreadLocal<SimpleDateFormat> dateFormat = ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")); // ThreadLocal for transaction context private static final ThreadLocal<Transaction> currentTransaction = new ThreadLocal<>(); static class RequestContext { private final String requestId; private final String userId; private final long timestamp; public RequestContext(String requestId, String userId) { this.requestId = requestId; this.userId = userId; this.timestamp = System.currentTimeMillis(); } // Getters public String getRequestId() { return requestId; } public String getUserId() { return userId; } public long getTimestamp() { return timestamp; } @Override public String toString() { return String.format("RequestContext{requestId='%s', userId='%s', timestamp=%d}", requestId, userId, timestamp); } } static class Transaction { private final String transactionId; private final List<String> operations = new ArrayList<>(); public Transaction(String transactionId) { this.transactionId = transactionId; } public void addOperation(String operation) { operations.add(operation); } public List<String> getOperations() { return new ArrayList<>(operations); } @Override public String toString() { return String.format("Transaction{id='%s', operations=%s}", transactionId, operations); } } public static void setRequestContext(RequestContext context) { requestContext.set(context); } public static RequestContext getRequestContext() { return requestContext.get(); } public static void clearRequestContext() { requestContext.remove(); } public static void startTransaction(String transactionId) { currentTransaction.set(new Transaction(transactionId)); } public static void addTransactionOperation(String operation) { Transaction transaction = currentTransaction.get(); if (transaction != null) { transaction.addOperation(operation); } } public static Transaction commitTransaction() { Transaction transaction = currentTransaction.get(); currentTransaction.remove(); return transaction; } public static void rollbackTransaction() { currentTransaction.remove(); } public static String formatDate(Date date) { return dateFormat.get().format(date); } // Example usage public static void main(String[] args) throws Exception { ExecutorService executor = Executors.newFixedThreadPool(3); // Simulate multiple requests for (int i = 0; i < 5; i++) { final int requestId = i; executor.submit(() -> { // Set request context for this thread setRequestContext(new RequestContext("REQ-" + requestId, "USER-" + requestId)); // Use thread-local date formatter System.out.println("[" + Thread.currentThread().getName() + "] " + formatDate(new Date()) + " - " + getRequestContext()); // Simulate transaction startTransaction("TX-" + requestId); addTransactionOperation("Operation 1"); addTransactionOperation("Operation 2"); Transaction tx = commitTransaction(); System.out.println("[" + Thread.currentThread().getName() + "] Committed: " + tx); // Clear thread-local clearRequestContext(); }); } executor.shutdown(); executor.awaitTermination(5, TimeUnit.SECONDS); } }

Key Concurrent Patterns Summary

Thread Pool - Reuse threads for better performance
Producer-Consumer - Decouple production and consumption
Future/CompletableFuture - Asynchronous programming
Read-Write Lock - Optimize read-heavy scenarios
Actor Model - Message-passing concurrency
Barrier/Latch - Coordinate multiple threads
Thread-Local - Per-thread data isolation

These patterns help build scalable, efficient, and maintainable concurrent applications while avoiding common pitfalls like race conditions and deadlocks.