Advanced Concurrency in Java

java.util.concurrent gives you thread pools, futures, async pipelines, divide-and-conquer pools and thread-safe data structures, so you rarely touch Thread directly. Every example is a single file run with the JDK 21 source launcher (java File.java). Each one collects its results through futures, join() or thread-safe collections, so the printed output is the same on every run — except timings, which are real numbers that will differ on your machine.

ExecutorService and Future

A pool separates submitting work from running it. submit returns a Future immediately; get() blocks until that task’s result is ready.

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class ExecutorDemo {
    public static void main(String[] args) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(4);
        List<Future<Integer>> futures = new ArrayList<>();
        long start = System.nanoTime();
        for (int i = 0; i < 8; i++) {
            int n = i;
            futures.add(pool.submit(() -> {
                Thread.sleep(100);          // simulate slow work
                return n * n;
            }));
        }
        List<Integer> results = new ArrayList<>();
        for (Future<Integer> f : futures) results.add(f.get());
        long ms = (System.nanoTime() - start) / 1_000_000;
        pool.shutdown();
        System.out.println(results);
        System.out.println("elapsed ~" + ms + " ms (varies)");
    }
}

$ java ExecutorDemo.java
[0, 1, 4, 9, 16, 25, 36, 49]
elapsed ~203 ms (varies)

Eight 100 ms tasks on four threads take about two rounds — roughly 200 ms instead of 800. The results are in submission order because we call get() on each future in order, not because the tasks finished in that order. Always shutdown() a pool you created, or its non-daemon threads keep the JVM alive.

Exceptions travel through the Future

import java.util.concurrent.*;

public class FutureFailure {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService pool = Executors.newSingleThreadExecutor();
        Future<Integer> f = pool.submit(() -> Integer.parseInt("forty-two"));
        try {
            System.out.println(f.get());
        } catch (ExecutionException e) {
            System.out.println("wrapped: " + e.getClass().getSimpleName());
            System.out.println("cause:   " + e.getCause());
        } finally {
            pool.shutdown();
        }
    }
}

$ java FutureFailure.java
wrapped: ExecutionException
cause:   java.lang.NumberFormatException: For input string: "forty-two"

The task threw on a pool thread, but nothing was printed there: submit captured the exception and get() rethrew it wrapped in ExecutionException. Never discard a Future whose failure you care about — an unchecked exception is silently stored in it.

invokeAll and invokeAny

invokeAll waits for every task; invokeAny returns the first task that succeeds and cancels the rest.

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

public class InvokeDemo {
    public static void main(String[] args) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(3);
        List<Callable<String>> mirrors = List.of(
            () -> { throw new IllegalStateException("mirror A down"); },
            () -> { Thread.sleep(50); return "data from mirror B"; },
            () -> { throw new IllegalStateException("mirror C down"); });

        for (Future<String> f : pool.invokeAll(mirrors)) {
            try { System.out.println("all: " + f.get()); }
            catch (ExecutionException e) { System.out.println("all: failed, " + e.getCause().getMessage()); }
        }
        System.out.println("any: " + pool.invokeAny(mirrors));
        pool.shutdown();
    }
}

$ java InvokeDemo.java
all: failed, mirror A down
all: data from mirror B
all: failed, mirror C down
any: data from mirror B

Two mirrors fail instantly, yet invokeAny still returns B: failures are skipped, and it only throws if every task fails.

CompletableFuture pipelines

CompletableFuture chains steps without blocking between them. thenApply transforms a result, thenCompose starts another async step that depends on it, and thenCombine joins two independent futures.

import java.util.concurrent.CompletableFuture;

public class Pipeline {
    record User(long id, String name) {}

    static CompletableFuture<User> fetchUser(long id) {
        return CompletableFuture.supplyAsync(() -> new User(id, "Alice"));
    }
    static CompletableFuture<Integer> fetchOrderCount(long userId) {
        return CompletableFuture.supplyAsync(() -> 3);
    }
    static CompletableFuture<String> fetchTier(long userId) {
        return CompletableFuture.supplyAsync(() -> "gold");
    }

    public static void main(String[] args) {
        String greeting = fetchUser(42).thenApply(u -> "Hello, " + u.name()).join();
        System.out.println(greeting);

        int orders = fetchUser(42).thenCompose(u -> fetchOrderCount(u.id())).join();
        System.out.println("orders: " + orders);

        String page = fetchOrderCount(42)
            .thenCombine(fetchTier(42), (count, tier) -> tier + " member, " + count + " orders")
            .join();
        System.out.println(page);
    }
}

$ java Pipeline.java
Hello, Alice
orders: 3
gold member, 3 orders

thenApply(u -> fetchOrderCount(...)) would give a CompletableFuture<CompletableFuture<Integer>>; thenCompose flattens it, the way flatMap does for streams. join() is like get() but throws an unchecked CompletionException, which is convenient in examples and lambdas.

Handling failures in a pipeline

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;

public class Recover {
    public static void main(String[] args) {
        CompletableFuture<Integer> failing = CompletableFuture.supplyAsync(() -> {
            if (true) throw new IllegalStateException("service unavailable");
            return 1;
        });

        int withDefault = failing.exceptionally(ex -> -1).join();
        System.out.println("exceptionally -> " + withDefault);

        String handled = failing.handle((value, ex) ->
            ex != null ? "error: " + ex.getCause().getMessage() : "value " + value).join();
        System.out.println("handle -> " + handled);

        try {
            failing.thenApply(v -> v * 2).join();
        } catch (CompletionException e) {
            System.out.println("thenApply -> " + e);
        }
    }
}

$ java Recover.java
exceptionally -> -1
handle -> error: service unavailable
thenApply -> java.util.concurrent.CompletionException: java.lang.IllegalStateException: service unavailable

exceptionally swaps a failure for a fallback value; handle sees both outcomes. A stage with no recovery, like thenApply, is skipped and passes the failure along, so the final join() throws CompletionException with the original exception as its cause. Inside handle the exception arrives wrapped in CompletionException, which is why the message comes from ex.getCause().

Waiting for many futures with allOf

import java.util.List;
import java.util.concurrent.CompletableFuture;

public class AllOf {
    static CompletableFuture<String> fetch(int id) {
        return CompletableFuture.supplyAsync(() -> "user-" + id);
    }

    public static void main(String[] args) {
        List<CompletableFuture<String>> futures = List.of(fetch(1), fetch(2), fetch(3));
        CompletableFuture<Void> all = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
        List<String> users = all.thenApply(v -> futures.stream().map(CompletableFuture::join).toList()).join();
        System.out.println(users);
    }
}

$ java AllOf.java
[user-1, user-2, user-3]

allOf returns CompletableFuture<Void> — it only signals that everything finished. The values come from calling join() on the original futures afterwards, which no longer blocks.

Fork/Join: divide and conquer

A RecursiveTask splits its range until it is small enough, forks one half to another worker, computes the other half itself, and joins.

import java.util.concurrent.*;

public class ForkJoinSum extends RecursiveTask<Long> {
    static final int THRESHOLD = 10_000;
    final long[] data; final int from, to;
    ForkJoinSum(long[] data, int from, int to) { this.data = data; this.from = from; this.to = to; }

    protected Long compute() {
        if (to - from <= THRESHOLD) {
            long sum = 0;
            for (int i = from; i < to; i++) sum += data[i];
            return sum;
        }
        int mid = (from + to) >>> 1;
        ForkJoinSum left = new ForkJoinSum(data, from, mid);
        left.fork();                                   // runs on another worker if one is free
        long right = new ForkJoinSum(data, mid, to).compute();
        return left.join() + right;
    }

    public static void main(String[] args) {
        long[] data = new long[1_000_000];
        for (int i = 0; i < data.length; i++) data[i] = i + 1;
        System.out.println("fork/join: " + ForkJoinPool.commonPool().invoke(new ForkJoinSum(data, 0, data.length)));
        System.out.println("parallel stream: " + java.util.Arrays.stream(data).parallel().sum());
    }
}

$ java ForkJoinSum.java
fork/join: 500000500000
parallel stream: 500000500000

Both equal n(n+1)/2 for n = 1,000,000. A parallel stream does the same splitting on the same common pool for you. Fork one half and compute() the other — forking both would leave the current thread idle while it waits.

Producer-consumer with BlockingQueue

put blocks while the queue is full and take blocks while it is empty, so producers and consumers pace each other with no explicit locking. A “poison pill” tells each consumer to stop.

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

public class ProducerConsumer {
    static final int POISON = -1;

    public static void main(String[] args) throws Exception {
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5);
        Set<Integer> processed = ConcurrentHashMap.newKeySet();
        int consumers = 3;

        List<Thread> workers = new ArrayList<>();
        for (int c = 0; c < consumers; c++) {
            workers.add(Thread.ofPlatform().start(() -> {
                try {
                    for (int task; (task = queue.take()) != POISON; ) processed.add(task * 10);
                } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            }));
        }
        for (int i = 1; i <= 20; i++) queue.put(i);          // blocks whenever 5 are waiting
        for (int c = 0; c < consumers; c++) queue.put(POISON);
        for (Thread t : workers) t.join();

        System.out.println(processed.size() + " tasks processed");
        System.out.println(new TreeSet<>(processed));
    }
}

$ java ProducerConsumer.java
20 tasks processed
[10, 20, 30, 40, 50, 60, 70, 80, 90, 100, 110, 120, 130, 140, 150, 160, 170, 180, 190, 200]

Which consumer handled which task changes on every run, so the program never prints that. It prints only what is guaranteed once every consumer has been join()ed: all 20 tasks were processed exactly once. One pill per consumer is needed, because each consumer exits after taking one.

Lost updates and the atomic fix

count++ is read, add, write — three steps another thread can interleave with.

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

public class Counters {
    static int plain = 0;

    public static void main(String[] args) throws Exception {
        AtomicInteger atomic = new AtomicInteger();
        LongAdder adder = new LongAdder();
        ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();

        try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
            for (int t = 0; t < 4; t++) {
                pool.submit(() -> {
                    for (int i = 0; i < 100_000; i++) {
                        plain++;
                        atomic.incrementAndGet();
                        adder.increment();
                        map.merge("hits", 1, Integer::sum);
                    }
                });
            }
        }   // close() waits for every task to finish
        System.out.println("plain int == 400000? " + (plain == 400_000));
        System.out.println("AtomicInteger: " + atomic.get());
        System.out.println("LongAdder:     " + adder.sum());
        System.out.println("map.merge:     " + map.get("hits"));
    }
}

$ java Counters.java
plain int == 400000? false
AtomicInteger: 400000
LongAdder:     400000
map.merge:     400000

The plain counter lost updates; how many varies from run to run, which is why only the comparison is printed. The other three are correct every time: incrementAndGet and merge are atomic, and LongAdder spreads contention over several cells, which usually scales better than AtomicInteger for a heavily shared counter. Since Java 19, ExecutorService is AutoCloseable, so try-with-resources shuts the pool down and waits.

Virtual threads for blocking work

A virtual thread is cheap enough to create one per task. Blocking a virtual thread (sleep, I/O) frees its carrier thread to run another.

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

public class VirtualThreads {
    public static void main(String[] args) {
        AtomicInteger done = new AtomicInteger();
        long start = System.nanoTime();
        try (ExecutorService exec = Executors.newVirtualThreadPerTaskExecutor()) {
            for (int i = 0; i < 10_000; i++) {
                exec.submit(() -> {
                    Thread.sleep(100);          // stands in for a blocking network call
                    return done.incrementAndGet();
                });
            }
        }
        long ms = (System.nanoTime() - start) / 1_000_000;
        System.out.println(done.get() + " tasks done in ~" + ms + " ms (varies)");
    }
}

$ java VirtualThreads.java
10000 tasks done in ~185 ms (varies)

Ten thousand tasks that each block for 100 ms finish in a fraction of a second. A fixed pool of 4 platform threads would need 10,000 × 100 ms ÷ 4 ≈ 250 seconds for the same work. Virtual threads help blocking I/O; they do not make CPU-bound work faster.

Semaphore: limit concurrent access

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

public class Throttle {
    public static void main(String[] args) {
        Semaphore permits = new Semaphore(3);
        AtomicInteger inside = new AtomicInteger(), maxInside = new AtomicInteger();
        try (ExecutorService exec = Executors.newVirtualThreadPerTaskExecutor()) {
            for (int i = 0; i < 30; i++) {
                exec.submit(() -> {
                    permits.acquire();
                    try {
                        maxInside.accumulateAndGet(inside.incrementAndGet(), Math::max);
                        Thread.sleep(20);
                        inside.decrementAndGet();
                    } finally { permits.release(); }
                    return null;
                });
            }
        }
        System.out.println("max callers inside at once: " + maxInside.get());
        System.out.println("never more than 3? " + (maxInside.get() <= 3));
    }
}

$ java Throttle.java
max callers inside at once: 3
never more than 3? true

Thirty tasks start at once, but acquire() blocks when all three permits are taken, so at most three are ever inside. This is the standard way to cap calls to a rate-limited API. Release in finally, or one exception leaks a permit forever.

Practice

1. Change ExecutorDemo to Executors.newFixedThreadPool(2). Predict the elapsed time for eight 100 ms tasks, then run it.

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class TwoThreads {
    public static void main(String[] args) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(2);
        List<Future<Integer>> futures = new ArrayList<>();
        long start = System.nanoTime();
        for (int i = 0; i < 8; i++) {
            int n = i;
            futures.add(pool.submit(() -> { Thread.sleep(100); return n * n; }));
        }
        for (Future<Integer> f : futures) f.get();
        System.out.println("elapsed ~" + (System.nanoTime() - start) / 1_000_000 + " ms (varies)");
        pool.shutdown();
    }
}

$ java TwoThreads.java
elapsed ~403 ms (varies)

2. Use thenCombine to add the results of two supplyAsync calls that return 20 and 22, and recover with exceptionally if either fails.

import java.util.concurrent.CompletableFuture;

public class Combine {
    public static void main(String[] args) {
        CompletableFuture<Integer> a = CompletableFuture.supplyAsync(() -> 20);
        CompletableFuture<Integer> b = CompletableFuture.supplyAsync(() -> 22);
        int sum = a.thenCombine(b, Integer::sum).exceptionally(ex -> 0).join();
        System.out.println(sum);

        CompletableFuture<Integer> bad = CompletableFuture.supplyAsync(() -> 1 / 0);
        System.out.println(a.thenCombine(bad, Integer::sum).exceptionally(ex -> 0).join());
    }
}

$ java Combine.java
42
0

3. Run ProducerConsumer with 4 consumers. What else has to change, and do all 20 tasks still get processed?

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

public class FourConsumers {
    public static void main(String[] args) throws Exception {
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5);
        Set<Integer> processed = ConcurrentHashMap.newKeySet();
        int consumers = 4;
        List<Thread> workers = new ArrayList<>();
        for (int c = 0; c < consumers; c++) {
            workers.add(Thread.ofPlatform().start(() -> {
                try {
                    for (int task; (task = queue.take()) != -1; ) processed.add(task);
                } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
            }));
        }
        for (int i = 1; i <= 20; i++) queue.put(i);
        for (int c = 0; c < consumers; c++) queue.put(-1);   // one pill per consumer
        for (Thread t : workers) t.join();
        System.out.println(processed.size() + " processed, missing: "
            + (processed.containsAll(java.util.stream.IntStream.rangeClosed(1, 20).boxed().toList()) ? "none" : "some"));
    }
}

$ java FourConsumers.java
20 processed, missing: none

4. Replace map.merge in Counters with map.put("hits", map.getOrDefault("hits", 0) + 1) on the same ConcurrentHashMap. Is a thread-safe map enough?

import java.util.concurrent.*;

public class CheckThenAct {
    public static void main(String[] args) {
        ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
        try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
            for (int t = 0; t < 4; t++) {
                pool.submit(() -> {
                    for (int i = 0; i < 100_000; i++) map.put("hits", map.getOrDefault("hits", 0) + 1);
                });
            }
        }
        System.out.println("correct? " + (map.get("hits") == 400_000));
    }
}

$ java CheckThenAct.java
correct? false