Phase 2: The java.util.concurrent toolkitCoreHighest priority

Module 6: Executors and thread pools

Real services do not create a thread per task. They submit tasks to a pool of reusable threads with a queue in front. This module covers the executor interfaces, how ThreadPoolExecutor decides where each task goes, queues and rejection, execute vs submit, shutdown, scheduling, and sizing.

12 lessons5 figures6 mastery checks

Warm-upWarm-up: one line each

Executors.newFixedThreadPool(10) receives tasks faster than it can run them. What grows, and what eventually happens?
Reveal
Its queue, which is an unbounded LinkedBlockingQueue. Memory fills with waiting tasks and latency grows until the service slows to a crawl or hits OutOfMemoryError.
A ThreadPoolExecutor has core 2, max 10 and a queue of capacity 1,000. Five tasks arrive at once. How many threads run them?
Reveal
Two. Extra threads beyond the core size start only when the queue is full. The other three tasks wait in the queue.
A task submitted with submit() throws an exception. Where does the exception go?
Reveal
Into the returned Future. You see it only by calling get(), which throws ExecutionException. If nobody calls get(), it is lost silently.
What is the difference between shutdown() and shutdownNow()?
Reveal
shutdown() stops accepting new tasks and lets queued and running tasks finish. shutdownNow() also interrupts running tasks and returns the queued tasks that never started.
Your program's main finishes, but the JVM does not exit. You used an executor. Why?
Reveal
Executor threads are non-daemon by default, and they stay alive waiting for work until the executor is shut down.

6.1Why thread pools

Studied

Try it first2 minutes

A server starts a new thread for every request. Traffic spikes to 5,000 concurrent requests. Name two things that go wrong.

Reveal the answer

Memory: every platform thread reserves its own stack, and thousands of them can fail with OutOfMemoryError: unable to create native thread. CPU: thousands of runnable threads spend their time context switching. Also, creating and destroying a thread per request costs time, and there is no limit protecting the database behind it.

Idea

A thread pool keeps a fixed set of worker threads and a queue of tasks. You submit tasks; idle workers take them from the queue. Threads are reused, and the pool size caps how much runs at once.

Analogy

A call centre with 20 agents and a hold queue. Callers do not each get a new agent hired on the spot; they wait on hold, and an agent who finishes a call takes the next one. When the hold queue is full, new callers hear "please call later".

Cost

You must choose sizes, a queue and a rejection policy, and you must shut the pool down. Every choice is a trade-off between throughput, latency and memory.

Figure 1A pool turns unlimited concurrency into a fixed number of workers plus a queue. The queue absorbs bursts; the pool size protects the CPU and everything downstream.

6.2Executor, ExecutorService and the Executors factories

Studied

Try it first2 minutes

Which interface would you accept as a constructor parameter in a class that only needs to run tasks: Executor or ExecutorService?

Reveal the answer

Executor, if it only calls execute. Depend on the smallest interface you need. Ask for ExecutorService when you need submit (futures) or lifecycle methods, and let the owner of the pool manage shutdown.

Interface	Adds	Key methods
`Executor`	Run a task somehow	`execute(Runnable)`
`ExecutorService`	Results and lifecycle	`submit`, `invokeAll`, `invokeAny`, `shutdown`, `shutdownNow`, `awaitTermination`, `close` (Java 19+)
`ScheduledExecutorService`	Delays and repeats	`schedule`, `scheduleAtFixedRate`, `scheduleWithFixedDelay`

The Executors factory methods are convenient, but read what they build before using one in a service:

Factory	What it really creates	Hidden risk
`newFixedThreadPool(n)`	n threads, unbounded `LinkedBlockingQueue`	Queue grows without limit under overload
`newSingleThreadExecutor()`	1 thread, unbounded queue	Same; useful when tasks must run one at a time, in order
`newCachedThreadPool()`	0 core, unlimited max threads, `SynchronousQueue`, idle threads die after 60 s	A burst creates thousands of threads
`newScheduledThreadPool(n)`	n threads, unbounded delay queue	Same queue risk; an exception thrown by a periodic task silently suppresses all its later runs
`newWorkStealingPool()`	A `ForkJoinPool` with one thread per core	For CPU-bound, divide-and-conquer work (Module 13)
`newVirtualThreadPerTaskExecutor()`	A new virtual thread for every task (Java 21)	Unlimited concurrency; limit access to scarce resources yourself (Module 13)

FirstPool.java

public class FirstPool {
    public static void main(String[] args) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(3);
        List<Future<String>> results = new ArrayList<>();
        for (int i = 1; i <= 5; i++) {
            int id = i;
            results.add(pool.submit(() -> "order " + id + " checked on " + Thread.currentThread().getName()));
        }
        for (Future<String> f : results) System.out.println(f.get());   // waits for each result in turn
        pool.shutdown();                                                // let the JVM exit
    }
}

Rule of thumb

In services, build ThreadPoolExecutor directly (or Spring's ThreadPoolTaskExecutor) with a bounded queue, named threads and an explicit rejection policy. Use the Executors shortcuts for tests, scripts and cases where you have checked the hidden risk does not apply.

Example: run a batch of checks with one deadline. invokeAll(tasks, timeout) waits for all of them, and cancels (interrupts) whatever is not done by the deadline:

OnboardingChecks.java

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

public class OnboardingChecks {
    static String check(String name, long ms) throws InterruptedException { Thread.sleep(ms); return name + " ok"; }

    public static void main(String[] args) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(3);
        List<Callable<String>> checks = List.of(
                () -> check("kyc", 50),
                () -> check("sanctions", 80),
                () -> check("credit bureau", 5_000));                          // too slow
        List<Future<String>> results = pool.invokeAll(checks, 500, TimeUnit.MILLISECONDS);
        for (Future<String> f : results) System.out.println(f.isCancelled() ? "timed out" : f.get());
        pool.shutdown();
    }
}

6.3ThreadPoolExecutor: how a task finds a thread

Studied

Try it first2 minutes

Core size 2, max size 4, queue capacity 3. Ten long tasks arrive at once. Where does each one go?

Reveal the answer

Tasks 1 and 2 start core threads. Tasks 3, 4 and 5 fill the queue. Tasks 6 and 7 start two extra threads (the pool reaches 4). Tasks 8, 9 and 10 are rejected. Check it in the simulator below.

A production-shaped pool

ThreadPoolExecutor pool = new ThreadPoolExecutor(
        4,                                   // corePoolSize: threads kept even when idle
        16,                                  // maximumPoolSize: hard upper limit
        30, TimeUnit.SECONDS,                // keepAliveTime: extra (non-core) threads die after idling this long
        new ArrayBlockingQueue<>(200),       // workQueue: BOUNDED
        Thread.ofPlatform().name("payout-worker-", 0).factory(),   // named threads (Java 21 builder)
        new ThreadPoolExecutor.CallerRunsPolicy());                 // what to do when everything is full

Figure 2The order ThreadPoolExecutor.execute follows for every task. Most people expect "add threads up to max, then queue". The real order is core, queue, then extra threads, then reject.

Parameter	Meaning
`corePoolSize`	Threads the pool keeps alive even when idle. A new task starts a new core thread while fewer than this many exist, even if others are idle
`maximumPoolSize`	The most threads the pool will ever have. Ignored in practice when the queue is unbounded, because the queue never fills
`keepAliveTime`	How long a thread above the core size may stay idle before it exits. `allowCoreThreadTimeOut(true)` applies it to core threads too
`workQueue`	Where tasks wait when all core threads are busy
`threadFactory`	Creates threads: set names, daemon flag, uncaught exception handler
`handler`	The `RejectedExecutionHandler` called when the queue is full and the pool is at max, or for any task submitted after shutdown

6.4Work queues and rejection policies

Studied

Try it first2 minutes

A pool calls a payment gateway. Under a traffic spike the queue fills. Which rejection policy slows down the code that is submitting tasks instead of failing or dropping work?

Reveal the answer

CallerRunsPolicy. The submitting thread runs the task itself, so it cannot submit more until it finishes. That pushes the slowdown back to the source (backpressure). For a request thread it means that request becomes slower rather than failed.

Queues (all implement BlockingQueue, Module 9):

Queue	Behaviour	Use when
`ArrayBlockingQueue(n)`	Bounded, fixed array	The default choice for a bounded pool
`LinkedBlockingQueue(n)`	Bounded if you pass a capacity; unbounded without one	Bounded variant is fine; the no-argument form is the classic memory leak
`SynchronousQueue`	Zero capacity: a hand-off that succeeds only if a thread is idle and waiting	You want new threads (up to max) instead of queueing, as in the cached pool
`PriorityBlockingQueue`	Unbounded, ordered by priority	Rarely in pools; low-priority tasks can starve

Rejection policies:

Policy	When the pool is full	Good for
`AbortPolicy` (default)	Throws `RejectedExecutionException` to the submitter	Fail fast; turn it into HTTP 429 or 503
`CallerRunsPolicy`	The submitting thread runs the task (after `shutdown()` it silently drops it instead)	Backpressure, when the caller can afford to slow down
`DiscardPolicy`	Drops the task silently	Almost never: invisible data loss
`DiscardOldestPolicy`	Drops the oldest queued task, then retries (after `shutdown()` it drops the new task)	Only when newer work replaces older (latest price updates)

Watch out

A bounded queue without a plan for rejection just moves the problem. Decide what the user sees when the pool is full: a quick 503 with a retry hint, a slower response, or work deferred to Kafka. Then count rejections in your metrics.

6.5execute() vs submit()

Studied

Try it first2 minutes

Both pool.execute(task) and pool.submit(task) run a task that throws. Which one prints a stack trace, and which one stays silent?

Reveal the answer

execute lets the exception escape to the worker thread, so the uncaught exception handler prints it (and the pool replaces that worker). submit catches it and stores it in the Future; it stays silent unless someone calls get().

ExecuteVsSubmit.java

public class ExecuteVsSubmit {
    public static void main(String[] args) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(1);

        pool.execute(() -> { throw new IllegalStateException("boom in execute"); }); // printed by the uncaught handler

        Future<?> f = pool.submit(() -> { throw new IllegalStateException("boom in submit"); }); // silent
        Thread.sleep(200);
        System.out.println("nothing printed for submit yet");
        try {
            f.get();
        } catch (ExecutionException e) {
            System.out.println("get() threw: " + e.getCause());     // the original exception is the cause
        }
        pool.shutdown();
    }
}

	`execute(Runnable)`	`submit(Runnable or Callable)`
Defined on	`Executor`	`ExecutorService`
Returns	`void`	`Future`
Exception in the task	Reaches the thread's uncaught exception handler; that worker dies and is replaced (on a `ScheduledThreadPoolExecutor`, even `execute` wraps the task in a hidden future, so the exception is swallowed there too)	Stored in the `Future`; rethrown wrapped in `ExecutionException` by `get()`
Use when	Fire and forget, with your own try/catch and logging inside the task	You need the result, the failure, or the ability to cancel

In production

The most common lost error in Java services: submit(...) with the Future thrown away. Either keep and check the future, use execute with logging inside the task, or use CompletableFuture with an explicit exceptionally or whenComplete (Module 7).

Example: a production-shaped thread factory. Names make dumps readable; the handler sends execute() failures to your logging or alerting instead of only stderr:

NamedPool.java

import java.util.concurrent.*;

public class NamedPool {
    public static void main(String[] args) throws InterruptedException {
        ThreadFactory factory = Thread.ofPlatform()
                .name("statement-gen-", 1)                                   // statement-gen-1, -2, ...
                .uncaughtExceptionHandler((t, e) -> System.out.println("ALERT " + t.getName() + ": " + e.getMessage()))
                .factory();
        ExecutorService pool = new ThreadPoolExecutor(2, 2, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(50), factory);

        pool.execute(() -> System.out.println("running on " + Thread.currentThread().getName()));
        pool.execute(() -> { throw new IllegalStateException("template missing"); });   // execute: reaches the handler
        pool.shutdown();
        pool.awaitTermination(5, TimeUnit.SECONDS);
    }
}

6.6Shutting down

Studied

Try it first2 minutes

A pool has 3 running tasks and 20 queued. Describe what happens after shutdown(), and after shutdownNow().

Reveal the answer

shutdown(): no new tasks are accepted; the 3 running and the 20 queued all run to completion, then the threads exit. shutdownNow(): no new tasks; the 3 running tasks are interrupted; the 20 queued tasks are removed and returned as a list, never run.

Figure 3Both calls return immediately. Neither waits; use awaitTermination to wait.

The standard shutdown pattern (adapted from the ExecutorService Javadoc)

static void shutdownGracefully(ExecutorService pool, Duration timeout) {
    pool.shutdown();                                  // stop accepting work, finish what we have
    try {
        if (!pool.awaitTermination(timeout.toMillis(), TimeUnit.MILLISECONDS)) {
            List<Runnable> neverRan = pool.shutdownNow();     // interrupt the stragglers
            System.err.println(neverRan.size() + " queued tasks dropped");
            if (!pool.awaitTermination(timeout.toMillis(), TimeUnit.MILLISECONDS))
                System.err.println("pool did not terminate");
        }
    } catch (InterruptedException e) {
        pool.shutdownNow();
        Thread.currentThread().interrupt();           // preserve the interrupt
    }
}

Since Java 19, ExecutorService is AutoCloseable. close() calls shutdown() and then waits until all tasks finish, with no timeout. If the waiting thread is interrupted, it calls shutdownNow() once but keeps waiting for termination, then restores the interrupt status, so a task that ignores interrupts still hangs it. Great for short scoped work in try-with-resources; risky for long-lived pools where one stuck task would hang shutdown forever.

Scoped pool with try-with-resources (Java 19+)

try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
    for (String file : List.of("a.csv", "b.csv", "c.csv")) pool.submit(() -> parse(file));
}   // close(): waits here until every parse has finished

Figure 4The five run states of ThreadPoolExecutor (described in the comments of its source code, next to the ctl field). After either shutdown call, execute rejects new tasks; awaitTermination returns once the pool reaches TERMINATED.

6.7Scheduled tasks

Studied

Try it first2 minutes

A job runs every 10 seconds with scheduleAtFixedRate. One run takes 25 seconds. Do runs overlap? And what if one run throws an exception?

Reveal the answer

They do not overlap: the next run starts late, right after the slow one finishes. If a run throws, all later runs are suppressed: the task's ScheduledFuture completes exceptionally and nothing logs it. Catch exceptions inside periodic tasks.

Figure 5Fixed rate keeps a clock (heartbeats, metrics). Fixed delay keeps a gap (polling a database or an SFTP folder, where runs should never pile up).

A periodic poller that survives failures

ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(
        Thread.ofPlatform().name("statement-poller").factory());

scheduler.scheduleWithFixedDelay(() -> {
    try {
        pollNewFiles();                 // an exception escaping here would suppress all later runs
    } catch (Exception e) {             // an Error (e.g. OutOfMemoryError) still stops the schedule
        log("poll failed, will retry next time: " + e);
    }
}, 0, 30, TimeUnit.SECONDS);

In production

With several pods, every pod runs its own scheduler, so a "once every 30 seconds" job runs N times. For jobs that must run once per cluster, use a database lock (for example the ShedLock library), a Kubernetes CronJob, or a leader election.

6.8Sizing pools and handling overload

Studied

Try it first2 minutes

A pool on a 4-core machine runs tasks that each spend 90 ms waiting on HTTP and 10 ms on CPU. Roughly how many threads keep the CPU busy?

Reveal the answer

About 4 x (1 + 90 / 10) = 40 threads at full CPU use, or about 32 at an 80% target. While a thread waits, the core can run others, so waiting work needs many more threads than cores.

Two formulas cover most cases:

Threads for a pool (Brian Goetz, Java Concurrency in Practice): threads = cores x target CPU use x (1 + wait time / compute time). Pure CPU work gives about one thread per core (Goetz suggests cores + 1).
Little's law: requests in flight = arrival rate x time per request. At 200 requests/s and 100 ms each, 20 requests are in progress at any moment, so you need at least 20 threads (and maybe 20 database connections) to keep up.

Then make overload safe:

Bound everything: pool size, queue length, and how long a task may wait. A queue of 10,000 tasks at 50 ms each is about 50 seconds of waiting for the last task, even with ten workers.
Separate pools for separate work (bulkheads). A slow partner API should fill its own pool, not the pool your health check or payment path uses (Module 14).
Size for the bottleneck. A pool of 200 threads calling a database with 10 connections just builds a queue inside the connection pool.
Measure: active threads, queue size, task wait time, rejections. ThreadPoolExecutor exposes getActiveCount(), getQueue().size(), getCompletedTaskCount(); Micrometer can bind them to metrics.

6.9Build it yourself: a minimal thread pool

Studied

Try it first2 minutes

A thread pool is fixed worker threads plus a queue. Sketch the three methods: execute, the worker loop, and shutdown (queued tasks must still run). How do the workers learn it is time to exit?

Reveal the answer

execute offers the task to a bounded queue and rejects when it is full or shut down. Each worker loops on queue.take() and runs tasks, catching exceptions so it survives. shutdown stops new submissions, then puts one "poison pill" per worker behind the queued tasks; a worker that takes a pill exits.

MiniPool.java

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

public class MiniPool {
    private static final Runnable POISON = () -> { };              // tells one worker to exit
    private final BlockingQueue<Runnable> queue;
    private final List<Thread> workers = new ArrayList<>();
    private final Object lock = new Object();
    private boolean shutdown;                                       // guarded by lock

    public MiniPool(int threads, int capacity) {
        queue = new LinkedBlockingQueue<>(capacity);                // bounded
        for (int i = 0; i < threads; i++) {
            Thread w = new Thread(this::workLoop, "mini-worker-" + i);
            workers.add(w);
            w.start();
        }
    }

    public void execute(Runnable task) {
        synchronized (lock) {                                       // no task can slip in after shutdown()
            if (shutdown) throw new RejectedExecutionException("shut down");
            if (!queue.offer(task)) throw new RejectedExecutionException("queue full");   // like AbortPolicy
        }
    }

    private void workLoop() {
        try {
            while (true) {
                Runnable task = queue.take();                       // wait for work
                if (task == POISON) return;                         // one pill per worker
                try {
                    task.run();
                } catch (RuntimeException e) {                      // a failing task must not kill the worker
                    System.out.println(Thread.currentThread().getName() + ": task failed: " + e);
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();                     // stop at once, like shutdownNow()
        }
    }

    /** No new tasks; tasks already queued still run; then every worker exits. */
    public void shutdown() throws InterruptedException {
        synchronized (lock) { shutdown = true; }
        for (int i = 0; i < workers.size(); i++) queue.put(POISON); // pills go behind every queued task
    }

    public void awaitTermination() throws InterruptedException {
        for (Thread w : workers) w.join();
    }

    public static void main(String[] args) throws InterruptedException {
        MiniPool pool = new MiniPool(3, 100);
        AtomicInteger done = new AtomicInteger();
        for (int i = 0; i < 20; i++) pool.execute(() -> { sleep(10); done.incrementAndGet(); });
        pool.execute(() -> { throw new IllegalStateException("boom"); });
        pool.shutdown();
        pool.awaitTermination();
        System.out.println("terminated, tasks done: " + done.get());
        try { pool.execute(() -> { }); } catch (RejectedExecutionException e) { System.out.println("after shutdown: rejected"); }
    }

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

What ThreadPoolExecutor adds on top of this sketch:

`MiniPool`	`ThreadPoolExecutor`
Fixed thread count	Core and max sizes, threads created lazily, threads above core retired after `keepAliveTime`
Rejects when full	Pluggable `RejectedExecutionHandler`
A lock around submit and shutdown	One atomic `ctl` field holding the run state and worker count, updated with CAS
Poison pills	`shutdown()` and `shutdownNow()` with interruption of idle or all workers
Nothing	`Future`s through `submit`, hooks `beforeExecute` / `afterExecute` / `terminated`, statistics

6.10Use cases: which executor for which job

Studied

Try it first2 minutes

A service generates PDF statements (CPU-heavy, 200 ms of CPU each) and also calls an SMS partner (mostly waiting, 300 ms). Should they share one pool?

Reveal the answer

No. Give the PDF work a pool of about one thread per core, and the SMS calls their own bounded I/O pool sized from the wait/compute ratio. Separate pools also mean a slow SMS partner cannot delay statements (a bulkhead).

Calls to a partner APIThreadPoolExecutor, bounded queue

Sized from latency (lesson 6.8); its own pool doubles as a bulkhead.

CPU-heavy work: PDFs, hashing, parsingfixed pool of about cores threads

More threads than cores only adds context switches.

Events that must run in order per accountN single-thread lanes, lane chosen by key hash

Same key, same lane, same order (Module 14).

Polling an SFTP folder or an outbox tableScheduledExecutorService, fixed delay

Runs never pile up; catch exceptions inside.

Fire-and-forget notificationsbounded @Async executor

Decide the full-queue behaviour: CallerRuns, or drop and count.

Thousands of blocking calls (Java 21)newVirtualThreadPerTaskExecutor + Semaphore

Cheap threads; the semaphore protects the dependency (Module 13).

Several steps of one request in paralleldedicated I/O pool + CompletableFuture

Module 7; never the common pool for blocking calls.

Short scoped batch in a script or testtry-with-resources ExecutorService

close() waits for everything (Java 19+).

Log or export these numbers for every pool

static String stats(ThreadPoolExecutor p) {
    return String.format("active=%d threads=%d largest=%d queued=%d completed=%d",
            p.getActiveCount(), p.getPoolSize(), p.getLargestPoolSize(), p.getQueue().size(), p.getCompletedTaskCount());
}

6.11Common mistakes

Studied

Common mistakes

newFixedThreadPool or newCachedThreadPool in a service without checking their unbounded queue or unbounded threads.
Expecting max threads before queueing. Extra threads start only after the queue is full.
Throwing away the Future from submit, and with it every exception.
Never shutting the pool down, so the JVM cannot exit or a redeploy leaks threads.
One shared pool for unrelated work, so one slow dependency starves everything.
Periodic tasks without try/catch: one exception and the job silently stops forever.
Unnamed pool threads (pool-7-thread-3) that make dumps and logs hard to read.
Sizing by guesswork. Start from the formulas, then load test and watch queue length and latency.

6.12Self-check

Studied

Self-checkFive more, after the lessons

In what order does ThreadPoolExecutor try to place a task?
Reveal
Start a core thread if fewer than core are running; else offer to the queue; else start an extra thread if below max; else reject.
What does CallerRunsPolicy do and why is it useful?
Reveal
The submitting thread runs the rejected task itself, which slows down submission: a simple form of backpressure.
Why can newFixedThreadPool cause an OutOfMemoryError?
Reveal
Its LinkedBlockingQueue is unbounded, so under sustained overload tasks pile up in memory.
What does awaitTermination do?
Reveal
After a shutdown request, it blocks until all tasks finish, the timeout passes, or the caller is interrupted, and returns whether the pool terminated.
A job must run every 5 minutes, but runs must never overlap or pile up. Fixed rate or fixed delay?
Reveal
Fixed delay: each run starts a fixed time after the previous run ends. (Fixed rate also never overlaps, but late runs fire back to back.)

Key takeaways

Use pools, not a thread per task: they cap concurrency and reuse threads.
Know what Executors factories build: fixed and single pools have unbounded queues, cached pools unbounded threads.
ThreadPoolExecutor places a task: core thread, then queue, then extra thread, then rejection.
execute failures reach the uncaught handler; submit failures hide in the Future.
Shut down with shutdown, awaitTermination with a timeout, then shutdownNow.
Size from the work (cores, wait/compute, Little's law) and bound every queue.

Mastery checklist

Tick an item only when it is true with your notes closed.

Explain why services use thread pools instead of a thread per task
List what each Executors factory really builds and its hidden risk
Configure a ThreadPoolExecutor with a bounded queue, named threads and a rejection policy, and predict where a burst of tasks goes
Explain how execute and submit handle exceptions
Shut a pool down gracefully with shutdown, awaitTermination and shutdownNow
Size a pool for CPU-bound and I/O-bound work and explain what happens under overload

PreviousModule 5: wait, notify and interruption NextModule 7: Future and CompletableFuture

Phase 2: The java.util.concurrent toolkitCoreHighest priority

Module 7: Future and CompletableFuture

A Future is a handle to a result that is not ready yet. CompletableFuture lets you chain and combine such results without blocking: call three services in parallel, merge the answers, recover from failures, and enforce timeouts. This module covers both, with the threading rules that decide whether your pipeline is fast or quietly stuck.

12 lessons4 figures6 mastery checks

Warm-upWarm-up: one line each

future.get() is called before the task finishes. What happens?
Reveal
The calling thread blocks until the task completes (or fails, or is cancelled). get(2, SECONDS) blocks at most 2 seconds, then throws TimeoutException.
thenApply or thenCompose: which one do you use when the next step itself returns a CompletableFuture?
Reveal
thenCompose. With thenApply you get a nested CompletableFuture<CompletableFuture<T>>.
supplyAsync(task) with no executor argument. Which threads run it?
Reveal
The shared ForkJoinPool.commonPool(). On Java 21 to 24, if the common pool has fewer than 2 threads (as on a 1 or 2 CPU container), a new thread per task instead. From Java 25 it is always the common pool, which on 1 or 2 CPUs means a single thread.
Does completableFuture.cancel(true) interrupt the running task?
Reveal
No. It completes the future with a CancellationException, but the code that was computing the value keeps running.
Three independent calls take 100 ms, 300 ms and 200 ms. How long does the whole thing take sequentially, and with allOf?
Reveal
600 ms sequentially; about 300 ms in parallel, the time of the slowest call.

7.1Future: a handle to a result

Studied

Try it first2 minutes

You submit a report task and receive a Future<Report>. Name four things you can do with the future.

Reveal the answer

Wait for the result with get(); wait with a limit using get(timeout, unit); check isDone() without blocking; cancel it with cancel(true) (which interrupts the running task for executor futures). Also isCancelled(), and since Java 19 state() and resultNow().

Idea

submit() returns at once with a Future: a receipt for a result that will exist later. The caller keeps working and collects the result when it needs it.

Analogy

A dry-cleaning ticket. You leave the shirt, get a ticket, and go about your day. Later you hand in the ticket: if the shirt is ready you get it now, if not you wait at the counter.

Cost

get() blocks. A plain Future cannot say "when you are done, do this next", so composing several of them means blocking threads to wait.

FutureBasics.java

public class FutureBasics {
    public static void main(String[] args) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(2);
        Future<String> report = pool.submit(() -> { Thread.sleep(300); return "report ready"; });

        System.out.println("isDone right away? " + report.isDone());   // false
        try {
            report.get(100, TimeUnit.MILLISECONDS);                     // give up after 100 ms
        } catch (TimeoutException e) {
            System.out.println("not ready within 100 ms");
        }
        System.out.println(report.get());                              // now wait for it

        Future<String> stuck = pool.submit(() -> { Thread.sleep(10_000); return "never"; });
        stuck.cancel(true);                                             // if it already started, it is interrupted; otherwise it never runs
        System.out.println("cancelled? " + stuck.isCancelled());
        pool.shutdown();
    }
}

Exceptions thrown by the task come back from get() wrapped in ExecutionException; the original is getCause().
get() on a cancelled future throws CancellationException.
invokeAll(tasks) submits a list and waits for all of them; invokeAny(tasks) returns the first successful result and cancels the rest.

7.2Why CompletableFuture

Studied

Try it first2 minutes

A checkout calls the price service, then (with the price) the tax service, while also fetching the user's loyalty points. With plain Futures, how many threads sit blocked waiting?

Reveal the answer

At least one thread blocks on get() for the price before it can call tax, and one more blocks to join the results. Blocked threads do nothing useful. CompletableFuture registers the next step as a callback, so no thread waits.

Idea

CompletableFuture<T> is a Future you can attach steps to: "when the price arrives, compute tax; when both tax and points arrive, build the response; if anything fails, use a fallback." Each step runs when its input is ready.

Analogy

A restaurant buzzer instead of standing at the counter. You give instructions in advance ("when my order is ready, bring it to table 4"), and nobody waits idle.

Cost

Stack traces get harder to read, exceptions arrive wrapped, and it is easy to run callbacks on the wrong threads. The next lessons cover each trap.

Figure 1A pipeline of stages. Each stage starts when the previous one completes. A failure skips the normal stages and travels to the first error handler, like an exception jumping to catch.

7.3Starting async work: supplyAsync and runAsync

Studied

Try it first2 minutes

You want to start a database lookup that returns a Customer in the background. Which factory method do you use, and what should the second argument be?

Reveal the answer

CompletableFuture.supplyAsync(() -> repo.find(id), ioExecutor). supplyAsync is for work that returns a value (runAsync for void). Pass your own executor for blocking I/O so it does not run on the shared common pool.

Method	Task type	Result
`supplyAsync(supplier)` / `supplyAsync(supplier, executor)`	Returns a value	`CompletableFuture<T>`
`runAsync(runnable)` / `runAsync(runnable, executor)`	No value	`CompletableFuture<Void>`
`completedFuture(value)`	Already done	Useful in tests and for cache hits
`new CompletableFuture<>()` then `complete(v)` / `completeExceptionally(ex)`	You complete it yourself	Bridges callback APIs into futures

Without an executor argument, async methods run on ForkJoinPool.commonPool(), a JVM-wide pool with (cores - 1) threads. On Java 21 to 24, if that gives fewer than 2 threads, as on a container limited to 1 or 2 CPUs, Java instead creates a new thread for every task. From Java 25 it always uses the common pool, so on 1 or 2 CPUs all default async work shares one thread. Neither is what you want for blocking I/O (lesson 7.8).

7.4Transforming: thenApply, thenCompose, thenAccept

Studied

Try it first2 minutes

findUser(id) returns CompletableFuture<User> and loadOrders(user) returns CompletableFuture<List<Order>>. What type do you get from findUser(id).thenApply(this::loadOrders), and how do you fix it?

Reveal the answer

CompletableFuture<CompletableFuture<List<Order>>>, a future of a future. Use thenCompose(this::loadOrders), which flattens it to CompletableFuture<List<Order>>. It is the async version of flatMap.

Method	Your function	Use it to	Like
`thenApply(fn)`	`T -> U`	Transform the value	`map`
`thenCompose(fn)`	`T -> CompletableFuture<U>`	Start another async call that depends on this value	`flatMap`
`thenAccept(consumer)`	`T -> void`	Use the value at the end of a chain	`forEach`
`thenRun(runnable)`	nothing	Do something after, ignoring the value

Chaining stages

CompletableFuture<Receipt> receipt =
    CompletableFuture.supplyAsync(() -> cart.total(), ioPool)      // Money
        .thenApply(total -> total.plusPercent(18))                    // Money -> Money (sync, cheap)
        .thenCompose(gross -> paymentClient.charge(gross))            // Money -> CompletableFuture<Receipt>
        .thenApply(r -> r.withNote("paid"));                          // Receipt -> Receipt
receipt.thenAccept(r -> log("charged " + r));

Which thread runs a callback? This decides whether your pipeline is safe:

thenApply(fn) (no Async suffix) runs fn on the thread that completed the previous stage, or, if that stage was already complete when you called thenApply, on your current thread.
thenApplyAsync(fn) runs fn on the common pool; thenApplyAsync(fn, executor) runs it on your executor.

So cheap transformations can use the plain methods. Anything slow or blocking in a callback should use the Async form with an explicit executor, or it will run on whatever thread happened to finish the previous step, such as an HTTP client's I/O thread.

Figure 2Rule: if your function returns a future, use thenCompose.

Figure 3Which thread runs a non-async callback. You do not control it, so keep plain callbacks cheap and give slow ones an executor.

7.5Combining: thenCombine, allOf, anyOf

Studied

Try it first2 minutes

A quote page needs the price, the stock level and the delivery estimate from three services that do not depend on each other. How do you call them so the page waits only for the slowest?

Reveal the answer

Start all three with supplyAsync before waiting on any of them, then join them with allOf(...) (or thenCombine for two). The total time is the slowest call, not the sum.

Method	Waits for	Result
`a.thenCombine(b, (x, y) -> ...)`	Both `a` and `b`	Your function's result
`CompletableFuture.allOf(f1, f2, f3)`	All of them	`CompletableFuture<Void>`; then read each with `join()` (they are done, so it does not block)
`CompletableFuture.anyOf(f1, f2)`	The first to finish	`CompletableFuture<Object>` (you cast)

allOf waits for every future, even if one fails early, and then completes exceptionally if any of them failed.

QuotePage.java: fan out, then fan in

import java.util.concurrent.*;

public class QuotePage {
    static String call(String what, long ms) {
        try { Thread.sleep(ms); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
        return what;
    }

    public static void main(String[] args) {
        ExecutorService io = Executors.newFixedThreadPool(8);
        long start = System.nanoTime();

        CompletableFuture<String> price    = CompletableFuture.supplyAsync(() -> call("price 499", 100), io);
        CompletableFuture<String> stock    = CompletableFuture.supplyAsync(() -> call("stock 12", 300), io);
        CompletableFuture<String> delivery = CompletableFuture.supplyAsync(() -> call("delivery 2 days", 200), io);

        String page = CompletableFuture.allOf(price, stock, delivery)
                .thenApply(v -> price.join() + " | " + stock.join() + " | " + delivery.join())   // all done: join is instant
                .join();

        long ms = (System.nanoTime() - start) / 1_000_000;
        System.out.println(page + " in " + ms + " ms");   // about 300 (the slowest call), not 600
        io.shutdown();
    }
}

Figure 4Fan-out, fan-in. Start every independent call first, then wait. Calling join() right after each supplyAsync would make them sequential again.

7.6Handling errors

Studied

Try it first2 minutes

A stage in the middle of a chain throws. What happens to the stages after it, and what does join() at the end throw?

Reveal the answer

The following thenApply / thenCompose stages are skipped; each completes exceptionally with the same error. join() throws an unchecked CompletionException whose cause is the original exception (get() throws a checked ExecutionException instead).

Method	Runs when	Can change the result?	Use it to
`exceptionally(ex -> fallback)`	Only on failure	Yes, replaces the error with a value	Fallbacks and defaults
`handle((value, ex) -> ...)`	Always	Yes	Map success and failure into one result
`whenComplete((value, ex) -> ...)`	Always	No, passes the original outcome on	Logging, metrics, cleanup
`exceptionallyCompose(ex -> otherFuture)` (Java 12+)	Only on failure	Yes, with another async call	Retry or call a backup service

Fallbacks.java

public class Fallbacks {
    static int loyaltyPoints(String user) { throw new IllegalStateException("loyalty service down"); }

    public static void main(String[] args) {
        CompletableFuture<Integer> points = CompletableFuture
                .supplyAsync(() -> loyaltyPoints("u42"))
                .thenApply(p -> p + 100)                                 // skipped
                .whenComplete((v, ex) -> { if (ex != null) System.out.println("log: " + ex); })
                .exceptionally(ex -> {
                    Throwable cause = ex instanceof CompletionException ? ex.getCause() : ex;  // usually wrapped
                    System.out.println("fallback because: " + cause.getMessage());
                    return 0;                                            // default: no points
                });
        System.out.println("points = " + points.join());
    }
}

Output

log: java.util.concurrent.CompletionException: java.lang.IllegalStateException: loyalty service down
fallback because: loyalty service down
points = 0

Watch out

The exception your handler receives is usually a CompletionException wrapping the real one (it is unwrapped only when that same future was completed directly with completeExceptionally). Unwrap with getCause() before checking its type, or ex instanceof TimeoutException will be false when you expect true.

7.7Timeouts and cancellation

Studied

Try it first2 minutes

You add .orTimeout(2, SECONDS) to a call to a slow partner API. After 2 seconds the future fails with TimeoutException. Has the HTTP request been stopped?

Reveal the answer

No. orTimeout only completes the future; the code computing the value keeps running and keeps its thread busy until the call returns. Set a timeout on the HTTP client itself as well, so the underlying work also ends.

Method (Java 9+)	Effect
`orTimeout(t, unit)`	If not done in time, complete exceptionally with `TimeoutException`
`completeOnTimeout(value, t, unit)`	If not done in time, complete normally with `value` (a default)
`get(t, unit)`	Blocking wait with a limit; throws `TimeoutException` but leaves the future running
`cancel(mayInterrupt)`	Completes with `CancellationException`. For `CompletableFuture` the flag has no effect: the running code is not interrupted

Two timeouts: one for the user, one for the work

CompletableFuture<Quote> quote = CompletableFuture
        .supplyAsync(() -> partnerClient.quote(order), partnerPool)    // the client has its own 1.5 s read timeout
        .completeOnTimeout(Quote.UNAVAILABLE, 2, TimeUnit.SECONDS)      // the page never waits more than 2 s
        .exceptionally(ex -> Quote.UNAVAILABLE);                        // any failure: degrade, do not break the page

Key idea

Every remote call needs two limits: a timeout on the future (what the caller waits for) and a timeout on the client (how long the work itself may run). Without the second, timed-out calls keep piling up on your pool threads.

7.8Common pool or your own executor

Studied

Try it first2 minutes

A service runs blocking JDBC calls with supplyAsync(() -> repo.load(id)) and no executor. The machine has 8 cores. How many of those calls can run at once, and what else suffers?

Reveal the answer

About 7, the size of the common pool. Further calls queue behind them. Everything else that uses the common pool (parallel streams, other *Async calls without an executor) is starved while those threads sit blocked on the database.

Pass your own executor for anything that blocks (JDBC, HTTP, file I/O): supplyAsync(task, ioExecutor). Size it for the work (Module 6) and for the resource behind it (Module 14).
The common pool is for short CPU work. It is shared by the whole JVM, including parallel streams.
Never block inside a callback on a small pool. join() or get() inside a stage that runs on the same bounded pool can deadlock it: every thread waits for work that has no thread left to run on (Module 11).
Only block at the edge. In a web request, one join() at the end (with a timeout set upstream) is fine. Inside the pipeline, compose instead of waiting.
With virtual threads (Module 13), blocking calls become cheap and many teams write plain sequential code instead of long future chains. CompletableFuture remains the tool for fan-out and timeouts.

7.9Recipes you will use

Studied

Try it first2 minutes

CompletableFuture.anyOf(a, b, c) gives you the first future to finish. Why is that not the same as "the first successful result"?

Reveal the answer

anyOf completes with whichever future completes first, success or failure. If the fastest mirror fails at once, anyOf fails too. "First success" needs its own small helper (recipe 4 below).

Five patterns that cover most real CompletableFuture code, in one program you can run:

Recipes.java

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

public class Recipes {
    static final ExecutorService io = Executors.newFixedThreadPool(8);

    static <T> CompletableFuture<T> async(Supplier<T> s) { return CompletableFuture.supplyAsync(s, io); }
    static void sleep(long ms) { try { Thread.sleep(ms); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }

    // 1. Two independent calls, combined
    static CompletableFuture<String> priceWithTax(String sku) {
        CompletableFuture<Long> price = async(() -> { sleep(50); return 1000L; });
        CompletableFuture<Integer> taxPct = async(() -> { sleep(80); return 18; });
        return price.thenCombine(taxPct, (p, t) -> sku + " costs " + (p + p * t / 100));
    }

    // 2. Retry with exponential back-off, without blocking a thread while waiting
    static <T> CompletableFuture<T> retry(Supplier<CompletableFuture<T>> call, int attempts, long backoffMs) {
        return call.get().exceptionallyCompose(ex -> {
            if (attempts <= 1) return CompletableFuture.failedFuture(ex);
            Executor later = CompletableFuture.delayedExecutor(backoffMs, TimeUnit.MILLISECONDS, io);
            return CompletableFuture.runAsync(() -> { }, later)
                    .thenCompose(v -> retry(call, attempts - 1, backoffMs * 2));
        });
    }

    // 3. Many calls in parallel, collected into one list
    static <T, R> CompletableFuture<List<R>> forAll(List<T> items, Function<T, CompletableFuture<R>> fn) {
        List<CompletableFuture<R>> fs = items.stream().map(fn).toList();      // start them all first
        return CompletableFuture.allOf(fs.toArray(new CompletableFuture[0]))
                .thenApply(v -> fs.stream().map(CompletableFuture::join).toList());
    }

    // 4. First successful result (anyOf would also return the first FAILURE)
    static <T> CompletableFuture<T> firstSuccess(List<CompletableFuture<T>> fs) {
        CompletableFuture<T> result = new CompletableFuture<>();
        AtomicInteger failures = new AtomicInteger();
        for (CompletableFuture<T> f : fs) {
            f.whenComplete((v, ex) -> {
                if (ex == null) result.complete(v);                          // the first complete() wins
                else if (failures.incrementAndGet() == fs.size())            // fail only when every one failed
                    result.completeExceptionally(new IllegalStateException("all failed", ex));
            });
        }
        return result;
    }

    // 5. A callback-style API turned into a future
    interface Callback { void onSuccess(String v); void onError(Throwable t); }
    static void legacyLookup(String id, Callback cb) { io.execute(() -> cb.onSuccess("customer-" + id)); }
    static CompletableFuture<String> lookup(String id) {
        CompletableFuture<String> f = new CompletableFuture<>();
        legacyLookup(id, new Callback() {
            public void onSuccess(String v) { f.complete(v); }
            public void onError(Throwable t) { f.completeExceptionally(t); }
        });
        return f;
    }

    public static void main(String[] args) throws Exception {
        System.out.println("1. " + priceWithTax("card-fee").get(1, TimeUnit.SECONDS));

        AtomicInteger calls = new AtomicInteger();
        String r = retry(() -> async(() -> {
            if (calls.incrementAndGet() < 3) throw new IllegalStateException("503 from partner");
            return "ok";
        }), 4, 20).get(2, TimeUnit.SECONDS);
        System.out.println("2. " + r + " after " + calls.get() + " attempts");

        System.out.println("3. " + forAll(List.of(1, 2, 3), id -> async(() -> "txn-" + id)).get(1, TimeUnit.SECONDS));

        CompletableFuture<String> down = async(() -> { throw new IllegalStateException("mirror down"); });
        CompletableFuture<String> slow = async(() -> { sleep(100); return "mirror-b"; });
        CompletableFuture<String> fast = async(() -> { sleep(30); return "mirror-a"; });
        System.out.println("4. " + firstSuccess(List.of(down, slow, fast)).get(1, TimeUnit.SECONDS)
                + ", then without fast: " + firstSuccess(List.of(down, slow)).get(1, TimeUnit.SECONDS));

        System.out.println("5. " + lookup("42").get(1, TimeUnit.SECONDS));
        io.shutdown();
    }
}

Recipe 6: handle results in the order they finish. When you do not need everything at once (sending each statement as soon as it is generated), ExecutorCompletionService hands back futures in completion order:

AsTheyFinish.java

import java.util.concurrent.*;

public class AsTheyFinish {
    public static void main(String[] args) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(3);
        CompletionService<String> done = new ExecutorCompletionService<>(pool);
        done.submit(() -> { Thread.sleep(300); return "slow statement"; });
        done.submit(() -> { Thread.sleep(50);  return "fast statement"; });
        done.submit(() -> { Thread.sleep(150); return "medium statement"; });
        for (int i = 0; i < 3; i++) System.out.println("send: " + done.take().get());   // completion order
        pool.shutdown();
    }
}

7.10Use cases: CompletableFuture in services

Studied

Try it first2 minutes

For each, pick the method: (a) call price and stock at the same time and merge; (b) call price, then use it to call tax; (c) show offers only if they arrive within 150 ms.

Reveal the answer

(a) thenCombine (or allOf for more than two). (b) thenCompose. (c) completeOnTimeout(defaultOffers, 150, MILLISECONDS) plus exceptionally for failures.

Account overview from independent servicessupplyAsync x N + allOf

Total time is the slowest call (lesson 7.5).

Price, then tax computed from the pricethenCompose

The second call needs the first result.

Optional enrichment: rewards, offerscompleteOnTimeout + exceptionally

The page never waits long and never breaks for an extra.

A flaky partner APIexceptionallyCompose + delayedExecutor

Recipe 2: retry with back-off, no blocked thread.

Send each result as soon as it is readyExecutorCompletionService

Recipe 6.

Fire and forget, but never silentlyrunAsync(task, pool).whenComplete(log)

Every failure reaches the logs.

Wrap a callback or listener APInew CompletableFuture + complete()

Recipe 5.

7.11Common mistakes

Studied

Common mistakes

Blocking right after starting: supplyAsync(a).join(); supplyAsync(b).join(); is sequential code with extra steps.
Blocking I/O on the common pool by omitting the executor argument.
thenApply with a function that returns a future, giving a nested future. Use thenCompose.
No error handler: a failed future nobody reads loses the exception.
Checking ex instanceof SomeException without unwrapping CompletionException.
Assuming orTimeout or cancel stops the work. They only complete the future.
join() inside callbacks on a bounded pool, which can exhaust or deadlock it.

7.12Self-check

Studied

Self-checkFive more, after the lessons

What does get() throw when the task failed, and how do you reach the original exception?
Reveal
ExecutionException; the original is getCause().
Which thread runs a non-async thenApply callback?
Reveal
The thread that completed the previous stage, or the caller's thread if that stage was already complete when thenApply was called.
When does allOf complete if one of three futures fails quickly?
Reveal
Only after all three have completed; then it completes exceptionally with a CompletionException caused by the failure.
Difference between handle and whenComplete?
Reveal
Both run on success and failure; handle returns a new result, whenComplete cannot change the outcome and passes it on.
Why pass an executor to supplyAsync for a database call?
Reveal
The common pool is small, shared JVM-wide and meant for CPU work; blocking calls would starve it. A dedicated pool can be sized for the I/O and for the connection pool.

Key takeaways

A Future is a receipt; get() blocks, get(timeout) blocks with a limit.
CompletableFuture chains steps: thenApply maps, thenCompose chains another call, thenCombine / allOf join.
Start every independent call before waiting on any of them.
Failures skip normal stages; recover with exceptionally or handle, observe with whenComplete, and unwrap CompletionException.
orTimeout, completeOnTimeout and cancel complete the future but do not stop the work: set client timeouts too.
Pass your own executor for anything that blocks; keep the common pool for short CPU work.

Mastery checklist

Tick an item only when it is true with your notes closed.

Use Future with get(timeout), isDone and cancel, and unwrap ExecutionException
Build a pipeline with supplyAsync, thenApply, thenCompose and thenAccept, and explain which thread runs each callback
Call three independent services in parallel with allOf and explain why the total time is the slowest call
Recover from failures with exceptionally, handle and whenComplete, unwrapping CompletionException
Add a user-facing timeout and a client timeout, and explain why orTimeout and cancel do not stop the work
Choose between the common pool and a custom executor, and avoid blocking inside callbacks

PreviousModule 6: Executors and thread pools NextModule 8: Atomics and CAS

Phase 2: The java.util.concurrent toolkitCoreHighest priority

Module 8: Atomic classes and compare-and-set

Atomic classes such as AtomicInteger and AtomicReference update a single variable safely without a lock, using a CPU instruction called compare-and-set (CAS). This module covers how CAS works, the atomic API, LongAdder for hot counters, and the limit everyone hits: two atomic variables do not make one atomic operation.

9 lessons3 figures6 mastery checks

Warm-upWarm-up: one line each

Is atomicCount.incrementAndGet() safe when called by 50 threads at once?
Reveal
Yes. It is a single atomic read-modify-write; no update is lost and every caller gets a distinct new value.
What does compareAndSet(5, 6) do if the current value is 7?
Reveal
Nothing changes and it returns false. It sets the new value only if the current value still equals the expected one.
Why should the function passed to updateAndGet have no side effects?
Reveal
Under contention it may be called several times: each failed CAS retries with a fresh value and calls the function again.
A metrics counter is incremented by every request thread and read once a minute. AtomicLong or LongAdder?
Reveal
LongAdder. It spreads updates over several cells so threads rarely collide; reads are rare, so the slower sum() does not matter.
A range has AtomicInteger lower and AtomicInteger upper with the rule lower <= upper. Is the range thread-safe?
Reveal
No. Each variable is atomic, but a check of one and an update of the other are two steps; two threads can each pass their check and together break the rule. Use one lock, or one AtomicReference to an immutable range.

8.1Atomic counters

Studied

Try it first2 minutes

Rewrite the racy count++ from Module 2 without synchronized.

Reveal the answer

Use AtomicInteger count = new AtomicInteger(); and call count.incrementAndGet() (or getAndIncrement()). The class performs read-add-write as one atomic step.

AtomicCounter.java

import java.util.concurrent.atomic.AtomicInteger;

public class AtomicCounter {
    public static void main(String[] args) throws InterruptedException {
        AtomicInteger count = new AtomicInteger();
        Runnable work = () -> { for (int i = 0; i < 100_000; i++) count.incrementAndGet(); };
        Thread a = new Thread(work), b = new Thread(work);
        a.start(); b.start();
        a.join(); b.join();
        System.out.println("count = " + count.get());    // always 200000, no lock used
    }
}

Method	Returns	Note
`get()`, `set(v)`	current value	Behave like a volatile read and write
`incrementAndGet()`, `decrementAndGet()`	the new value	`++x`
`getAndIncrement()`, `getAndDecrement()`	the old value	`x++`
`addAndGet(d)`, `getAndAdd(d)`	new or old value	Add any amount
`compareAndSet(expected, new)`	`true` if it changed	The building block for everything else
`updateAndGet(fn)`, `getAndUpdate(fn)`	new or old value	Apply any function atomically
`accumulateAndGet(x, fn)`	the new value	Combine with an argument, for example `Math::max`

The family: AtomicInteger, AtomicLong, AtomicBoolean, AtomicReference<V>, plus array versions (AtomicIntegerArray, ...). Every atomic has volatile visibility: a set in one thread is seen by get in another.

Example: run something exactly once. AtomicBoolean.compareAndSet(false, true) succeeds for one caller only:

StartOnce.java

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

public class StartOnce {
    private final AtomicBoolean started = new AtomicBoolean(false);
    final AtomicInteger starts = new AtomicInteger();

    void ensureStarted() {
        if (started.compareAndSet(false, true)) {        // only the first caller gets true
            starts.incrementAndGet();                     // start the consumer, open the connection, ...
        }
    }

    public static void main(String[] args) throws InterruptedException {
        StartOnce s = new StartOnce();
        ExecutorService pool = Executors.newFixedThreadPool(8);
        for (int i = 0; i < 100; i++) pool.execute(s::ensureStarted);
        pool.shutdown();
        pool.awaitTermination(5, TimeUnit.SECONDS);
        System.out.println("started " + s.starts.get() + " time(s)");
    }
}

Watch out

Other callers return immediately, possibly before the winner has finished starting. If they must wait until the work is done, use a CountDownLatch or a CompletableFuture that the winner completes.

8.2How compare-and-set works

Studied

Try it first2 minutes

A shared document has a version number. You read version 7, edit locally, and try to save. The server accepts a save only if the version is still 7. What do you do if it rejects your save?

Reveal the answer

Re-read the latest version, apply your change again on top of it, and try to save again. That loop, read, compute, conditional write, retry, is exactly how CAS-based atomics work. Databases call it optimistic locking.

Idea

Compare-and-set (CAS) is one CPU instruction: "if the value at this address is still expected, replace it with new; tell me whether it worked". No thread ever blocks. A thread that loses the race simply retries with the fresh value.

Analogy

Saving a shared document only if nobody saved since you opened it. If someone did, you merge onto their version and save again.

Cost

Under heavy contention many threads retry over and over, wasting CPU. CAS also works on one variable at a time only.

Figure 1The same race as the lost update in Module 2, now with CAS. B's first attempt fails because the value changed under it, so B retries from 6.

What incrementAndGet does, written out as a CAS loop

class CasCounter {
    private final AtomicInteger value = new AtomicInteger();

    int increment() {
        while (true) {
            int current = value.get();                 // 1. read
            int next = current + 1;                    // 2. compute
            if (value.compareAndSet(current, next)) {  // 3. write only if unchanged
                return next;
            }                                          // 4. lost the race: loop and retry
        }
    }
}

On x86 the CAS is the LOCK CMPXCHG instruction; ARM has its own equivalents. Because nothing blocks, atomics never deadlock and a thread descheduled mid-update cannot hold others up.

8.3updateAndGet, accumulateAndGet and AtomicReference

Studied

Try it first2 minutes

You track the highest transaction amount seen so far from many threads. Which single atomic call does it?

Reveal the answer

maxAmount.accumulateAndGet(amount, Math::max). It repeatedly reads the current maximum, computes max(current, amount) and CASes it in until it succeeds.

Functions applied atomically

class Stats {
    private final AtomicLong maxAmount = new AtomicLong(Long.MIN_VALUE);
    private final AtomicInteger inFlight = new AtomicInteger();

    void record(long amount) {
        maxAmount.accumulateAndGet(amount, Math::max);   // atomic "keep the larger"
    }
    boolean tryStart(int limit) {
        // atomic "increment, but never above limit"; returns false when full
        int before = inFlight.getAndUpdate(n -> n < limit ? n + 1 : n);
        return before < limit;
    }
    void finish() { inFlight.decrementAndGet(); }
}

Watch out

The function passed to updateAndGet, getAndUpdate or accumulateAndGet may run more than once when threads collide. Keep it pure: no logging, no I/O, no changes to other state.

AtomicReference<V> applies CAS to a reference. Combine it with an immutable object and you can update several fields as one atomic step: build a new object from the current one and swap it in.

A lock-free rate limiter: two fields updated as one

final class FixedWindowLimiter {
    private record Window(long startMillis, int count) { }      // immutable state

    private final AtomicReference<Window> window = new AtomicReference<>(new Window(0, 0));
    private final int limit;
    private final long windowMillis;

    FixedWindowLimiter(int limit, long windowMillis) { this.limit = limit; this.windowMillis = windowMillis; }

    boolean tryAcquire() {
        while (true) {
            long now = System.currentTimeMillis();
            Window cur = window.get();
            Window next;
            if (now - cur.startMillis() >= windowMillis) next = new Window(now, 1);       // new window
            else if (cur.count() < limit)                next = new Window(cur.startMillis(), cur.count() + 1);
            else                                         return false;                     // over the limit
            if (window.compareAndSet(cur, next)) return true;   // both fields changed together
        }
    }
}

compareAndSet on an AtomicReference compares with == (same object), not equals(). That is exactly right here, because cur is the object we just read.

Example: a state machine where only one transition can win. A card delivery is OUT_FOR_DELIVERY. The courier webhook says DELIVERED while a timeout job says UNDELIVERED, at the same moment. With compareAndSet exactly one of them wins; the loser gets false and reads the current state with get() (or uses compareAndExchange, Java 9+, which returns the value it found in the same atomic step):

Figure 2A CAS-guarded transition: "move from the state I saw to the next state, only if nobody moved it first". The same idea as UPDATE ... WHERE status = 'OUT_FOR_DELIVERY' in a database.

DeliveryStatus.java

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

public class DeliveryStatus {
    enum Status { DISPATCHED, OUT_FOR_DELIVERY, DELIVERED, UNDELIVERED }

    private static final Map<Status, Set<Status>> ALLOWED = Map.of(
            Status.DISPATCHED, Set.of(Status.OUT_FOR_DELIVERY),
            Status.OUT_FOR_DELIVERY, Set.of(Status.DELIVERED, Status.UNDELIVERED),
            Status.DELIVERED, Set.of(),
            Status.UNDELIVERED, Set.of(Status.OUT_FOR_DELIVERY));

    private final AtomicReference<Status> status = new AtomicReference<>(Status.OUT_FOR_DELIVERY);

    /** Moves from 'from' to 'to' only if the status is still 'from' and the move is legal. */
    boolean transition(Status from, Status to) {
        if (!ALLOWED.get(from).contains(to)) throw new IllegalArgumentException(from + " -> " + to);
        return status.compareAndSet(from, to);       // enums are singletons, so == comparison is exactly right
    }

    public static void main(String[] args) throws Exception {
        DeliveryStatus d = new DeliveryStatus();
        ExecutorService pool = Executors.newFixedThreadPool(8);
        CountDownLatch start = new CountDownLatch(1);
        AtomicInteger winners = new AtomicInteger();
        for (int i = 0; i < 8; i++) {
            Status target = i % 2 == 0 ? Status.DELIVERED : Status.UNDELIVERED;
            pool.execute(() -> {
                try { start.await(); } catch (InterruptedException e) { return; }
                if (d.transition(Status.OUT_FOR_DELIVERY, target)) winners.incrementAndGet();
            });
        }
        start.countDown();
        pool.shutdown();
        pool.awaitTermination(5, TimeUnit.SECONDS);
        System.out.println("winners = " + winners.get() + ", final status = " + d.status.get());
    }
}

8.4LongAdder for hot counters

Studied

Try it first2 minutes

Sixty-four request threads increment one AtomicLong on every request. Why can that become slow, even though nothing blocks?

Reveal the answer

Every increment is a CAS on the same memory location. Cores fight over the same cache line, CASes fail and retry, and throughput drops. LongAdder gives each colliding thread its own cell to update.

Figure 3LongAdder trades a cheap write for a more expensive read. Perfect for counters that are written constantly and read occasionally.

Counters for metrics

class ApiMetrics {
    private final LongAdder requests = new LongAdder();
    private final LongAdder failures = new LongAdder();

    void onRequest(boolean ok) {
        requests.increment();          // cheap even with many threads
        if (!ok) failures.increment();
    }
    String snapshot() {
        return requests.sum() + " requests, " + failures.sum() + " failures";   // approximate under load
    }
}

sum() is not an atomic snapshot: increments that happen while it adds the cells may or may not be counted. Fine for metrics, wrong for anything that needs an exact value at a point in time (use AtomicLong or a lock).
There is no incrementAndGet; LongAdder cannot hand out unique sequence numbers.
LongAccumulator is the general form, for example new LongAccumulator(Math::max, Long.MIN_VALUE).

Go deeper

CPUs move memory in 64-byte cache lines. If two hot variables written by different threads sit in the same line, each write invalidates the line in the other core's cache even though the threads never touch each other's variable: false sharing. LongAdder avoids it by padding its cells (they are marked @Contended inside the JDK), so neighbouring cells live in different lines. You rarely need to do this yourself; if a profiler shows it, separate the fields or use LongAdder.

8.5Choosing volatile, atomic or lock

Studied

Try it first2 minutes

Is this class thread-safe?

java class Range { private final AtomicInteger lower = new AtomicInteger(0), upper = new AtomicInteger(10); void setLower(int v) { if (v > upper.get()) throw new IllegalArgumentException(); lower.set(v); } void setUpper(int v) { if (v < lower.get()) throw new IllegalArgumentException(); upper.set(v); } }

Reveal the answer

No. With the range (0, 10), thread 1 calls setLower(5) while thread 2 calls setUpper(4). Each check passes against the old values, both writes happen, and the range becomes (5, 4). The invariant links two variables, so both must change under one lock, or live together in one AtomicReference<Range>.

You need	Use
One flag or reference, written independently of its old value	`volatile`
One counter, sequence number or single value updated from its current value	`AtomicInteger` / `AtomicLong` / `AtomicReference`
A counter hammered by many threads and read rarely	`LongAdder`
Several variables that must change together, or a check plus an action across them	A lock (or `AtomicReference` to an immutable holder)
Waiting until some condition becomes true	A lock with a condition, or a higher-level tool (Modules 9 and 10)

Key idea

Atomic parts do not make an atomic whole. ConcurrentHashMap plus AtomicLong plus volatile in one method still needs a design for the steps between them.

Go deeper

The ABA problem

CAS checks that the value is the same, not that nothing happened. Thread 1 reads A; thread 2 changes it to B and back to A; thread 1's CAS succeeds as if nothing changed. For counters this is harmless. It matters for lock-free data structures that reuse nodes (a popped node pushed back). AtomicStampedReference pairs the reference with a version stamp so the CAS fails if anything changed in between. Day-to-day service code rarely needs it.

8.6Build it yourself: a spin lock with CAS

Studied

Try it first2 minutes

Using only AtomicBoolean, how would you build a lock? What does a thread do while it waits, and why is that a problem with many threads?

Reveal the answer

lock() loops on compareAndSet(false, true) until it wins; unlock() sets the flag back to false. A waiting thread spins on a CPU instead of sleeping, which wastes a core and can starve the lock holder of CPU time. That is why real locks park waiting threads (Module 10).

SpinLock.java

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

public class SpinLock {
    private final AtomicBoolean locked = new AtomicBoolean(false);

    public void lock() {
        while (!locked.compareAndSet(false, true)) {
            Thread.onSpinWait();                  // hint to the CPU that we are busy-waiting
        }
    }
    public void unlock() { locked.set(false); }   // a volatile write: publishes the critical section's writes

    static int counter = 0;                        // guarded by the spin lock

    public static void main(String[] args) throws InterruptedException {
        SpinLock lock = new SpinLock();
        ExecutorService pool = Executors.newFixedThreadPool(4);
        for (int t = 0; t < 4; t++) {
            pool.execute(() -> {
                for (int i = 0; i < 10_000; i++) {
                    lock.lock();
                    try { counter++; } finally { lock.unlock(); }
                }
            });
        }
        pool.shutdown();
        pool.awaitTermination(10, TimeUnit.SECONDS);
        System.out.println("counter = " + counter);   // 40000
    }
}

It works because a successful CAS has the effect of a volatile read and write, and unlock() is a volatile write: together they create the happens-before edge a lock needs. What it lacks, compared with ReentrantLock: waiting threads sleep instead of spin, re-entry by the same thread (this one deadlocks on itself), fairness, timeouts and interruptible waits. Spin locks only make sense for critical sections of a few instructions, inside libraries.

8.7Use cases: atomics in services

Studied

Try it first2 minutes

Your service needs unique order numbers. Inside one JVM AtomicLong.incrementAndGet() works. Why does it fail once the service runs on 4 pods?

Reveal the answer

Each pod has its own AtomicLong, so the pods hand out the same numbers. Cluster-wide IDs need a shared source: a database sequence, a UUID, or an ID scheme that includes a node ID (Snowflake style).

Request and error countersLongAdder

Many writers, rare reads.

Sequence numbers inside one JVMAtomicLong.incrementAndGet

Unique per process; across pods use a DB sequence or UUIDs.

Start a consumer or connection onceAtomicBoolean.compareAndSet

The StartOnce example.

State transitions that only one thread may winAtomicReference + compareAndSet

The delivery state machine.

Peak value tracking (max latency)accumulateAndGet(x, Math::max)

Or LongAccumulator for hot paths.

In-flight request limitgetAndUpdate with a bound

Or simply a Semaphore (Module 10).

Several fields updated togetherAtomicReference to an immutable record

The rate limiter in lesson 8.3.

8.8Common mistakes

Studied

Common mistakes

get() then set() on an atomic, which is check-then-act again. Use compareAndSet or updateAndGet.
Side effects inside updateAndGet functions, which can run several times.
Several atomics for one invariant. They do not change together.
LongAdder.sum() used as an exact value at a point in time.
Spinning CAS loops under extreme contention on one variable, burning CPU. Reduce sharing (LongAdder, sharding) or use a lock.
Expecting AtomicReference.compareAndSet to compare with equals(). It compares identity.

8.9Self-check

Studied

Self-checkFive more, after the lessons

What does compareAndSet(expected, new) guarantee?
Reveal
It atomically sets the new value only if the current value equals expected, and returns whether it did.
Why do atomics never deadlock?
Reveal
No thread holds a lock or waits for another; a thread that loses a CAS just retries.
When is LongAdder better than AtomicLong?
Reveal
Many threads update the counter often and reads are rare; LongAdder avoids contention on one memory location.
How do you update two related fields atomically without a lock?
Reveal
Put them in an immutable object held by an AtomicReference, build a new object from the current one, and compareAndSet it in a retry loop.
What is the ABA problem?
Reveal
A CAS succeeds because the value is back to what was read, even though it changed in between. AtomicStampedReference detects it with a version stamp.

Key takeaways

Atomic classes update one variable safely without locks, using compare-and-set.
CAS = read, compute, write-if-unchanged, retry on failure. Nothing blocks and nothing deadlocks.
Functions passed to updateAndGet may run more than once: keep them pure.
LongAdder for hot counters; sum() is approximate under concurrent updates.
AtomicReference plus an immutable object updates several fields at once.
Two atomics do not make one atomic operation: invariants across variables need a lock or one reference.

Mastery checklist

Tick an item only when it is true with your notes closed.

Replace a racy counter with AtomicInteger and explain why no update can be lost
Explain CAS and write an increment as an explicit CAS retry loop
Use updateAndGet and accumulateAndGet correctly, with side-effect-free functions
Update several fields atomically with AtomicReference and an immutable object
Choose LongAdder vs AtomicLong and explain why sum() is not a snapshot
Explain why two atomics do not make a multi-step operation atomic, and pick volatile, atomic or lock for a given problem

PreviousModule 7: Future and CompletableFuture NextModule 9: Concurrent collections

Phase 2: The java.util.concurrent toolkitCoreHighest priority

Module 9: Concurrent collections and BlockingQueue

java.util.concurrent ships collections built for many threads: ConcurrentHashMap, copy-on-write lists, non-blocking queues and blocking queues for producer-consumer designs. This module covers how they work, the atomic operations that replace check-then-act, and why a thread-safe collection still does not make your workflow thread-safe.

9 lessons3 figures6 mastery checks

Warm-upWarm-up: one line each

Why can ConcurrentHashMap serve many writers at once while Collections.synchronizedMap cannot?
Reveal
The synchronized wrapper uses one lock for the whole map. ConcurrentHashMap locks only the bin being changed (and reads take no lock), so writers to different bins proceed in parallel.
Can you put a null value into a ConcurrentHashMap?
Reveal
No, nulls are rejected with NullPointerException. Otherwise get returning null would be ambiguous: missing key or null value, and you could not check atomically which.
How do you count words from many threads into one ConcurrentHashMap<String, Long> in one call per word?
Reveal
counts.merge(word, 1L, Long::sum). It inserts 1 or adds 1 atomically.
A consumer calls queue.take() on an empty BlockingQueue. What happens?
Reveal
It blocks (state WAITING) until a producer puts an item, or until the thread is interrupted.
Iterating a ConcurrentHashMap while other threads modify it: exception?
Reveal
No. Its iterators are weakly consistent: they never throw ConcurrentModificationException and may or may not show changes made during the iteration.

9.1Synchronized wrappers vs concurrent collections

Studied

Try it first2 minutes

You wrap a HashMap with Collections.synchronizedMap and iterate over it in one thread while another thread adds keys. What can happen?

Reveal the answer

ConcurrentModificationException, or corrupted reads, because each method locks but the whole iteration does not. The Javadoc requires you to hold the map's lock for the entire loop: synchronized (map) { for (...) }, which blocks every writer meanwhile.

Figure 1Coarse versus fine-grained locking. Highlighted bins are being written right now by three threads at once.

	`Collections.synchronizedMap(new HashMap<>())`	`ConcurrentHashMap`
Locking	One lock for every operation	Per-bin locking for writes; reads take no lock
Iteration	Must hold the map's lock for the whole loop; otherwise fail-fast `ConcurrentModificationException`	Weakly consistent; never throws, may miss concurrent changes
Compound actions	Hold the lock yourself around them	Built-in atomic methods: `putIfAbsent`, `compute`, `merge`, ...
Nulls	Allowed	Not allowed
Use today	Rarely; legacy code	The default shared map

9.2Inside ConcurrentHashMap

Studied

Try it first2 minutes

Two threads put keys that land in two different, empty bins. Does either one take a lock?

Reveal the answer

No. Inserting into an empty bin is a single CAS. A lock (on the first node of the bin) is taken only when the bin already has entries.

Since Java 8, ConcurrentHashMap is an array of bins, like HashMap:

get reads without locking; the fields it reads are volatile, so it sees completed writes.
put into an empty bin uses CAS. Into a non-empty bin it locks only that bin (it synchronizes on the bin's first node).
When a bin collects many collisions (8 or more, once the table has at least 64 bins), it turns into a small balanced tree, so even bad hash codes stay at O(log n) per bin.
Resizing happens in pieces, and threads that call put during a resize help move bins.
size() and mappingCount() are estimates while other threads are writing. Use them for metrics, not for decisions.
Iterators, keySet(), values() and streams are weakly consistent: they reflect the map at some point during the traversal and never throw ConcurrentModificationException.

Analogy

Think of a bank branch with one counter per account range instead of one counter for the whole branch. Customers for different ranges are served at the same time; only customers for the same range queue.

Figure 2The write path of ConcurrentHashMap.put (simplified). Contention exists only between writers to the same bin.

9.3Atomic map operations

Studied

Try it first2 minutes

Two threads run this for the same new userId. What goes wrong, and what is the one-line fix?

java Session s = sessions.get(userId); if (s == null) { s = new Session(userId); sessions.put(userId, s); } return s;

Reveal the answer

Both threads see null, both create a Session, and the second put replaces the first, so two callers hold different sessions. Fix: return sessions.computeIfAbsent(userId, Session::new);, which checks and inserts as one atomic step and runs the function at most once per missing key.

Method	Atomically does
`putIfAbsent(k, v)`	Insert only if the key is missing; returns the existing value or `null`
`computeIfAbsent(k, fn)`	If missing, compute the value once and insert it; returns the current value
`computeIfPresent(k, fn)`	If present, replace with `fn(k, old)`; `null` result removes it
`compute(k, fn)`	Replace with `fn(k, oldOrNull)` in any case
`merge(k, v, fn)`	Insert `v` if missing, else replace with `fn(old, v)`
`replace(k, expected, new)`	Compare-and-set for one key
`remove(k, expected)`	Remove only if the key still maps to `expected`
`getOrDefault(k, d)`	Read with a default (no insert)

WordCount.java

import java.util.concurrent.*;

public class WordCount {
    public static void main(String[] args) throws InterruptedException {
        ConcurrentHashMap<String, Long> counts = new ConcurrentHashMap<>();
        String[] words = "pay refund pay pay chargeback refund".split(" ");

        ExecutorService pool = Executors.newFixedThreadPool(4);
        for (int t = 0; t < 4; t++) {
            pool.execute(() -> {
                for (int i = 0; i < 10_000; i++)
                    for (String w : words) counts.merge(w, 1L, Long::sum);   // atomic insert-or-add
            });
        }
        pool.shutdown();
        pool.awaitTermination(10, TimeUnit.SECONDS);
        System.out.println(new java.util.TreeMap<>(counts));
    }
}

Watch out

The function you pass to computeIfAbsent, compute or merge runs while the bin is locked. Keep it short: no HTTP calls, no database queries, and never modify the same map inside it (Java 9+ may throw IllegalStateException: Recursive update). For slow loads, store a CompletableFuture as the value instead (Lab 4).

Example: per-merchant counters under heavy traffic. Combine the map with LongAdder: computeIfAbsent creates each counter once, and LongAdder.increment() spreads contention across cells instead of locking:

MerchantCounters.java

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

public class MerchantCounters {
    private final ConcurrentHashMap<String, LongAdder> counts = new ConcurrentHashMap<>();

    void record(String merchant) {
        counts.computeIfAbsent(merchant, m -> new LongAdder()).increment();   // create once; increment() never locks
    }
    long count(String merchant) {
        LongAdder a = counts.get(merchant);
        return a == null ? 0 : a.sum();
    }

    public static void main(String[] args) throws InterruptedException {
        MerchantCounters c = new MerchantCounters();
        ExecutorService pool = Executors.newFixedThreadPool(8);
        for (int i = 0; i < 80_000; i++) {
            String m = i % 2 == 0 ? "swiggy" : "zomato";
            pool.execute(() -> c.record(m));
        }
        pool.shutdown();
        pool.awaitTermination(10, TimeUnit.SECONDS);
        System.out.println("swiggy=" + c.count("swiggy") + " zomato=" + c.count("zomato"));
    }
}

9.4Copy-on-write lists, skip lists and non-blocking queues

Studied

Try it first2 minutes

A payment gateway client keeps a list of listeners notified on every transaction. Listeners are added at startup and almost never change; notifications happen thousands of times per second. Which list?

Reveal the answer

CopyOnWriteArrayList. Reads and iteration take no lock and never throw; each rare write copies the array. With frequent writes or a large list it would be a poor choice.

Collection	How it works	Use for	Avoid when
`CopyOnWriteArrayList` / `CopyOnWriteArraySet`	Every write copies the whole array; iterators see a snapshot	Listener lists, small read-mostly config lists	Frequent writes or large lists
`ConcurrentLinkedQueue` / `ConcurrentLinkedDeque`	Lock-free linked queue, unbounded, never blocks	Hand-off where consumers poll and do something else when empty	You need waiting or a size limit (use a `BlockingQueue`)
`ConcurrentSkipListMap` / `ConcurrentSkipListSet`	Sorted, concurrent, O(log n)	Concurrent sorted data: order books, time-ordered events, leaderboards	Plain key lookups (`ConcurrentHashMap` is faster)
`ConcurrentHashMap.newKeySet()`	A concurrent `Set` backed by the map	Deduplication across threads (`add` returns false if present)

Example: a listener registry. Iteration works on a snapshot, so a listener may even register another listener during a notification:

Listeners.java

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

public class Listeners {
    public static void main(String[] args) {
        List<Consumer<String>> listeners = new CopyOnWriteArrayList<>();
        listeners.add(e -> System.out.println("audit: " + e));
        listeners.add(e -> {
            System.out.println("fraud: " + e);
            listeners.add(x -> System.out.println("late: " + x));   // with ArrayList: ConcurrentModificationException
        });
        for (Consumer<String> l : listeners) l.accept("txn-1");     // iterates the snapshot taken at the start
        System.out.println("listeners now: " + listeners.size());
    }
}

Example: events by time. A ConcurrentSkipListMap keyed by timestamp gives sorted, concurrent range queries, handy for sliding windows:

EventWindow.java

import java.util.concurrent.ConcurrentSkipListMap;

public class EventWindow {
    public static void main(String[] args) {
        ConcurrentSkipListMap<Long, String> byTime = new ConcurrentSkipListMap<>();
        byTime.put(1000L, "login");
        byTime.put(1500L, "add card");
        byTime.put(2600L, "payment");
        byTime.put(4100L, "logout");
        System.out.println("1000..3000: " + byTime.subMap(1000L, true, 3000L, true).values());
        System.out.println("latest: " + byTime.lastEntry().getValue());
        byTime.headMap(2000L).clear();                                  // drop everything older than 2000
        System.out.println("after cleanup: " + byTime.keySet());
    }
}

Keys are unique: two events with the same timestamp would overwrite each other. In real code, key by (timestamp, event ID) or keep a list per timestamp.

9.5BlockingQueue and producer-consumer

Studied

Try it first2 minutes

A file reader thread produces records far faster than three database writer threads can store them. Which queue keeps memory under control, and what happens to the reader when it is full?

Reveal the answer

A bounded BlockingQueue, for example new ArrayBlockingQueue<>(1_000). When it is full, put blocks the reader until a writer takes an item, so the fast producer is slowed to the pace of the consumers (backpressure).

Figure 3The queue decouples producers from consumers and absorbs bursts. Its bound is what turns "too fast" into "wait" instead of "out of memory".

Every BlockingQueue offers four styles for each operation (from its Javadoc):

	Throws an exception	Returns a special value	Blocks	Blocks with a timeout
Insert	`add(e)`	`offer(e)` returns `false`	`put(e)`	`offer(e, time, unit)`
Remove	`remove()`	`poll()` returns `null`	`take()`	`poll(time, unit)`
Examine	`element()`	`peek()`

Implementation	Bounded?	Notes
`ArrayBlockingQueue`	Always	One lock; optional fairness
`LinkedBlockingQueue`	Only if you pass a capacity	Separate locks for put and take, so higher throughput
`SynchronousQueue`	Zero capacity	Each `put` waits for a `take`: a direct hand-off
`PriorityBlockingQueue`	Unbounded	Takes the highest-priority element first
`DelayQueue`	Unbounded	Elements become available after their delay: retry and expiry schedules

Pipeline.java: bounded producer-consumer with poison pills

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

public class Pipeline {
    private static final String POISON = "__END__";          // marker that tells a consumer to stop

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(100);
        int consumers = 3;
        CountDownLatch done = new CountDownLatch(consumers);
        LongAdder stored = new LongAdder();

        for (int c = 0; c < consumers; c++) {
            new Thread(() -> {
                try {
                    while (true) {
                        String record = queue.take();          // blocks while empty
                        if (record.equals(POISON)) break;
                        stored.increment();                    // pretend to write to the DB
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    done.countDown();
                }
            }, "writer-" + c).start();
        }

        for (int i = 0; i < 10_000; i++) queue.put("record-" + i);   // blocks while full
        for (int c = 0; c < consumers; c++) queue.put(POISON);        // one pill per consumer
        done.await();
        System.out.println("stored " + stored.sum() + " records");
    }
}

drainTo(list, max) takes many items in one call, which is handy for writing to a database in batches. Kafka applies the producer-consumer idea across machines, with one big difference: its log is limited by retention, not by consumers. A slow consumer never slows producers down; it falls behind (lag), and records older than the retention period are deleted whether or not they were read.

Example: a retry queue. DelayQueue.take() waits until the earliest item is due:

RetryQueue.java

import java.util.concurrent.*;

public class RetryQueue {
    record Retry(String paymentId, long dueAtNanos) implements Delayed {
        public long getDelay(TimeUnit unit) { return unit.convert(dueAtNanos - System.nanoTime(), TimeUnit.NANOSECONDS); }
        public int compareTo(Delayed o) { return Long.compare(dueAtNanos - ((Retry) o).dueAtNanos, 0); }   // nanoTime: compare differences
    }

    public static void main(String[] args) throws InterruptedException {
        DelayQueue<Retry> retries = new DelayQueue<>();
        long now = System.nanoTime();
        retries.put(new Retry("pay-3", now + TimeUnit.MILLISECONDS.toNanos(300)));
        retries.put(new Retry("pay-1", now + TimeUnit.MILLISECONDS.toNanos(100)));
        retries.put(new Retry("pay-2", now + TimeUnit.MILLISECONDS.toNanos(200)));
        for (int i = 0; i < 3; i++) System.out.println("retrying " + retries.take().paymentId());  // waits for each due time
    }
}

An in-memory retry queue is lost on restart. For payments, keep retries in a database table and poll it; use DelayQueue for things that may be lost (cache refreshes, local back-off).

9.6A thread-safe collection is not a thread-safe workflow

Studied

Try it first2 minutes

Both maps are ConcurrentHashMap. Is this transfer of a reservation thread-safe?

java Seat seat = available.remove(seatId); if (seat != null) reserved.put(seatId, seat);

Reveal the answer

Each call is atomic, and the remove makes sure only one thread gets the seat, so two users cannot both reserve it. But another thread looking at both maps between the two lines sees the seat in neither, so a "count all seats" check is wrong for a moment. If readers need the two maps consistent with each other, guard both with one lock or keep a single map of seat to status.

Concurrent collections make each call safe. They do not make a sequence of calls safe:

Workflow	Problem	Fix
`if (!map.containsKey(k)) map.put(k, v)`	Check-then-act	`putIfAbsent` / `computeIfAbsent`
`v = map.get(k); map.put(k, v + 1)`	Read-modify-write	`merge(k, 1, Integer::sum)` or `compute`
`if (queue.size() < 100) queue.offer(x)`	`size()` is stale by the time you act	A bounded queue: `offer` returns `false` when full
Moving an item between two collections	Readers can see it in neither or both	One lock around both, or one data structure
Iterate, then decide and update	The data changed during the iteration	Compute per key with `compute`, or snapshot under a lock

9.7Use cases: picking a concurrent collection

Studied

Try it first2 minutes

Match: (a) cache of loaded merchant configs; (b) message IDs already processed; (c) listeners notified on every transaction; (d) hand records from a reader thread to writer threads.

Reveal the answer

(a) ConcurrentHashMap with computeIfAbsent; (b) ConcurrentHashMap.newKeySet(), where add returns false for duplicates; (c) CopyOnWriteArrayList; (d) a bounded ArrayBlockingQueue.

Cache of loaded objectsConcurrentHashMap.computeIfAbsent

Loads once per key; keep the function fast (or store futures, Lab 4).

Per-key counters under heavy loadConcurrentHashMap<K, LongAdder>

The MerchantCounters example.

Deduplicate message IDsConcurrentHashMap.newKeySet()

add(id) is false for a duplicate. In-memory only: pods and restarts need the database.

Listener or plugin registryCopyOnWriteArrayList

Fast, safe iteration; rare writes.

Events by time, sliding windowsConcurrentSkipListMap

Sorted keys, range views, concurrent.

Hand work from producers to consumersArrayBlockingQueue (bounded)

Backpressure through put.

Retry or expire later, in memoryDelayQueue

Items become available at their due time.

Jobs with prioritiesPriorityBlockingQueue

Unbounded: cap it yourself.

9.8Common mistakes

Studied

Common mistakes

containsKey then put, or get then put, on a concurrent map.
Slow or recursive functions inside computeIfAbsent, which block that bin for every thread.
LinkedBlockingQueue with no capacity, an unbounded buffer in disguise.
Deciding with size() or isEmpty() on collections other threads are changing.
CopyOnWriteArrayList for frequently written data, which copies the whole array on each write.
Forgetting one poison pill per consumer, so some consumers wait forever at shutdown.
Iterating a synchronized wrapper without holding its lock.

9.9Self-check

Studied

Self-checkFive more, after the lessons

Why are ConcurrentHashMap reads fast?
Reveal
They take no lock; they read volatile fields and see completed writes.
What does weakly consistent iteration mean?
Reveal
The iterator never throws ConcurrentModificationException and reflects the map at some point at or after it was created; it may or may not show concurrent changes.
Which BlockingQueue methods block, and which return immediately?
Reveal
put and take block; offer and poll return immediately (or after a timeout in their timed forms); add and remove throw on failure.
When is CopyOnWriteArrayList the right choice?
Reveal
Small, read-mostly lists such as listeners, where iteration must be fast and safe and writes are rare.
Why should the computeIfAbsent function be short?
Reveal
It runs while that bin is locked, so other threads updating keys in the same bin wait for it, and it must not touch the same map.

Key takeaways

ConcurrentHashMap locks per bin and reads without locks; synchronized wrappers lock everything.
Replace check-then-act on maps with putIfAbsent, computeIfAbsent, compute and merge.
Keep mapping functions short and never touch the same map inside them.
Concurrent iterators are weakly consistent; size() is an estimate while writers run.
Use bounded BlockingQueues for hand-off; choose the method style (throw, value, block, timeout) on purpose.
Each call is atomic, a sequence of calls is not: design workflows, not just data structures.

Mastery checklist

Tick an item only when it is true with your notes closed.

Explain how ConcurrentHashMap differs from a synchronized map in locking, iteration and nulls
Replace check-then-act and read-modify-write map code with computeIfAbsent, merge and compute
Pick between CopyOnWriteArrayList, ConcurrentLinkedQueue and ConcurrentSkipListMap for a use case
Build a bounded producer-consumer pipeline with a BlockingQueue and shut it down with poison pills
Use the four method styles of BlockingQueue (throw, special value, block, timeout) correctly
Show why a thread-safe collection does not make a multi-step workflow thread-safe, and fix one

PreviousModule 8: Atomics and CAS NextModule 10: Locks and synchronizers

Phase 2: The java.util.concurrent toolkitCoreUnderstand and practise

Module 10: Explicit locks and synchronizers

When synchronized is not enough, java.util.concurrent.locks gives you ReentrantLock, Condition and read-write locks. Alongside them sit the synchronizers that coordinate whole groups of threads: CountDownLatch, CyclicBarrier and Semaphore. This module covers what each one solves and its traps.

11 lessons5 figures6 mastery checks

Warm-upWarm-up: one line each

Why must lock.unlock() go in a finally block?
Reveal
If the code between lock() and unlock() throws, the lock would stay held forever and every other thread would wait forever. synchronized releases automatically; ReentrantLock does not.
Name one thing ReentrantLock can do that synchronized cannot.
Reveal
Try to acquire without waiting (tryLock()), wait with a timeout (tryLock(2, SECONDS)), be interrupted while waiting (lockInterruptibly()), be fair, or have several conditions.
Can a CountDownLatch be reset and reused?
Reveal
No. Once the count reaches zero it stays open. Use CyclicBarrier (or Phaser) for a reusable meeting point.
You must allow at most 5 concurrent calls to a partner API from a service. Which tool?
Reveal
A Semaphore with 5 permits: acquire() before the call, release() in finally.
A thread holds a read lock of a ReentrantReadWriteLock and asks for the write lock. What happens?
Reveal
It blocks forever. The write lock waits for all readers to leave, including this thread. Upgrading is not supported; release the read lock first.

10.1ReentrantLock

Studied

Try it first2 minutes

A money transfer must lock two accounts. If it cannot get a lock within 1 second it should give up and report "busy, retry" rather than wait forever. Can synchronized express that?

Reveal the answer

No: a thread waiting for a monitor waits forever. ReentrantLock.tryLock(1, TimeUnit.SECONDS) returns false after the timeout, so you can back off, release what you hold and retry. Module 11 uses this to avoid deadlocks.

Lock with a timeout, unlock in finally

class Account {
    final long id;
    long balancePaise;
    final ReentrantLock lock = new ReentrantLock();
    Account(long id, long balance) { this.id = id; this.balancePaise = balance; }
}

class TransferService {
    boolean transfer(Account from, Account to, long amount) throws InterruptedException {
        if (!from.lock.tryLock(1, TimeUnit.SECONDS)) return false;     // give up instead of waiting forever
        try {
            if (!to.lock.tryLock(1, TimeUnit.SECONDS)) return false;
            try {
                if (from.balancePaise < amount) return false;
                from.balancePaise -= amount;
                to.balancePaise += amount;
                return true;
            } finally {
                to.lock.unlock();
            }
        } finally {
            from.lock.unlock();                                        // always, even after an exception
        }
    }
}

	`synchronized`	`ReentrantLock`
Release	Automatic at block end	You call `unlock()`, always in `finally`
Try without waiting	No	`tryLock()`
Wait with a timeout	No	`tryLock(time, unit)`
Interruptible while waiting	No	`lockInterruptibly()`
Fairness	No	`new ReentrantLock(true)`: longest waiter goes first (slower)
Conditions per lock	One wait set	Any number: `newCondition()`
Reentrant, visibility guarantees	Yes	Yes
Shows in thread dumps	Monitor owner shown	Shown as an "ownable synchronizer" (use `jstack -l`)

Rule of thumb

Default to synchronized: shorter, impossible to forget to unlock. Reach for ReentrantLock when you need a timeout, interruptible waiting, fairness or several conditions, or, on Java 21 to 23 with virtual threads, when the critical section can block (to avoid pinning).

Go deeper

How locks work inside: AQS and LockSupport

ReentrantLock, Semaphore, CountDownLatch, ReentrantReadWriteLock and the thread pool's workers are all built on AbstractQueuedSynchronizer (AQS):

One int state: for a lock, 0 means free and N is the hold count; for a semaphore, the permits left; for a latch, the remaining count.
Acquiring is a CAS on state. A thread that fails joins a FIFO wait queue and goes to sleep with LockSupport.park() (state WAITING (parking) in a dump, Module 15).
Releasing updates state and wakes the next queued thread with LockSupport.unpark(thread).
Non-fair (the default) lets an arriving thread grab a free lock even while others are queued. It can barge ahead of sleepers but avoids a wake-up delay, so throughput is higher. Fair mode takes a free lock only if no thread is already queued; otherwise it joins the end of the queue.

That is why ReentrantLock waits show as parking to wait for <...ReentrantLock$NonfairSync> rather than as a monitor in thread dumps.

10.2Condition: wait sets you can name

Studied

Try it first2 minutes

In the wait/notifyAll bounded buffer of Module 5, every put wakes both producers and consumers. How could producers and consumers wait in separate places, so a put wakes only consumers?

Reveal the answer

Use one ReentrantLock with two Conditions, notFull and notEmpty. Producers await notFull, consumers await notEmpty; put signals notEmpty and take signals notFull.

Bounded queue with two conditions

class BoundedQueue<T> {
    private final Deque<T> items = new ArrayDeque<>();
    private final int capacity;
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();

    BoundedQueue(int capacity) { this.capacity = capacity; }

    void put(T item) throws InterruptedException {
        lock.lock();
        try {
            while (items.size() == capacity) notFull.await();   // releases the lock while waiting
            items.addLast(item);
            notEmpty.signal();                                   // wake one consumer only
        } finally {
            lock.unlock();
        }
    }

    T take() throws InterruptedException {
        lock.lock();
        try {
            while (items.isEmpty()) notEmpty.await();
            T item = items.removeFirst();
            notFull.signal();                                    // wake one producer only
            return item;
        } finally {
            lock.unlock();
        }
    }
}

Same rules as wait/notify: hold the lock to call await or signal, and await in a while loop. signal() is safe here because each condition has only one kind of waiter. This is essentially how ArrayBlockingQueue is written.

Figure 1Two conditions on one lock: each signal reaches a thread that can actually proceed.

10.3Read-write locks

Studied

Try it first2 minutes

A currency-rate table is read on every request (thousands per second) and refreshed once a minute. With one ordinary lock, readers block each other. What would let readers run together?

Reveal the answer

A ReentrantReadWriteLock: many threads may hold the read lock at once; the write lock is exclusive. Even simpler here: keep an immutable map in a volatile field and swap it on refresh (Module 2), with no locks at all.

Figure 2Many readers or one writer, never both.

Read-mostly data behind a read-write lock

class RateTable {
    private final Map<String, BigDecimal> rates = new HashMap<>();
    private final ReentrantReadWriteLock rw = new ReentrantReadWriteLock();

    BigDecimal rate(String pair) {
        rw.readLock().lock();
        try { return rates.get(pair); }
        finally { rw.readLock().unlock(); }
    }

    void refresh(Map<String, BigDecimal> fresh) {
        rw.writeLock().lock();
        try { rates.clear(); rates.putAll(fresh); }
        finally { rw.writeLock().unlock(); }
    }
}

No upgrade. Holding the read lock and requesting the write lock blocks forever. Release the read lock first, then re-check after getting the write lock. Downgrading (write, then read, then release write) is allowed.
Worth it only when reads are frequent, take meaningful time, and writes are rare. A read-write lock does more bookkeeping than a plain lock, so for very short reads a ReentrantLock, a ConcurrentHashMap or an immutable snapshot is often faster.
StampedLock adds optimistic reads for extreme read-heavy cases; it is not reentrant and is easy to misuse. Learn it later, when a profiler says you need it.

10.4CountDownLatch

Studied

Try it first2 minutes

A service starts three caches in parallel and must not accept traffic until all three are warm. How does the main thread wait for exactly three events?

Reveal the answer

Create new CountDownLatch(3). Each loader calls countDown() when done; the main thread calls await() and continues once the count reaches zero.

Figure 3A latch is a gate with a counter. Waiters pass when the count reaches zero.

WarmUp.java

import java.util.concurrent.*;

public class WarmUp {
    public static void main(String[] args) throws InterruptedException {
        String[] caches = {"bins", "merchants", "fx-rates"};
        CountDownLatch ready = new CountDownLatch(caches.length);
        ExecutorService pool = Executors.newFixedThreadPool(3);
        for (String name : caches) {
            pool.execute(() -> {
                try {
                    Thread.sleep(100);                       // load the cache
                    System.out.println(name + " loaded");
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    ready.countDown();                       // in finally: a failure must not leave main waiting
                }
            });
        }
        if (ready.await(5, TimeUnit.SECONDS)) System.out.println("all " + caches.length + " caches warm, accepting traffic");
        else System.out.println("warm-up timed out");
        pool.shutdown();
    }
}

Two classic uses: an end gate (wait for N tasks, as above) and a start gate (new CountDownLatch(1): many threads await(), then one countDown() releases them all at once, which is how Module 15 makes race tests overlap).

10.5CyclicBarrier

Studied

Try it first2 minutes

A simulation runs in rounds. Three worker threads each compute one third of a round, and no worker may start round 2 until all three have finished round 1. Latch or barrier?

Reveal the answer

CyclicBarrier(3). Each worker calls await() at the end of a round; the last one to arrive releases everyone, and the barrier resets for the next round. A latch cannot be reused.

Rounds.java

import java.util.concurrent.*;

public class Rounds {
    public static void main(String[] args) {
        CyclicBarrier barrier = new CyclicBarrier(3, () -> System.out.println("round complete"));  // runs once per round
        for (int w = 1; w <= 3; w++) {
            int id = w;
            new Thread(() -> {
                try {
                    for (int round = 1; round <= 2; round++) {
                        Thread.sleep(50L * id);               // uneven work
                        barrier.await();                      // wait for the other two
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } catch (BrokenBarrierException e) {
                    System.out.println("another worker failed; stopping");
                }
            }).start();
        }
    }
}

	`CountDownLatch`	`CyclicBarrier`
Who waits	Any threads wait for N events	N threads wait for each other
Reusable	No	Yes, it resets after each trip
Extra		A barrier action runs when the last thread arrives
Failure		If one waiter is interrupted or times out, the others get `BrokenBarrierException`

Phaser is a more flexible, reusable barrier where the number of parties can change between phases. You will rarely need it in service code.

Figure 4Dashed bars are time spent waiting in await(). Everyone leaves together, and the barrier resets for the next round.

10.6Semaphore

Studied

Try it first2 minutes

A partner's API allows at most 3 concurrent requests per client. Your service has 50 request threads that may call it. How do you enforce the limit inside one JVM?

Reveal the answer

Semaphore permits = new Semaphore(3). Each call does acquire() (or tryAcquire with a timeout) before the request and release() in finally. At most 3 threads are inside at once; the others wait, or fail fast if you use tryAcquire.

Idea

A semaphore holds a number of permits. acquire() takes one, waiting if none are left; release() returns one. It limits how many threads do something at the same time.

Analogy

A car park with 3 spaces and a counter at the gate. Cars enter while spaces remain and wait at the gate when it shows zero. Any leaving car frees a space, whoever it is.

Cost

Permits are not tied to threads: release() from any thread, or a release without an acquire, silently adds a permit. Always pair them in try/finally.

PartnerLimiter.java

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

public class PartnerLimiter {
    private static final Semaphore permits = new Semaphore(3);
    private static final AtomicInteger inFlight = new AtomicInteger(), maxSeen = new AtomicInteger();

    static String callPartner(int id) throws InterruptedException {
        if (!permits.tryAcquire(2, TimeUnit.SECONDS)) return "busy";   // fail fast instead of piling up
        try {
            maxSeen.accumulateAndGet(inFlight.incrementAndGet(), Math::max);
            Thread.sleep(100);                                         // the remote call
            return "ok " + id;
        } finally {
            inFlight.decrementAndGet();
            permits.release();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ExecutorService pool = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 10; i++) {
            int id = i;
            pool.execute(() -> { try { callPartner(id); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } });
        }
        pool.shutdown();
        pool.awaitTermination(10, TimeUnit.SECONDS);
        System.out.println("max in flight = " + maxSeen.get());       // 3, never more
    }
}

In production

This is a bulkhead: a cap on concurrent calls to one dependency, so a slow partner cannot take every thread you have (Module 14). With several pods, each pod has its own semaphore: 4 pods x 3 permits = 12 concurrent calls in total. A cluster-wide limit needs a shared store such as Redis or the partner's own rate limit.

Figure 5Permits are counted, not owned: whoever releases frees a slot for the next waiter.

10.7Build it yourself: a resource pool with a Semaphore

Studied

Try it first2 minutes

You have 3 expensive objects (say, SFTP sessions) and many threads that each need one for a while. Which two structures make a correct pool, and in which order do borrow and giveBack touch them?

Reveal the answer

A Semaphore with 3 permits and a thread-safe queue of the objects. borrow: acquire a permit, then take an object. giveBack: put the object back, then release the permit. That order guarantees a thread holding a permit always finds an object waiting.

ResourcePool.java

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

public class ResourcePool<T> {
    private final Semaphore permits;
    private final ConcurrentLinkedQueue<T> items;

    public ResourcePool(Collection<T> resources) {
        items = new ConcurrentLinkedQueue<>(resources);
        permits = new Semaphore(resources.size(), true);          // fair: waiters served in order
    }

    public T borrow(long timeoutMs) throws InterruptedException {
        if (!permits.tryAcquire(timeoutMs, TimeUnit.MILLISECONDS)) throw new IllegalStateException("pool exhausted");
        return items.poll();                                       // a permit guarantees an item is waiting
    }

    public void giveBack(T item) {
        items.offer(item);                                         // first return the item...
        permits.release();                                         // ...then free the permit
    }

    public static void main(String[] args) throws InterruptedException {
        ResourcePool<String> sftp = new ResourcePool<>(List.of("session-1", "session-2", "session-3"));
        AtomicInteger inUse = new AtomicInteger(), maxInUse = new AtomicInteger();
        ExecutorService pool = Executors.newFixedThreadPool(8);
        for (int i = 0; i < 16; i++) {
            pool.execute(() -> {
                try {
                    String s = sftp.borrow(2_000);
                    try {
                        maxInUse.accumulateAndGet(inUse.incrementAndGet(), Math::max);
                        Thread.sleep(50);                         // upload a file
                    } finally {
                        inUse.decrementAndGet();
                        sftp.giveBack(s);                         // always give it back
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
        pool.shutdown();
        pool.awaitTermination(10, TimeUnit.SECONDS);
        System.out.println("16 uploads done, max sessions in use at once = " + maxInUse.get());
    }
}

This is the core of every connection pool (HikariCP adds validation, timeouts, leak detection and much faster hand-off). Notice giveBack in finally: a pool that leaks objects slowly runs dry, the classic connection-leak incident.

10.8Choosing the right tool

Studied

Try it first2 minutes

Match each need to a tool: (a) wait until 5 files are parsed; (b) at most 10 threads in a block of code; (c) 4 threads must finish step 1 together before any starts step 2, repeatedly; (d) a lock you can give up on after 500 ms.

Reveal the answer

(a) CountDownLatch(5) (b) Semaphore(10) (c) CyclicBarrier(4) (d) ReentrantLock.tryLock(500, MILLISECONDS).

I need to...	Use
Protect shared state, simply	`synchronized`
Protect shared state with a timeout, interruptible waits, fairness or several conditions	`ReentrantLock` (+ `Condition`)
Let many readers in at once, writers alone	`ReentrantReadWriteLock`, or an immutable snapshot in a `volatile` field
Wait until N things have happened, once	`CountDownLatch`
Make N threads meet at a point, repeatedly	`CyclicBarrier` (or `Phaser`)
Limit how many threads do something at once	`Semaphore`
Pass work from producers to consumers	`BlockingQueue` (Module 9)
Get a result from another thread	`Future` / `CompletableFuture` (Module 7)

Rarely needed in service code: Phaser (a reusable barrier where parties can register and deregister at any time) and Exchanger (two threads swap objects at a meeting point).

10.9Use cases: synchronizers in services

Studied

Try it first2 minutes

A test must make 50 threads call debit() at exactly the same moment, then wait until all 50 are done. Which two synchronizers?

Reveal the answer

A CountDownLatch(1) as a start gate (all threads await(), the test calls countDown()), and a CountDownLatch(50) as an end gate (each thread counts down in finally, the test awaits with a timeout).

Limit concurrent calls to a partnerSemaphore

A bulkhead in a few lines (Lab 8).

Pool of expensive objectsSemaphore + queue

The ResourcePool above.

Wait for startup tasks before serving trafficCountDownLatch

With a timeout, and countDown() in finally.

Make test threads start togetherCountDownLatch(1) as a start gate

Module 15.

Batch rounds that must stay in stepCyclicBarrier

Simulations, phased batch jobs.

A lock with a timeout or interruptible waitReentrantLock.tryLock / lockInterruptibly

Transfers that report "busy" instead of hanging.

Producers and consumers with separate waitsReentrantLock + two Conditions

Or simply ArrayBlockingQueue.

Read-heavy data, rare writes, slow readsReentrantReadWriteLock

Often an immutable snapshot in a volatile field is simpler.

10.10Common mistakes

Studied

Common mistakes

unlock() outside finally, which leaks the lock on any exception.
lock() placed inside the try: if lock() itself fails, the finally unlocks a lock you never held. Call lock() just before try.
if instead of while around Condition.await().
Upgrading a read lock to a write lock, which self-deadlocks.
countDown() not in finally, so one failure leaves the waiter stuck until its timeout (or forever).
release() without a matching acquire(), silently increasing the permit count.
Fair locks by default. Fairness costs throughput; use it only when starvation is a real problem.

10.11Self-check

Studied

Self-checkFive more, after the lessons

What does tryLock() without arguments do on a fair ReentrantLock?
Reveal
It takes the lock immediately if it is free, even if other threads are waiting (it ignores fairness). tryLock(0, unit) respects fairness.
Why can signal() be used safely in the two-condition bounded queue?
Reveal
Each condition has only one kind of waiter (only producers on notFull, only consumers on notEmpty), so any thread woken can make progress.
Latch or barrier: N threads must wait for each other at the end of every batch?
Reveal
CyclicBarrier, because it resets after each trip.
What does a semaphore protect against that a lock does not?
Reveal
Too many threads doing something at once: it allows up to N at a time rather than exactly 1.
When is a read-write lock not worth it?
Reveal
When reads are very short or writes are frequent; the extra bookkeeping costs more than it saves.

Key takeaways

ReentrantLock: always unlock() in finally; use it for timeouts, interruptible waits, fairness or several conditions.
Condition = a named wait set on a lock; same rules as wait: hold the lock, wait in a loop.
Read-write locks allow many readers or one writer; upgrading a read lock deadlocks.
CountDownLatch waits for N events once; CyclicBarrier makes N threads meet repeatedly.
Semaphore limits how many threads do something at once; permits are not owned by threads.
All of these are built on AQS: a CAS on one state plus a queue of parked threads.

Mastery checklist

Tick an item only when it is true with your notes closed.

Use ReentrantLock with try/finally, tryLock with a timeout and lockInterruptibly, and compare it with synchronized
Build a bounded queue with one lock and two Conditions
Use a read-write lock correctly and explain why upgrading deadlocks
Use CountDownLatch as a start gate and as an end gate
Coordinate repeated rounds with CyclicBarrier
Limit concurrent calls to a dependency with a Semaphore and explain its limits across pods

PreviousModule 9: Concurrent collections NextModule 11: Deadlock, livelock, starvation

6.1Why thread pools

6.2Executor, ExecutorService and the Executors factories

6.3ThreadPoolExecutor: how a task finds a thread

6.4Work queues and rejection policies

6.5execute() vs submit()

6.6Shutting down

6.7Scheduled tasks

6.8Sizing pools and handling overload

6.9Build it yourself: a minimal thread pool

6.10Use cases: which executor for which job

6.11Common mistakes

Common mistakes

6.12Self-check

Key takeaways

Mastery checklist

7.1Future: a handle to a result

7.2Why CompletableFuture

7.3Starting async work: supplyAsync and runAsync

7.4Transforming: thenApply, thenCompose, thenAccept

7.5Combining: thenCombine, allOf, anyOf

7.6Handling errors

7.7Timeouts and cancellation

7.8Common pool or your own executor

7.9Recipes you will use

7.10Use cases: CompletableFuture in services

7.11Common mistakes

Common mistakes

7.12Self-check

Key takeaways

Mastery checklist

8.1Atomic counters

8.2How compare-and-set works

8.3updateAndGet, accumulateAndGet and AtomicReference

8.4LongAdder for hot counters

False sharing

8.5Choosing volatile, atomic or lock

The ABA problem

8.6Build it yourself: a spin lock with CAS

8.7Use cases: atomics in services

8.8Common mistakes

Common mistakes

8.9Self-check

Key takeaways

Mastery checklist

9.1Synchronized wrappers vs concurrent collections

9.2Inside ConcurrentHashMap

9.3Atomic map operations

9.4Copy-on-write lists, skip lists and non-blocking queues

9.5BlockingQueue and producer-consumer

9.6A thread-safe collection is not a thread-safe workflow

9.7Use cases: picking a concurrent collection

9.8Common mistakes

Common mistakes

9.9Self-check

Key takeaways

Mastery checklist

10.1ReentrantLock

How locks work inside: AQS and LockSupport

10.2Condition: wait sets you can name

10.3Read-write locks

10.4CountDownLatch

10.5CyclicBarrier

10.6Semaphore

10.7Build it yourself: a resource pool with a Semaphore

10.8Choosing the right tool

10.9Use cases: synchronizers in services

10.10Common mistakes

Common mistakes

10.11Self-check

Key takeaways

Mastery checklist