Advanced Concurrency in Java

ExecutorService

ExecutorService decouples task submission from thread management:

import java.util.concurrent.*;
import java.util.*;

public class ExecutorDemo {

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

        // Fixed pool — good for CPU-bound work (threads = CPU cores)
        int cores = Runtime.getRuntime().availableProcessors();
        ExecutorService pool = Executors.newFixedThreadPool(cores);

        // Submit tasks and collect futures
        List<Future<Integer>> futures = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            final int taskId = i;
            futures.add(pool.submit(() -> {
                // Simulate work
                Thread.sleep(100);
                return taskId * taskId;
            }));
        }

        // Collect results
        for (Future<Integer> f : futures) {
            System.out.println(f.get()); // blocks until result is ready
        }

        // Always shut down the pool
        pool.shutdown();
        if (!pool.awaitTermination(30, TimeUnit.SECONDS)) {
            pool.shutdownNow();
        }
    }
}

ExecutorService Factory Methods

// Fixed pool — predictable resource usage
ExecutorService fixed = Executors.newFixedThreadPool(8);

// Cached pool — creates threads on demand, reuses idle ones (can grow unbounded)
ExecutorService cached = Executors.newCachedThreadPool();

// Single-thread — sequential execution, tasks queued
ExecutorService single = Executors.newSingleThreadExecutor();

// Scheduled — run tasks after a delay or on a schedule
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
scheduler.scheduleAtFixedRate(() -> checkHealth(), 0, 30, TimeUnit.SECONDS);

// Virtual thread per task (Java 21) — best for I/O-bound work
ExecutorService vtp = Executors.newVirtualThreadPerTaskExecutor();

invokeAll and invokeAny

List<Callable<String>> tasks = List.of(
    () -> fetchFromServiceA(),
    () -> fetchFromServiceB(),
    () -> fetchFromServiceC()
);

// invokeAll — wait for ALL tasks, returns List<Future<T>>
List<Future<String>> all = pool.invokeAll(tasks, 10, TimeUnit.SECONDS);
for (Future<String> f : all) {
    if (!f.isCancelled()) System.out.println(f.get());
}

// invokeAny — return the FIRST successful result, cancel the rest
String fastest = pool.invokeAny(tasks, 10, TimeUnit.SECONDS);
System.out.println("Fastest: " + fastest);

CompletableFuture

CompletableFuture composes async operations into pipelines:

import java.util.concurrent.*;
import java.util.function.*;

public class CompletableFutureDemo {

    // Simulated async operations
    static CompletableFuture<User> fetchUser(long id) {
        return CompletableFuture.supplyAsync(() -> {
            sleep(50);
            return new User(id, "Alice");
        });
    }

    static CompletableFuture<List<Order>> fetchOrders(long userId) {
        return CompletableFuture.supplyAsync(() -> {
            sleep(80);
            return List.of(new Order("ORD-1", userId, 99.99));
        });
    }

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

        // 1. Basic async computation
        CompletableFuture<String> cf = CompletableFuture
            .supplyAsync(() -> "hello")
            .thenApply(String::toUpperCase)
            .thenApply(s -> s + "!");
        System.out.println(cf.get()); // HELLO!

        // 2. Sequential composition with thenCompose (flatMap equivalent)
        CompletableFuture<List<Order>> pipeline = fetchUser(42)
            .thenCompose(user -> fetchOrders(user.id())); // runs after fetchUser
        System.out.println(pipeline.get());

        // 3. Combine two independent futures with thenCombine
        CompletableFuture<User>        userFuture   = fetchUser(42);
        CompletableFuture<UserProfile> profileFuture = fetchProfile(42);

        CompletableFuture<UserPage> page = userFuture.thenCombine(
            profileFuture,
            (user, profile) -> new UserPage(user, profile)
        );

        // 4. Wait for ALL futures
        CompletableFuture<Void> allDone = CompletableFuture.allOf(
            fetchUser(1), fetchUser(2), fetchUser(3)
        );
        allDone.get(); // blocks until all complete

        // 5. Wait for ANY future (first to complete)
        CompletableFuture<Object> anyDone = CompletableFuture.anyOf(
            fetchFromRegionA(), fetchFromRegionB()
        );
        Object result = anyDone.get();

        // 6. Error handling
        CompletableFuture<User> withFallback = fetchUser(42)
            .exceptionally(ex -> {
                System.err.println("Fetch failed: " + ex.getMessage());
                return new User(-1, "Guest"); // fallback value
            });

        // 7. whenComplete — runs after success OR failure
        fetchUser(42).whenComplete((user, ex) -> {
            if (ex != null) System.err.println("Error: " + ex);
            else            System.out.println("Got user: " + user);
        });

        // 8. handle — like exceptionally but always runs
        CompletableFuture<String> handled = fetchUser(42)
            .handle((user, ex) -> ex != null ? "error" : user.name());
    }

    static void sleep(long ms) {
        try { Thread.sleep(ms); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }
}

Parallel HTTP Calls with CompletableFuture

import java.net.http.*;
import java.net.URI;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.*;

public class ParallelRequests {

    static final HttpClient HTTP = HttpClient.newHttpClient();

    static CompletableFuture<String> fetch(String url) {
        return HTTP.sendAsync(
            HttpRequest.newBuilder(URI.create(url)).build(),
            HttpResponse.BodyHandlers.ofString()
        ).thenApply(HttpResponse::body);
    }

    public static void main(String[] args) throws Exception {
        List<String> urls = List.of(
            "https://api.example.com/users/1",
            "https://api.example.com/users/2",
            "https://api.example.com/users/3"
        );

        // Fire all requests simultaneously
        List<CompletableFuture<String>> futures = urls.stream()
            .map(ParallelRequests::fetch)
            .toList();

        // Collect all results
        List<String> responses = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
            .thenApply(v -> futures.stream().map(CompletableFuture::join).collect(Collectors.toList()))
            .get(10, TimeUnit.SECONDS);

        responses.forEach(System.out::println);
    }
}

Fork/Join Framework

Designed for divide-and-conquer recursive tasks:

import java.util.concurrent.*;

public class ForkJoinDemo {

    // RecursiveTask<T> — returns a value
    static class ParallelSum extends RecursiveTask<Long> {
        private static final int THRESHOLD = 10_000;
        private final long[] array;
        private final int from, to;

        ParallelSum(long[] array, int from, int to) {
            this.array = array; this.from = from; this.to = to;
        }

        @Override
        protected Long compute() {
            int size = to - from;
            if (size <= THRESHOLD) {
                // Base case — do the work directly
                long sum = 0;
                for (int i = from; i < to; i++) sum += array[i];
                return sum;
            }
            // Recursive case — split in half
            int mid = from + size / 2;
            ParallelSum left  = new ParallelSum(array, from, mid);
            ParallelSum right = new ParallelSum(array, mid, to);

            left.fork();         // submit left subtask asynchronously
            long rightResult = right.compute(); // compute right in this thread
            long leftResult  = left.join();     // wait for left result

            return leftResult + rightResult;
        }
    }

    public static void main(String[] args) throws Exception {
        long[] data = new long[1_000_000];
        for (int i = 0; i < data.length; i++) data[i] = i + 1;

        ForkJoinPool pool = ForkJoinPool.commonPool();
        long sum = pool.invoke(new ParallelSum(data, 0, data.length));
        System.out.println("Sum: " + sum); // 500000500000

        // Parallel streams use the common ForkJoinPool implicitly
        long sum2 = java.util.Arrays.stream(data).parallel().sum();
        System.out.println("Stream sum: " + sum2); // same result
    }
}

BlockingQueue — Producer-Consumer

import java.util.concurrent.*;

public class ProducerConsumer {

    record Task(int id, String payload) {}

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

        BlockingQueue<Task> queue = new LinkedBlockingQueue<>(100); // bounded

        // Producer — puts tasks into the queue
        Thread producer = Thread.ofVirtual().start(() -> {
            try {
                for (int i = 0; i < 20; i++) {
                    queue.put(new Task(i, "data-" + i)); // blocks if full
                    System.out.println("Produced task " + i);
                    Thread.sleep(50);
                }
                // Poison pill — signal consumers to stop
                for (int c = 0; c < 3; c++) queue.put(new Task(-1, "STOP"));
            } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        });

        // 3 consumers
        for (int c = 0; c < 3; c++) {
            final int consumerId = c;
            Thread.ofVirtual().start(() -> {
                try {
                    while (true) {
                        Task task = queue.take(); // blocks if empty
                        if (task.id() == -1) break; // stop on poison pill
                        System.out.println("Consumer " + consumerId + " processed task " + task.id());
                        Thread.sleep(100);
                    }
                } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            });
        }

        producer.join();
    }
}

Concurrent Data Structures

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;

// ConcurrentHashMap — thread-safe, no full locking
ConcurrentMap<String, Integer> wordCount = new ConcurrentHashMap<>();
wordCount.merge("hello", 1, Integer::sum); // atomic increment
wordCount.compute("world", (k, v) -> v == null ? 1 : v + 1);
wordCount.computeIfAbsent("java", k -> 0);

// CopyOnWriteArrayList — reads are lock-free (copy on write)
// Use when reads >> writes (e.g., listener lists)
CopyOnWriteArrayList<String> listeners = new CopyOnWriteArrayList<>();
listeners.add("listener1");
for (String l : listeners) { /* safe even if listeners is modified elsewhere */ }

// Atomic variables — lock-free single-variable operations
AtomicInteger counter = new AtomicInteger(0);
counter.incrementAndGet();
counter.addAndGet(5);
counter.compareAndSet(5, 10); // CAS — atomic if 5, set to 10

AtomicLong requestCount = new AtomicLong();
LongAdder  hotCounter   = new LongAdder(); // better than AtomicLong for high-contention counting
hotCounter.increment();
long total = hotCounter.sum();

// AtomicReference — atomic object reference update
AtomicReference<String> config = new AtomicReference<>("v1");
config.compareAndSet("v1", "v2"); // atomic swap

// Semaphore — limit concurrent access
Semaphore permits = new Semaphore(10); // allow 10 threads at once
permits.acquire();
try { callExternalApi(); }
finally { permits.release(); }