The java.util.concurrent toolbox

Part of the free Low-Level Design course on LogicWiz, in the chapter: Concurrency in LLD. The class-diagram exercises run in the app.

The java.util.concurrent toolbox

"You start a new Thread for every order. When the sale begins, the server will run out of memory before it runs out of orders."

Priya, reading a load-test report

One thread per order, and the pool that replaces it

Arjun's first version of order dispatch looked like this.

public void onOrderPlaced(Order order) {
    new Thread(() -> dispatcher.assignRider(order)).start();   // a new thread every time
}

Each thread needs memory for its own stack, and the operating system must switch between all of them. If 9,000 orders arrive in one minute, the server spends its time managing threads instead of assigning riders.

A thread pool fixes this. It is a fixed group of worker threads that take tasks from a queue. A task is one small piece of work. When every worker is busy, new tasks wait in line. In Java you use a pool through an ExecutorService.

Think of a dark store. It does not hire a new rider for every order. It keeps a few riders and a list of orders. The comparison stops at one point: if every rider is stuck waiting for something, the list never moves. The same happens when pool tasks wait for each other.

ExecutorService pool = Executors.newFixedThreadPool(3);
List<Future<Delivery>> futures = new ArrayList<>();
for (int i = 1; i <= 5; i++) {
    final int n = i;
    futures.add(pool.submit(() -> {
        Thread.sleep(50 * (6 - n));              // later orders finish first
        return new Delivery("order-" + n, 10 + n);
    }));
}
for (Future<Delivery> f : futures) {
    System.out.println(f.get());                 // we read them in the order we submitted
}
// Delivery[order=order-1, minutes=11] ... up to order-5, minutes=15

The pool runs the tasks in any order, and up to three at a time. We read the results in the order we submitted them, so the output is the same on every run. A task that returns a value is a Callable. A task that returns nothing is a Runnable.

submit returns a Future. A Future is a receipt for a result that may not be ready yet, like the token you get at a coffee counter. get() waits until the result is ready. If the task threw an exception, get() throws an ExecutionException, and getCause() holds the real error. get(2, TimeUnit.SECONDS) stops waiting after two seconds. cancel(true) asks the worker to stop a task that has not finished.

Now the part that beginners forget, which is the shutdown. Pool workers keep the program alive. If you forget to shut the pool down, the program never exits. There are three calls to know:

• shutdown() stops taking new tasks and lets the queued ones finish.
• awaitTermination(5, TimeUnit.SECONDS) waits until the workers are done, or the time runs out.
• shutdownNow() interrupts the workers and gives up on tasks that have not started.

pool.shutdown();
if (!pool.awaitTermination(5, TimeUnit.SECONDS)) {
    pool.shutdownNow();                          // too slow: stop the workers
}
// after shutdown, pool.submit(...) throws RejectedExecutionException

💡 Tip: For CPU-heavy tasks, use about as many threads as the machine has cores. For tasks that mostly wait for a network reply, more threads help. newFixedThreadPool keeps an unlimited queue, so a flood of tasks can still fill memory. Real services often build a ThreadPoolExecutor with a bounded queue.

CompletableFuture: chaining without waiting

Calling get() on every Future makes your thread wait. A CompletableFuture is a Future you can attach next steps to, so nobody has to wait in between. Each step runs when the one before it finishes.

// chain: step two uses the result of step one
CompletableFuture<String> label = CompletableFuture
    .supplyAsync(() -> priceOf("milk"))
    .thenApply(price -> "milk costs " + price);         // milk costs 30

// combine: two independent steps, then join their results
CompletableFuture<Integer> price = CompletableFuture.supplyAsync(() -> priceOf("milk"));
CompletableFuture<Integer> stock = CompletableFuture.supplyAsync(() -> stockOf("milk"));
price.thenCombine(stock, (p, s) -> "stock value " + p * s);   // stock value 360

thenApply changes a result. If step two is itself asynchronous and returns a CompletableFuture, use thenCompose instead, or you end up with a future inside a future. thenCombine waits for both futures and then calls your function with both results. The two steps run at the same time, so the total time is about the slower one, not the sum.

Errors and timeouts need their own steps:

CompletableFuture<Integer> safe = CompletableFuture
    .<Integer>supplyAsync(() -> { throw new IllegalStateException("price service down"); })
    .exceptionally(ex -> -1);                    // recover with a default value

CompletableFuture<String> slow = CompletableFuture
    .supplyAsync(() -> slowCall())               // takes 2 seconds
    .completeOnTimeout("fallback", 200, TimeUnit.MILLISECONDS);   // prints fallback

exceptionally turns an error into a value. handle receives both the result and the error, and you decide. The error may arrive wrapped in a CompletionException, so call getCause() to see the real one. orTimeout fails the future with a TimeoutException. completeOnTimeout gives a default value instead.

⚠️ Warning: A timeout does not stop the slow work. The caller moves on, but the slow task keeps running in the background until it ends by itself. By default these steps run on a shared pool, so pass your own executor when the work blocks on a network call.

Limiting and waiting: Semaphore and CountDownLatch

SwiftKart's dark store has two packing benches. A semaphore holds a number of permits. A thread takes a permit with acquire() and returns it with release(). When no permit is left, the next thread waits. It works like a parking lot with a fixed number of spots. SwiftKart could use one to allow at most five calls at a time to a payment partner that cannot take more. It differs from a lock in one way: any thread may release a permit, even one that never acquired it. Nothing tracks who owns which permit.

Semaphore benches = new Semaphore(2);
// inside each of 6 packer tasks:
benches.acquire();                    // waits when both benches are busy
try {
    pack(order);                      // at most 2 threads are here at once
} finally {
    benches.release();                // always give the permit back
}
// the program counted the packers inside: most at once was 2

A latch is different. A CountDownLatch starts at a number, such as 3. Each worker calls countDown() when it finishes. A thread calling await() waits until the count reaches zero. Think of a train guard who waits until all doors have closed. It cannot be reset, so use it once.

CountDownLatch ready = new CountDownLatch(3);       // three dark stores
for (String store : stores) {
    pool.submit(() -> {
        try {
            loadStock(store);
        } finally {
            ready.countDown();                      // count down even if loading fails
        }
    });
}
boolean ok = ready.await(5, TimeUnit.SECONDS);      // true: all three are ready

Put countDown() in finally. If a worker fails and never counts down, await() waits until its timeout.

Many readers, one writer: ReadWriteLock

SwiftKart shows prices on every product screen, and prices change a few times a day. A normal lock makes two readers wait for each other, though reading does not change anything. A ReadWriteLock has two parts. Many threads may hold the read lock together. The write lock is exclusive: a writer waits for the readers to leave, and then nobody else enters until it finishes.

private final ReadWriteLock lock = new ReentrantReadWriteLock();

int price(String sku) {
    lock.readLock().lock();
    try { return prices.getOrDefault(sku, -1); }
    finally { lock.readLock().unlock(); }
}

void set(String sku, int price) {
    lock.writeLock().lock();
    try { prices.put(sku, price); }
    finally { lock.writeLock().unlock(); }
}
// two readers held the read lock at the same moment: true

It helps when reads are many, writes are rare, and each read takes noticeable time. It does not help when a read is one map lookup, because the lock's own bookkeeping costs more than the waiting it saves. It also does not help when writes are frequent. A thread holding the read lock must never ask for the write lock. That thread waits for itself forever.

If the data is small, an immutable snapshot can be simpler: build a new map when prices change, and swap one volatile reference to it.

Counters and maps: atomics and ConcurrentHashMap

The thread safety basics lesson showed that merge and compute update one key in one step. Here are the three methods side by side.

cartsWith.merge("milk", 1, Integer::sum);                       // add one, or start at 1
stock.compute("milk", (k, left) -> left > 0 ? left - 1 : left); // never goes below 0
Integer old = stock.putIfAbsent("curd", 7);                     // null: added; 7 later: kept

Eight threads running each line 1,000 times gave cartsWith 8000 and left stock at 0, never below. Keep the function inside compute short. It holds the key's lock while it runs, and it must not touch the same map.

For a plain counter, you have two choices. AtomicInteger and AtomicLong give exact reads and compareAndSet. LongAdder keeps several partial counts and adds them up when you call sum(). Many threads can increment it with little waiting, so it is usually faster under heavy contention. A sum() taken while updates continue is not a frozen snapshot. With eight threads adding 100,000 each, both reached 800000, while a plain int finished near 311,000 in the same run.

Choosing the right tool

Start from the need, not from the class name. Ask two questions first. Is the shared thing one variable, or a rule across several? And can the data be immutable, so that nothing needs protecting? The thread safety basics lesson gave the same advice, and it still applies. These tools add ways to run work and to wait for it, not new ways to protect a rule.

Now apply the first row. The starter below creates a thread for every order. Draw a DeliveryPool that owns an executor, accepts jobs and shuts down.

📌 Summary: Reuse threads with an ExecutorService and shut it down. Chain steps with CompletableFuture. Limit users with a Semaphore, wait with a latch, and use atomic map methods or LongAdder for counts.

What Arjun learns next

You now know the parts that Java gives you. Next, Arjun builds three designs that interviewers ask for by hand: a rate limiter, a bounded queue written with wait and notify, and a safe money transfer. You will also see why a plain flag can hang a program.

Check your understanding

1. Arjun makes `Executors.newFixedThreadPool(4)`, submits ten tasks, prints the results and reaches the end of `main`. The program never exits. Why?
2. A product page needs a price (200 ms) and a stock count (300 ms) from two independent calls. Which step gives one combined answer in about 300 ms?
3. A CompletableFuture uses `orTimeout(200, MILLISECONDS)` and the slow call needs 2 seconds. What happens to the slow call after 200 ms?
4. Product prices are read by thousands of threads and change twice a day. Each read is one map lookup. Which choice fits best?
5. A hits counter is incremented by 64 threads all day and read once a minute for a report. Which choice fits best?