Executors and CompletableFuture

An executor separates submitting a task from running it: code hands a Runnable or Callable to an ExecutorService, and a pool of reused threads runs it. This lesson covers how a ThreadPoolExecutor decides to queue, grow, or reject, how to size a pool, how ForkJoinPool differs, and how to compose asynchronous results with CompletableFuture. It builds on Locks and Synchronizers.

API	Role
`ExecutorService`	Submit tasks, get `Future`s, shut down
`ThreadPoolExecutor`	Configurable pool: core, max, queue, keep-alive, rejection
`ForkJoinPool`	Work-stealing pool for divide-and-conquer and async tasks
`Future`	Handle to a result: `get`, `cancel`, `isDone`
`CompletableFuture`	Composable future with callbacks and error handling

Example 1: submit, Future, and shutdown

submit(Callable) returns a Future. Future.get() blocks until the result is ready and wraps any exception thrown by the task in an ExecutionException.

Java

import java.util.concurrent.*;

public class FirstPool {
    public static void main(String[] args) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(2);
        Future<Integer> sum = pool.submit(() -> {
            int s = 0;
            for (int i = 1; i <= 100; i++) s += i;
            return s;
        });
        Future<Integer> bad = pool.submit(() -> { throw new IllegalStateException("boom"); });

        System.out.println("sum = " + sum.get());
        try {
            bad.get();
        } catch (ExecutionException e) {
            System.out.println("cause: " + e.getCause());
        }

        pool.shutdown();                                  // no new tasks; queued ones still run
        System.out.println("terminated: " + pool.awaitTermination(5, TimeUnit.SECONDS));
    }
}

Output

sum = 5050
cause: java.lang.IllegalStateException: boom
terminated: true

shutdown() stops accepting tasks and lets submitted ones finish. shutdownNow() also interrupts running tasks and returns the queued ones that never started. A pool that is never shut down keeps its non-daemon threads alive, so the JVM does not exit. An exception in a task given to submit is stored in the Future; if nobody calls get(), it is never seen. With execute(Runnable), the exception goes to the thread's uncaught exception handler instead.

Example 2: ThreadPoolExecutor parameters and rejection

The ThreadPoolExecutor constructor takes core size, max size, keep-alive time, a work queue, a thread factory, and a rejection handler. A new task follows this order:

Fewer than corePoolSize threads: start a new thread for it.
Otherwise, offer it to the queue.
Queue full and fewer than maximumPoolSize threads: start a new thread.
Otherwise: call the RejectedExecutionHandler.

Java

import java.util.concurrent.*;

public class Rejection {
    public static void main(String[] args) throws InterruptedException {
        CountDownLatch release = new CountDownLatch(1);
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                1, 2, 30, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1),
                new ThreadPoolExecutor.AbortPolicy());

        Runnable blocker = () -> {
            try { release.await(); } catch (InterruptedException ignored) {}
        };
        for (int i = 1; i <= 4; i++) {
            try {
                pool.execute(blocker);
                System.out.println("task " + i + " accepted, threads = " + pool.getPoolSize()
                        + ", queued = " + pool.getQueue().size());
            } catch (RejectedExecutionException e) {
                System.out.println("task " + i + " rejected");
            }
        }
        release.countDown();
        pool.shutdown();
        pool.awaitTermination(5, TimeUnit.SECONDS);
        System.out.println("completed = " + pool.getCompletedTaskCount());
    }
}

Output

task 1 accepted, threads = 1, queued = 0
task 2 accepted, threads = 1, queued = 1
task 3 accepted, threads = 2, queued = 1
task 4 rejected
completed = 3

Task 1 starts the core thread, task 2 goes to the queue, task 3 finds the queue full and starts a second thread up to the max, and task 4 is rejected. The built-in policies:

Policy	Behavior
`AbortPolicy` (default)	Throws `RejectedExecutionException`
`CallerRunsPolicy`	Runs the task in the submitting thread, which slows the producer (back-pressure)
`DiscardPolicy`	Drops the task silently
`DiscardOldestPolicy`	Drops the oldest queued task and retries

The queue choice changes the pool's behavior. With an unbounded LinkedBlockingQueue, step 3 never happens, so the pool never grows beyond core size and the queue can grow until memory runs out. Executors.newFixedThreadPool uses exactly that unbounded queue. Executors.newCachedThreadPool uses a SynchronousQueue (no capacity) with an unbounded max, so it creates a thread per concurrent task. Production code usually builds a ThreadPoolExecutor with a bounded queue, named threads, and an explicit rejection policy.

Example 3: Sizing pools and naming threads

Pool size follows from the kind of work:

CPU-bound tasks keep a core busy. More threads than cores add context switches without adding throughput. A common size is availableProcessors() or that plus one.
IO-bound tasks spend most of their time waiting. The formula from Java Concurrency in Practice is threads = cores * utilization * (1 + wait time / compute time). A task that waits 90 ms and computes 10 ms on 8 cores at full utilization gives 8 * 1 * (1 + 9) = 80 threads.

The formula gives a starting point; the limit usually comes from downstream capacity (database connections, remote rate limits). For IO-bound work on Java 21, virtual threads remove the need to size a pool at all (Virtual Threads in Depth).

Java

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

public class Sizing {
    static ThreadFactory named(String prefix) {
        AtomicInteger n = new AtomicInteger();
        return r -> {
            Thread t = new Thread(r, prefix + "-" + n.incrementAndGet());
            t.setDaemon(false);
            return t;
        };
    }

    public static void main(String[] args) throws Exception {
        int cores = 8;                                  // fixed for a deterministic example
        double waitMs = 90, computeMs = 10;
        int ioThreads = (int) (cores * (1 + waitMs / computeMs));
        System.out.println("cpu pool = " + cores + ", io pool = " + ioThreads);

        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                2, 2, 0, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(100), named("pricing"),
                new ThreadPoolExecutor.CallerRunsPolicy());
        Future<String> name = pool.submit(() -> Thread.currentThread().getName());
        System.out.println("ran on " + name.get());
        pool.shutdown();
    }
}

Output

cpu pool = 8, io pool = 80
ran on pricing-1

Named threads make thread dumps and logs readable. The first task starts the first core thread, so it runs on pricing-1.

Example 4: ForkJoinPool and work stealing

A ForkJoinPool gives each worker its own double-ended queue (deque). A worker pushes and pops its own tasks at one end (LIFO, cache-friendly), and idle workers steal from the other end of a busy worker's deque (FIFO, takes the largest, oldest subtasks). This suits recursive divide-and-conquer tasks that fork() subtasks and join() them.

Java

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

public class SumTask extends RecursiveTask<Long> {
    private static final int THRESHOLD = 10_000;
    private final long[] data;
    private final int from, to;

    SumTask(long[] data, int from, int to) { this.data = data; this.from = from; this.to = to; }

    @Override
    protected Long compute() {
        if (to - from <= THRESHOLD) {
            long s = 0;
            for (int i = from; i < to; i++) s += data[i];
            return s;
        }
        int mid = (from + to) >>> 1;
        SumTask left = new SumTask(data, from, mid);
        left.fork();                                   // push to own deque
        long right = new SumTask(data, mid, to).compute();
        return right + left.join();
    }

    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;
        try (ForkJoinPool pool = new ForkJoinPool(4)) {
            System.out.println(pool.invoke(new SumTask(data, 0, data.length)));
        }
    }
}

Output

500000500000

The pattern is fork one half, compute the other half in the current thread, then join. Blocking calls inside fork/join tasks starve the pool, because a blocked worker cannot steal. ForkJoinPool.commonPool() is a JVM-wide shared instance with parallelism availableProcessors() - 1 by default; parallel streams and the default CompletableFuture async methods use it. ForkJoinPool implements AutoCloseable from Java 19, which the try block uses.

Example 5: CompletableFuture composition

CompletableFuture<T> is a future that can be completed by code and that runs callbacks when it completes. The core methods:

Method	Input	Use
`supplyAsync(supplier, executor)`	none	Start an async computation
`thenApply(fn)`	`T -> U`	Transform the value (like `map`)
`thenCompose(fn)`	`T -> CompletableFuture<U>`	Chain a dependent async call (like `flatMap`)
`thenCombine(other, fn)`	two values	Join two independent results
`allOf(cfs...)`	many futures	Completes when all complete, returns `CompletableFuture<Void>`
`thenAccept` / `thenRun`	value / nothing	Side effect at the end

Java

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

public class Quote {
    static CompletableFuture<Double> price(String sym, Executor ex) {
        return CompletableFuture.supplyAsync(() -> switch (sym) {
            case "AAPL" -> 190.0;
            case "MSFT" -> 410.0;
            default -> 0.0;
        }, ex);
    }

    static CompletableFuture<Double> fxRate(String ccy, Executor ex) {
        return CompletableFuture.supplyAsync(() -> ccy.equals("EUR") ? 0.5 : 1.0, ex);
    }

    public static void main(String[] args) {
        try (ExecutorService ex = Executors.newFixedThreadPool(4)) {
            CompletableFuture<String> eur = price("AAPL", ex)
                    .thenCombine(fxRate("EUR", ex), (p, r) -> p * r)
                    .thenApply(v -> "AAPL in EUR = " + v);
            System.out.println(eur.join());

            CompletableFuture<Double> chained = CompletableFuture
                    .supplyAsync(() -> "MSFT", ex)
                    .thenCompose(sym -> price(sym, ex));     // dependent async call
            System.out.println("MSFT = " + chained.join());

            List<CompletableFuture<Double>> all = List.of("AAPL", "MSFT").stream()
                    .map(s -> price(s, ex)).toList();
            double total = CompletableFuture.allOf(all.toArray(new CompletableFuture[0]))
                    .thenApply(v -> all.stream().mapToDouble(CompletableFuture::join).sum())
                    .join();
            System.out.println("total = " + total);
        }
    }
}

Output

AAPL in EUR = 95.0
MSFT = 410.0
total = 600.0

join() is like get() but throws the unchecked CompletionException instead of checked exceptions. After allOf completes, each join() on the parts returns at once. Callbacks without the Async suffix run either in the thread that completes the previous stage or in the thread that registers the callback, if the stage is already done. The Async variants without an executor argument run on ForkJoinPool.commonPool().

Example 6: Exception handling

An exception in any stage completes the future exceptionally, and later thenApply stages are skipped. Three methods handle it:

exceptionally(fn): called only on failure; returns a fallback value.
handle((value, error) -> ...): called always; one argument is null.
whenComplete((value, error) -> ...): called always for side effects; the result passes through unchanged.

Java

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

public class Errors {
    public static void main(String[] args) {
        CompletableFuture<Integer> failing = CompletableFuture
                .supplyAsync(() -> Integer.parseInt("12x"))
                .thenApply(n -> n * 2);                     // skipped

        System.out.println(failing.exceptionally(e -> -1).join());

        String h = failing.handle((v, e) ->
                e == null ? "ok " + v : "failed: " + e.getCause().getClass().getSimpleName()).join();
        System.out.println(h);

        try {
            failing.join();
        } catch (CompletionException e) {
            System.out.println("join threw " + e.getCause().getClass().getSimpleName());
        }

        CompletableFuture<String> slow = new CompletableFuture<>();
        slow.completeOnTimeout("default", 10, java.util.concurrent.TimeUnit.MILLISECONDS);
        System.out.println(slow.join());
    }
}

Output

-1
failed: NumberFormatException
join threw NumberFormatException
default

Inside handle and exceptionally, the error arriving from an earlier async stage is usually a CompletionException wrapping the original, so code reads getCause(). completeOnTimeout and orTimeout (Java 9) add deadlines. cancel(true) on a CompletableFuture does not interrupt the running computation; it only completes the future with a CancellationException.

Example 7: Common pool pitfalls and a dedicated executor

supplyAsync(supplier) without an executor uses the common pool. Blocking IO there occupies a small, shared set of threads that parallel streams and other libraries also use. The common pool's parallelism is availableProcessors() - 1 (at least 1), so on a small container it may have a single worker. Earlier releases fell back to a new thread per async task when that parallelism was 1; on Java 25 supplyAsync runs on the common pool's worker even then, which an experiment with -XX:ActiveProcessorCount=1 shows.

Java

import java.util.concurrent.*;

public class DedicatedPool {
    static String fetch(int id) {
        try { Thread.sleep(20); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        return "user-" + id;
    }

    public static void main(String[] args) {
        ThreadPoolExecutor io = new ThreadPoolExecutor(
                16, 16, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000),
                r -> new Thread(r, "io"), new ThreadPoolExecutor.CallerRunsPolicy());
        try (io) {
            var futures = new java.util.ArrayList<CompletableFuture<String>>();
            for (int i = 0; i < 32; i++) {
                int id = i;
                futures.add(CompletableFuture.supplyAsync(() -> fetch(id), io)
                        .orTimeout(2, TimeUnit.SECONDS));
            }
            long ok = futures.stream().map(CompletableFuture::join).filter(s -> s.startsWith("user-")).count();
            System.out.println("fetched " + ok);
        }
        System.out.println("pool terminated: " + io.isTerminated());
    }
}

Output

fetched 32
pool terminated: true

The dedicated pool isolates blocking calls from the common pool. Since Java 19, ExecutorService extends AutoCloseable. Its close() calls shutdown() and waits for all tasks to finish; if the waiting thread is interrupted, it calls shutdownNow(). That is why isTerminated() is true after the try block.

Interview points

ThreadPoolExecutor order: core threads, then queue, then grow to max, then reject.
An unbounded queue means the pool never grows past core size and memory is the only limit; newFixedThreadPool has this property.
CallerRunsPolicy gives back-pressure by running the task on the submitting thread.
CPU-bound pools size near the core count; IO-bound pools use cores * (1 + wait/compute), capped by downstream capacity.
ForkJoinPool uses per-worker deques with LIFO local work and FIFO stealing; blocking in it starves the pool.
thenApply maps, thenCompose flat-maps, thenCombine joins two independent futures, allOf waits for many.
Errors skip later stages; handle them with exceptionally, handle, or whenComplete; causes arrive wrapped in CompletionException.
Default async methods use the common pool; pass a dedicated executor for blocking work.
ExecutorService.close() (Java 19) shuts down and waits.

Summary

Executors reuse threads and decouple submission from execution.
Future.get wraps task exceptions in ExecutionException.
Pool behavior depends on core size, max size, queue type, and rejection policy.
Work stealing balances recursive tasks across ForkJoinPool workers.
CompletableFuture composes asynchronous stages and propagates failures.
cancel(true) on a CompletableFuture does not interrupt the computation.
shutdown lets queued tasks run; shutdownNow interrupts and returns queued tasks.