Java Concurrency Utilities (java.util.concurrent) provide robust, high-performance threading utilities that go beyond basic synchronized blocks and Thread classes.

1. Executor Framework

Thread Pools and Executors

import java.util.concurrent.*;
public class ExecutorFrameworkExample {
public static void main(String[] args) {
demonstrateExecutors();
demonstrateScheduledExecutor();
demonstrateForkJoinPool();
}
public static void demonstrateExecutors() {
System.out.println("=== Executor Framework Examples ===");
// 1. Fixed Thread Pool
ExecutorService fixedPool = Executors.newFixedThreadPool(4);
// 2. Cached Thread Pool
ExecutorService cachedPool = Executors.newCachedThreadPool();
// 3. Single Thread Executor
ExecutorService singleThread = Executors.newSingleThreadExecutor();
// 4. Work Stealing Pool (ForkJoinPool)
ExecutorService workStealingPool = Executors.newWorkStealingPool();
// Submit tasks
for (int i = 1; i <= 10; i++) {
final int taskId = i;
fixedPool.submit(() -> {
System.out.println("Task " + taskId + " executed by " + 
Thread.currentThread().getName());
try {
Thread.sleep(1000); // Simulate work
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
}
// Proper shutdown
fixedPool.shutdown();
try {
if (!fixedPool.awaitTermination(60, TimeUnit.SECONDS)) {
fixedPool.shutdownNow();
}
} catch (InterruptedException e) {
fixedPool.shutdownNow();
Thread.currentThread().interrupt();
}
}
public static void demonstrateScheduledExecutor() {
System.out.println("\n=== Scheduled Executor Examples ===");
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
// Schedule a task to run after 2 seconds
ScheduledFuture<?> future = scheduler.schedule(() -> {
System.out.println("Scheduled task executed at: " + System.currentTimeMillis());
}, 2, TimeUnit.SECONDS);
// Schedule a task to run repeatedly every 1 second after initial 1 second delay
scheduler.scheduleAtFixedRate(() -> {
System.out.println("Fixed rate task executed at: " + System.currentTimeMillis());
}, 1, 1, TimeUnit.SECONDS);
// Schedule with fixed delay between executions
scheduler.scheduleWithFixedDelay(() -> {
try {
Thread.sleep(500); // Simulate task execution time
System.out.println("Fixed delay task executed at: " + System.currentTimeMillis());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}, 1, 2, TimeUnit.SECONDS);
// Let it run for a while then shutdown
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
scheduler.shutdown();
}
}

Custom ThreadPoolExecutor

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
public class CustomThreadPoolExample {
public static class CustomThreadPoolExecutor extends ThreadPoolExecutor {
private final AtomicLong taskCount = new AtomicLong(0);
private final ThreadLocal<Long> startTime = new ThreadLocal<>();
public CustomThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
long keepAliveTime, TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
}
@Override
protected void beforeExecute(Thread t, Runnable r) {
super.beforeExecute(t, r);
startTime.set(System.currentTimeMillis());
System.out.println("Task starting in thread: " + t.getName());
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
super.afterExecute(r, t);
long duration = System.currentTimeMillis() - startTime.get();
startTime.remove();
long count = taskCount.incrementAndGet();
System.out.println("Task completed in " + duration + "ms. Total tasks: " + count);
if (t != null) {
System.err.println("Task failed with exception: " + t.getMessage());
}
}
@Override
protected void terminated() {
super.terminated();
System.out.println("ThreadPool terminated. Total tasks executed: " + taskCount.get());
}
}
public static void main(String[] args) {
CustomThreadPoolExecutor executor = new CustomThreadPoolExecutor(
2, 4, 30, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(10)
);
// Rejected execution handler
executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
for (int i = 0; i < 15; i++) {
final int taskId = i;
executor.execute(() -> {
System.out.println("Executing task " + taskId + " in " + 
Thread.currentThread().getName());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
}
executor.shutdown();
}
}

2. Callable and Future

import java.util.concurrent.*;
import java.util.*;
public class CallableFutureExample {
public static void main(String[] args) throws Exception {
demonstrateCallableFuture();
demonstrateCompletableFuture();
}
public static void demonstrateCallableFuture() throws Exception {
System.out.println("=== Callable and Future Examples ===");
ExecutorService executor = Executors.newFixedThreadPool(3);
// Callable that returns a result
Callable<String> task = () -> {
Thread.sleep(1000);
return "Task completed at " + System.currentTimeMillis();
};
// Submit Callable and get Future
Future<String> future = executor.submit(task);
// Do other work while task is executing
System.out.println("Doing other work...");
// Get result (blocks if not ready)
try {
String result = future.get(2, TimeUnit.SECONDS);
System.out.println("Result: " + result);
} catch (TimeoutException e) {
System.err.println("Task timed out");
future.cancel(true);
}
// Multiple tasks example
List<Callable<Integer>> tasks = Arrays.asList(
() -> { Thread.sleep(1000); return 1; },
() -> { Thread.sleep(2000); return 2; },
() -> { Thread.sleep(1500); return 3; }
);
// Invoke all tasks
List<Future<Integer>> futures = executor.invokeAll(tasks);
for (Future<Integer> f : futures) {
if (!f.isCancelled()) {
System.out.println("Task result: " + f.get());
}
}
// Invoke any task
Integer firstResult = executor.invokeAny(tasks);
System.out.println("First completed task result: " + firstResult);
executor.shutdown();
}
public static void demonstrateCompletableFuture() {
System.out.println("\n=== CompletableFuture Examples ===");
// Simple CompletableFuture
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return "Hello";
});
// Transform result
CompletableFuture<String> greeting = future.thenApply(s -> s + " World!");
// Consume result
greeting.thenAccept(System.out::println);
// Combine multiple futures
CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "World");
future1.thenCombine(future2, (s1, s2) -> s1 + " " + s2 + "!")
.thenAccept(System.out::println);
// Exception handling
CompletableFuture.supplyAsync(() -> {
if (Math.random() > 0.5) {
throw new RuntimeException("Simulated error");
}
return "Success";
}).exceptionally(ex -> {
System.err.println("Error: " + ex.getMessage());
return "Fallback";
}).thenAccept(System.out::println);
// Wait for completion
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}

3. Concurrent Collections

import java.util.concurrent.*;
import java.util.*;
import java.util.concurrent.atomic.*;
public class ConcurrentCollectionsExample {
public static void main(String[] args) throws Exception {
demonstrateConcurrentHashMap();
demonstrateBlockingQueues();
demonstrateCopyOnWrite();
demonstrateSynchronizedCollections();
}
public static void demonstrateConcurrentHashMap() {
System.out.println("=== ConcurrentHashMap Examples ===");
ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
// Atomic operations
map.put("key1", 1);
map.compute("key1", (k, v) -> v == null ? 1 : v + 1);
map.merge("key2", 1, Integer::sum);
// Thread-safe iteration
map.put("A", 1);
map.put("B", 2);
map.put("C", 3);
map.forEach(2, (k, v) -> 
System.out.println(Thread.currentThread().getName() + " - " + k + ": " + v));
// Search with parallelism threshold
String result = map.search(2, (k, v) -> v > 2 ? k : null);
System.out.println("Found key with value > 2: " + result);
// Reduce operation
Integer sum = map.reduceValues(2, Integer::sum);
System.out.println("Sum of values: " + sum);
}
public static void demonstrateBlockingQueues() throws InterruptedException {
System.out.println("\n=== BlockingQueue Examples ===");
// ArrayBlockingQueue - bounded
BlockingQueue<String> arrayQueue = new ArrayBlockingQueue<>(10);
// LinkedBlockingQueue - optionally bounded
BlockingQueue<String> linkedQueue = new LinkedBlockingQueue<>(20);
// PriorityBlockingQueue - unbounded, ordered
BlockingQueue<Integer> priorityQueue = new PriorityBlockingQueue<>();
// SynchronousQueue - direct handoff
BlockingQueue<String> syncQueue = new SynchronousQueue<>();
// Producer-Consumer example
BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5);
Thread producer = new Thread(() -> {
try {
for (int i = 0; i < 10; i++) {
queue.put(i); // Blocks if queue is full
System.out.println("Produced: " + i);
Thread.sleep(100);
}
queue.put(-1); // Poison pill
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
Thread consumer = new Thread(() -> {
try {
while (true) {
Integer item = queue.take(); // Blocks if queue is empty
if (item == -1) break;
System.out.println("Consumed: " + item);
Thread.sleep(200);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
producer.start();
consumer.start();
producer.join();
consumer.join();
}
public static void demonstrateCopyOnWrite() {
System.out.println("\n=== CopyOnWrite Collections ===");
// CopyOnWriteArrayList - thread-safe List for read-heavy workloads
CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>();
list.add("A");
list.add("B");
list.add("C");
// Iteration is safe even during modification
Thread writer = new Thread(() -> {
for (int i = 0; i < 5; i++) {
list.add("Element-" + i);
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
});
Thread reader = new Thread(() -> {
for (String element : list) {
System.out.println("Read: " + element);
try {
Thread.sleep(50);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
});
writer.start();
reader.start();
try {
writer.join();
reader.join();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
// CopyOnWriteArraySet
CopyOnWriteArraySet<String> set = new CopyOnWriteArraySet<>();
set.add("Apple");
set.add("Banana");
set.addAll(list);
System.out.println("Set contents: " + set);
}
public static void demonstrateSynchronizedCollections() {
System.out.println("\n=== Synchronized Collections ===");
// Creating synchronized collections
List<String> syncList = Collections.synchronizedList(new ArrayList<>());
Map<String, Integer> syncMap = Collections.synchronizedMap(new HashMap<>());
Set<String> syncSet = Collections.synchronizedSet(new HashSet<>());
// Safe to use from multiple threads
syncList.add("Item1");
syncList.add("Item2");
// Manual synchronization required for iteration
synchronized(syncList) {
for (String item : syncList) {
System.out.println("List item: " + item);
}
}
// Concurrent collections are generally preferred
}
}

4. Locks and Synchronizers

import java.util.concurrent.*;
import java.util.concurrent.locks.*;
import java.util.concurrent.atomic.*;
public class LocksAndSynchronizers {
public static void main(String[] args) throws Exception {
demonstrateReentrantLock();
demonstrateReadWriteLock();
demonstrateStampedLock();
demonstrateCountDownLatch();
demonstrateCyclicBarrier();
demonstrateSemaphore();
demonstrateExchanger();
demonstratePhaser();
}
public static void demonstrateReentrantLock() {
System.out.println("=== ReentrantLock Examples ===");
ReentrantLock lock = new ReentrantLock();
AtomicInteger counter = new AtomicInteger(0);
Runnable task = () -> {
lock.lock();
try {
int current = counter.incrementAndGet();
System.out.println(Thread.currentThread().getName() + 
" acquired lock, counter: " + current);
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
lock.unlock();
}
};
ExecutorService executor = Executors.newFixedThreadPool(3);
for (int i = 0; i < 5; i++) {
executor.execute(task);
}
executor.shutdown();
try {
executor.awaitTermination(5, TimeUnit.SECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("Final counter value: " + counter.get());
System.out.println("Lock was held by: " + lock.getHoldCount() + " threads");
System.out.println("Queue length: " + lock.getQueueLength());
}
public static void demonstrateReadWriteLock() {
System.out.println("\n=== ReadWriteLock Examples ===");
ReadWriteLock rwLock = new ReentrantReadWriteLock();
Lock readLock = rwLock.readLock();
Lock writeLock = rwLock.writeLock();
List<String> data = new ArrayList<>();
data.add("Initial Data");
// Multiple readers can read concurrently
Runnable reader = () -> {
readLock.lock();
try {
System.out.println(Thread.currentThread().getName() + " reading: " + data);
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
readLock.unlock();
}
};
// Only one writer at a time
Runnable writer = () -> {
writeLock.lock();
try {
String newData = "Data from " + Thread.currentThread().getName();
data.add(newData);
System.out.println(Thread.currentThread().getName() + " wrote: " + newData);
Thread.sleep(200);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
writeLock.unlock();
}
};
ExecutorService executor = Executors.newFixedThreadPool(5);
// Start 3 readers and 2 writers
for (int i = 0; i < 3; i++) {
executor.execute(reader);
}
for (int i = 0; i < 2; i++) {
executor.execute(writer);
}
for (int i = 0; i < 2; i++) {
executor.execute(reader);
}
executor.shutdown();
}
public static void demonstrateStampedLock() {
System.out.println("\n=== StampedLock Examples ===");
StampedLock lock = new StampedLock();
String sharedData = "Initial Value";
// Optimistic reading
Runnable optimisticReader = () -> {
long stamp = lock.tryOptimisticRead();
String dataSnapshot = sharedData;
if (!lock.validate(stamp)) {
// Fallback to pessimistic read
stamp = lock.readLock();
try {
dataSnapshot = sharedData;
} finally {
lock.unlockRead(stamp);
}
}
System.out.println(Thread.currentThread().getName() + " read: " + dataSnapshot);
};
// Write operation
Runnable writer = () -> {
long stamp = lock.writeLock();
try {
sharedData = "Modified by " + Thread.currentThread().getName();
System.out.println(Thread.currentThread().getName() + " wrote: " + sharedData);
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
lock.unlockWrite(stamp);
}
};
ExecutorService executor = Executors.newFixedThreadPool(3);
executor.execute(optimisticReader);
executor.execute(writer);
executor.execute(optimisticReader);
executor.shutdown();
}
public static void demonstrateCountDownLatch() throws InterruptedException {
System.out.println("\n=== CountDownLatch Examples ===");
int workerCount = 3;
CountDownLatch startSignal = new CountDownLatch(1);
CountDownLatch doneSignal = new CountDownLatch(workerCount);
for (int i = 0; i < workerCount; i++) {
final int workerId = i;
new Thread(() -> {
try {
System.out.println("Worker " + workerId + " waiting for start signal");
startSignal.await(); // Wait for start signal
System.out.println("Worker " + workerId + " working...");
Thread.sleep(1000 + workerId * 100); // Simulate work
System.out.println("Worker " + workerId + " completed");
doneSignal.countDown(); // Signal completion
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}).start();
}
Thread.sleep(1000); // Let workers get ready
System.out.println("Starting all workers...");
startSignal.countDown(); // Start all workers
doneSignal.await(); // Wait for all workers to complete
System.out.println("All workers completed!");
}
public static void demonstrateCyclicBarrier() {
System.out.println("\n=== CyclicBarrier Examples ===");
int partySize = 3;
CyclicBarrier barrier = new CyclicBarrier(partySize, () -> {
System.out.println("All parties reached barrier! Proceeding...");
});
for (int i = 0; i < partySize; i++) {
final int partyId = i;
new Thread(() -> {
try {
System.out.println("Party " + partyId + " starting journey");
Thread.sleep(1000 + partyId * 500); // Different arrival times
System.out.println("Party " + partyId + " reached barrier");
barrier.await(); // Wait for other parties
System.out.println("Party " + partyId + " continuing after barrier");
} catch (InterruptedException | BrokenBarrierException e) {
Thread.currentThread().interrupt();
}
}).start();
}
}
public static void demonstrateSemaphore() {
System.out.println("\n=== Semaphore Examples ===");
Semaphore semaphore = new Semaphore(3); // 3 permits available
AtomicInteger counter = new AtomicInteger(0);
Runnable limitedTask = () -> {
try {
System.out.println(Thread.currentThread().getName() + " acquiring permit");
semaphore.acquire();
int current = counter.incrementAndGet();
System.out.println(Thread.currentThread().getName() + 
" got permit. Active: " + current);
Thread.sleep(2000); // Simulate work
counter.decrementAndGet();
semaphore.release();
System.out.println(Thread.currentThread().getName() + " released permit");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
};
ExecutorService executor = Executors.newFixedThreadPool(10);
for (int i = 0; i < 10; i++) {
executor.execute(limitedTask);
}
executor.shutdown();
}
public static void demonstrateExchanger() throws InterruptedException {
System.out.println("\n=== Exchanger Examples ===");
Exchanger<String> exchanger = new Exchanger<>();
Thread producer = new Thread(() -> {
try {
String data = "Data from Producer";
System.out.println("Producer sending: " + data);
String response = exchanger.exchange(data);
System.out.println("Producer received: " + response);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
Thread consumer = new Thread(() -> {
try {
String data = "Data from Consumer";
System.out.println("Consumer sending: " + data);
String response = exchanger.exchange(data);
System.out.println("Consumer received: " + response);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
producer.start();
consumer.start();
producer.join();
consumer.join();
}
public static void demonstratePhaser() {
System.out.println("\n=== Phaser Examples ===");
Phaser phaser = new Phaser(1); // Register main thread
for (int i = 0; i < 3; i++) {
final int taskId = i;
phaser.register(); // Register new party
new Thread(() -> {
System.out.println("Task " + taskId + " starting phase " + phaser.getPhase());
phaser.arriveAndAwaitAdvance(); // Wait for all tasks
System.out.println("Task " + taskId + " completed phase " + (phaser.getPhase() - 1));
phaser.arriveAndDeregister(); // Task completed
}).start();
}
// Main thread controls phaser
phaser.arriveAndAwaitAdvance();
System.out.println("All tasks completed first phase");
phaser.arriveAndDeregister();
}
}

5. Atomic Variables and Non-blocking Algorithms

import java.util.concurrent.atomic.*;
import java.util.function.*;
public class AtomicExamples {
public static void main(String[] args) {
demonstrateAtomicVariables();
demonstrateAtomicArrays();
demonstrateAtomicFieldUpdaters();
demonstrateAccumulators();
}
public static void demonstrateAtomicVariables() {
System.out.println("=== Atomic Variables Examples ===");
AtomicInteger atomicInt = new AtomicInteger(0);
AtomicLong atomicLong = new AtomicLong(0L);
AtomicBoolean atomicBool = new AtomicBoolean(false);
AtomicReference<String> atomicRef = new AtomicReference<>("initial");
// Basic operations
atomicInt.set(42);
System.out.println("AtomicInt value: " + atomicInt.get());
// Compare-and-set (CAS) operation
boolean success = atomicInt.compareAndSet(42, 100);
System.out.println("CAS successful: " + success + ", new value: " + atomicInt.get());
// Atomic updates
System.out.println("Increment and get: " + atomicInt.incrementAndGet());
System.out.println("Add and get: " + atomicInt.addAndGet(50));
// Functional updates
atomicInt.updateAndGet(x -> x * 2);
System.out.println("After updateAndGet: " + atomicInt.get());
// Atomic reference example
atomicRef.set("Hello");
atomicRef.accumulateAndGet(" World", (a, b) -> a + b);
System.out.println("AtomicReference: " + atomicRef.get());
}
public static void demonstrateAtomicArrays() {
System.out.println("\n=== Atomic Arrays Examples ===");
AtomicIntegerArray atomicIntArray = new AtomicIntegerArray(5);
AtomicLongArray atomicLongArray = new AtomicLongArray(3);
AtomicReferenceArray<String> atomicRefArray = new AtomicReferenceArray<>(4);
// Initialize arrays
for (int i = 0; i < atomicIntArray.length(); i++) {
atomicIntArray.set(i, i * 10);
}
// Atomic operations on array elements
atomicIntArray.getAndAdd(2, 5); // Add 5 to element at index 2
atomicIntArray.compareAndSet(1, 10, 100); // CAS operation
// Print array contents
System.out.print("AtomicIntArray: ");
for (int i = 0; i < atomicIntArray.length(); i++) {
System.out.print(atomicIntArray.get(i) + " ");
}
System.out.println();
// Reference array
atomicRefArray.set(0, "First");
atomicRefArray.set(1, "Second");
atomicRefArray.compareAndSet(1, "Second", "Updated");
System.out.print("AtomicReferenceArray: ");
for (int i = 0; i < atomicRefArray.length(); i++) {
System.out.print(atomicRefArray.get(i) + " ");
}
System.out.println();
}
public static void demonstrateAtomicFieldUpdaters() {
System.out.println("\n=== Atomic Field Updaters Examples ===");
class Counter {
private volatile int count;
private volatile long total;
public int getCount() { return count; }
public long getTotal() { return total; }
}
Counter counter = new Counter();
// Create field updaters
AtomicIntegerFieldUpdater<Counter> countUpdater =
AtomicIntegerFieldUpdater.newUpdater(Counter.class, "count");
AtomicLongFieldUpdater<Counter> totalUpdater =
AtomicLongFieldUpdater.newUpdater(Counter.class, "total");
// Use updaters instead of synchronized blocks
countUpdater.incrementAndGet(counter);
totalUpdater.addAndGet(counter, 100L);
System.out.println("Count: " + counter.getCount() + ", Total: " + counter.getTotal());
// CAS operation with field updater
boolean updated = countUpdater.compareAndSet(counter, 1, 42);
System.out.println("CAS successful: " + updated + ", new count: " + counter.getCount());
}
public static void demonstrateAccumulators() {
System.out.println("\n=== Accumulators and Adders Examples ===");
// LongAdder - better performance under high contention
LongAdder longAdder = new LongAdder();
LongAccumulator longAccumulator = new LongAccumulator(Long::max, 0L);
// DoubleAdder and DoubleAccumulator also available
// Multiple threads updating adders
ExecutorService executor = Executors.newFixedThreadPool(10);
for (int i = 0; i < 100; i++) {
executor.execute(() -> {
longAdder.increment();
longAccumulator.accumulate(Thread.currentThread().getId());
});
}
executor.shutdown();
try {
executor.awaitTermination(5, TimeUnit.SECONDS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("LongAdder sum: " + longAdder.sum());
System.out.println("LongAccumulator max: " + longAccumulator.get());
// Reset accumulators
longAdder.reset();
longAccumulator.reset();
System.out.println("After reset - LongAdder: " + longAdder.sum() + 
", LongAccumulator: " + longAccumulator.get());
}
}

6. Practical Examples and Patterns

Producer-Consumer with ThreadPool

import java.util.concurrent.*;
import java.util.*;
public class ProducerConsumerPattern {
public static class DataProcessor {
private final ExecutorService producerExecutor;
private final ExecutorService consumerExecutor;
private final BlockingQueue<DataItem> queue;
private final CountDownLatch completionLatch;
private final AtomicInteger processedCount = new AtomicInteger(0);
public DataProcessor(int producerCount, int consumerCount, int queueCapacity) {
this.producerExecutor = Executors.newFixedThreadPool(producerCount);
this.consumerExecutor = Executors.newFixedThreadPool(consumerCount);
this.queue = new ArrayBlockingQueue<>(queueCapacity);
this.completionLatch = new CountDownLatch(producerCount);
}
public void startProcessing(int itemsPerProducer) {
// Start producers
for (int i = 0; i < completionLatch.getCount(); i++) {
final int producerId = i;
producerExecutor.execute(() -> produceData(producerId, itemsPerProducer));
}
// Start consumers
for (int i = 0; i < 3; i++) {
final int consumerId = i;
consumerExecutor.execute(() -> consumeData(consumerId));
}
}
private void produceData(int producerId, int itemCount) {
try {
Random random = new Random();
for (int i = 0; i < itemCount; i++) {
DataItem item = new DataItem(producerId, i, 
"Data-" + producerId + "-" + i, random.nextInt(1000));
queue.put(item); // Blocks if queue is full
System.out.println("Producer " + producerId + " produced: " + item);
Thread.sleep(random.nextInt(100)); // Simulate production time
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
completionLatch.countDown();
System.out.println("Producer " + producerId + " finished");
}
}
private void consumeData(int consumerId) {
try {
while (true) {
// Stop when all producers are done and queue is empty
if (completionLatch.getCount() == 0 && queue.isEmpty()) {
break;
}
// Poll with timeout to avoid permanent blocking
DataItem item = queue.poll(100, TimeUnit.MILLISECONDS);
if (item != null) {
processItem(consumerId, item);
processedCount.incrementAndGet();
}
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
System.out.println("Consumer " + consumerId + " finished");
}
private void processItem(int consumerId, DataItem item) throws InterruptedException {
System.out.println("Consumer " + consumerId + " processing: " + item);
Thread.sleep(item.getProcessingTime()); // Simulate processing
System.out.println("Consumer " + consumerId + " completed: " + item);
}
public void awaitCompletion() throws InterruptedException {
producerExecutor.shutdown();
consumerExecutor.shutdown();
if (!producerExecutor.awaitTermination(1, TimeUnit.MINUTES)) {
producerExecutor.shutdownNow();
}
if (!consumerExecutor.awaitTermination(1, TimeUnit.MINUTES)) {
consumerExecutor.shutdownNow();
}
System.out.println("Processing completed. Total items processed: " + processedCount.get());
}
}
public static class DataItem {
private final int producerId;
private final int itemId;
private final String data;
private final int processingTime;
public DataItem(int producerId, int itemId, String data, int processingTime) {
this.producerId = producerId;
this.itemId = itemId;
this.data = data;
this.processingTime = processingTime;
}
public int getProcessingTime() { return processingTime; }
@Override
public String toString() {
return String.format("DataItem[producer=%d, id=%d, data=%s, time=%dms]", 
producerId, itemId, data, processingTime);
}
}
public static void main(String[] args) throws Exception {
DataProcessor processor = new DataProcessor(3, 2, 10);
processor.startProcessing(5);
processor.awaitCompletion();
}
}

Rate Limiter with Semaphore

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
public class RateLimiterExample {
public static class SimpleRateLimiter {
private final Semaphore semaphore;
private final int maxPermits;
private final AtomicInteger currentPermits;
public SimpleRateLimiter(int maxPermits) {
this.maxPermits = maxPermits;
this.semaphore = new Semaphore(maxPermits);
this.currentPermits = new AtomicInteger(maxPermits);
startPermitRefill();
}
private void startPermitRefill() {
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
scheduler.scheduleAtFixedRate(() -> {
int current = currentPermits.get();
if (current < maxPermits) {
int newPermits = Math.min(maxPermits, current + 1);
if (currentPermits.compareAndSet(current, newPermits)) {
semaphore.release(1);
System.out.println("Refilled permit. Available: " + newPermits);
}
}
}, 1, 1, TimeUnit.SECONDS);
}
public boolean tryAcquire() {
return semaphore.tryAcquire();
}
public void acquire() throws InterruptedException {
semaphore.acquire();
currentPermits.decrementAndGet();
}
public int getAvailablePermits() {
return semaphore.availablePermits();
}
}
public static void main(String[] args) throws Exception {
SimpleRateLimiter rateLimiter = new SimpleRateLimiter(3);
ExecutorService executor = Executors.newFixedThreadPool(10);
for (int i = 0; i < 20; i++) {
final int taskId = i;
executor.execute(() -> {
try {
if (rateLimiter.tryAcquire()) {
System.out.println("Task " + taskId + " acquired permit");
Thread.sleep(2000); // Simulate work
System.out.println("Task " + taskId + " completed");
} else {
System.out.println("Task " + taskId + " rejected - rate limit exceeded");
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
Thread.sleep(200); // Space out task submission
}
executor.shutdown();
executor.awaitTermination(30, TimeUnit.SECONDS);
}
}

Conclusion

Key Concurrency Utilities Summary:

Best Practices:

1. Prefer high-level utilities over low-level synchronized blocks
2. Use thread pools instead of creating threads directly
3. Choose the right concurrent collection for your access pattern
4. Understand the trade-offs between different lock implementations
5. Always handle InterruptedException properly
6. Use atomic variables for simple state management
7. Consider CompletableFuture for complex async workflows
8. Test thoroughly - concurrency bugs are hard to reproduce

Java Concurrency Utilities provide robust, tested building blocks for creating high-performance, thread-safe applications while avoiding many common concurrency pitfalls.