Java’s executors run your tasks on a fixed pool of threads and hand back a Future. Learn submit, invokeAll, cancellation, CompletableFuture chains, timeouts, semaphores and blocking queues, with output that never flakes.
Starting a thread by hand for every job works in a small program. In a server it doesn’t, because each job pays for a new thread and nothing stops a burst of work from creating thousands. The java.util.concurrent package gives you executors instead: a fixed set of threads that take tasks from a queue and hand you back a result to wait on.
This post covers ExecutorService, Future, exceptions and cancellation in tasks, pool sizing, CompletableFuture, CountDownLatch, Semaphore, and two concurrent collections. Every program below was run on Java 25, and its output is pasted from the run. To run one yourself, save it as Main.java and run java Main.java. As in the part on threads and shared state, the concurrent programs force their timing with latches, so they print the same thing on every run.
Why executors: a pool of threads and a queue
A thread per task has two costs. Each platform thread is an operating system thread with its own stack, reserved at 1 MB by default on 64-bit Linux, and creating one takes real work. And the number isn’t bounded: ten thousand requests arriving at once means ten thousand threads.
An executor fixes both. Executors.newFixedThreadPool(4) makes four threads and keeps them. Tasks you hand it wait in a queue until one of the four is free. This program sends ten tasks that all block on a gate, then looks inside the pool:
void main() throws InterruptedException {
var running = new AtomicInteger();
var mostAtOnce = new AtomicInteger();
var fourStarted = new CountDownLatch(4);
var gate = new CountDownLatch(1);
var done = new AtomicInteger();
try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
for (int i = 0; i < 10; i++) {
pool.execute(() -> {
mostAtOnce.accumulateAndGet(running.incrementAndGet(), Math::max);
fourStarted.countDown();
awaitQuietly(gate);
running.decrementAndGet();
done.incrementAndGet();
});
}
fourStarted.await();
var queue = ((ThreadPoolExecutor) pool).getQueue();
IO.println("tasks running: " + running.get());
IO.println("tasks waiting in the queue: " + queue.size());
gate.countDown();
} // close() waits for every task to finish
IO.println("tasks finished: " + done.get());
IO.println("most tasks running at once: " + mostAtOnce.get());
}
void awaitQuietly(CountDownLatch latch) {
try {
latch.await();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}It prints:
tasks running: 4
tasks waiting in the queue: 6
tasks finished: 10
most tasks running at once: 4Four tasks started and got stuck at the gate. The other six sat in the queue, because there was no free thread to run them. Once main opened the gate, the four threads worked through the queue, and at no point did more than four tasks run.
execute hands the pool a Runnable and returns nothing. The cast to ThreadPoolExecutor is only there to peek at the queue. You won’t need it in real code.
The try block is try-with-resources. ExecutorService has been AutoCloseable since Java 19 (checked: javac --release 18 rejects this program), and close() stops the pool from accepting new tasks and waits for the queued ones to finish. Without it, the pool’s threads keep the JVM alive after main returns.
A fixed pool of four. Four tasks run, six wait in the queue, and the caller holds a Future for each task it submitted. The threads are made once and reused for every task.
Explain it like I’m ten
A thread pool is a restaurant kitchen with a fixed number of cooks and an order rail. The restaurant doesn’t hire a new cook for every order. You hand in an order ticket, and it goes on the rail. When a cook is free, they take the next ticket off the rail and make that dish.
In return for your ticket you get a buzzer. That’s the Future. You can sit down and do something else, and the buzzer lights up when the food’s ready. If you’d rather stand at the counter and wait, that’s get().
At closing time the manager can stop taking tickets and let the cooks finish the rail. Or the manager can pull every ticket off the rail and tell the cooks to stop now.
The precise version
newFixedThreadPool(n) returns a ThreadPoolExecutor with n worker threads and a LinkedBlockingQueue of tasks. Each worker loops: take a task from the queue, run it, repeat. submit wraps your task in a FutureTask, which is both the thing the worker runs and the Future you hold. When the task returns or throws, the FutureTask stores the result or the exception, and any thread blocked in get() wakes up.
Where the analogy breaks: the rail in newFixedThreadPool has no length limit, so a flood of tasks piles up in memory instead of being turned away. The buzzer also holds the outcome, including a failure, not just a signal. And a cook told to stop can ignore it: in Java, stopping a running task is a request the task has to honour, as the section on cancellation shows.
submit, invokeAll and invokeAny
submit takes a Callable that returns a value, or a Runnable, and gives you a Future. invokeAll submits a list of tasks and waits for all of them. invokeAny waits for the first one that succeeds:
void main() throws Exception {
try (var pool = Executors.newFixedThreadPool(3)) {
Future<Integer> length = pool.submit(() -> "executor".length());
IO.println("submit: get() returned " + length.get());
var thirdDone = new CountDownLatch(1);
List<Callable<String>> jobs = List.of(
() -> {
thirdDone.await(); // finish last on purpose
return "first";
},
() -> "second",
() -> {
thirdDone.countDown();
return "third";
});
var results = new ArrayList<String>();
for (Future<String> f : pool.invokeAll(jobs)) {
results.add(f.get());
}
IO.println("invokeAll: " + results);
List<Callable<String>> mirrors = List.of(
() -> { throw new IOException("mirror A is down"); },
() -> "downloaded from mirror B",
() -> { throw new IOException("mirror C is down"); });
IO.println("invokeAny: " + pool.invokeAny(mirrors));
}
}It prints:
submit: get() returned 8
invokeAll: [first, second, third]
invokeAny: downloaded from mirror Bget() blocks until the task has finished, then returns its value.
The first job waits until the third has finished, so the jobs finish out of order. invokeAll still returns the futures in the order you submitted the tasks, so reading them in a loop gives a fixed order however the threads ran.
invokeAny returned mirror B’s result, because the other two tasks threw. When one task succeeds, it cancels the rest. If every task fails, it throws ExecutionException carrying one of the failures.
Stopping a pool: shutdown and shutdownNow
A pool stops in one of two ways, and they differ in what happens to the queue. Here each pool has one thread, one running task blocked on a gate, and three tasks queued behind it:
void main() throws InterruptedException {
IO.println("shutdown(): " + finishedTasks(false));
IO.println("shutdownNow(): " + finishedTasks(true));
}
String finishedTasks(boolean now) throws InterruptedException {
var pool = Executors.newFixedThreadPool(1);
var started = new CountDownLatch(1);
var gate = new CountDownLatch(1);
var finished = new AtomicInteger();
pool.execute(() -> {
started.countDown();
try {
gate.await();
finished.incrementAndGet();
} catch (InterruptedException e) {
Thread.currentThread().interrupt(); // interrupted: give up
}
});
for (int i = 0; i < 3; i++) {
pool.execute(finished::incrementAndGet);
}
started.await(); // one task running, three in the queue
int neverStarted = 0;
if (now) {
neverStarted = pool.shutdownNow().size();
} else {
pool.shutdown();
}
String rejected = "";
try {
pool.execute(finished::incrementAndGet);
} catch (RejectedExecutionException e) {
rejected = "new task rejected, ";
}
gate.countDown();
boolean terminated = pool.awaitTermination(5, TimeUnit.SECONDS);
return rejected + "finished " + finished.get() + ", never started " + neverStarted
+ ", terminated " + terminated;
}It prints:
shutdown(): new task rejected, finished 4, never started 0, terminated true
shutdownNow(): new task rejected, finished 0, never started 3, terminated trueBoth methods make the pool reject new tasks with RejectedExecutionException.
shutdown() lets the running task and the whole queue finish, so all four finished. shutdownNow() removed the three queued tasks, returned them as a list, and interrupted the running one. That task caught the interrupt and gave up, so nothing finished.
awaitTermination waits up to a timeout for the pool’s threads to end, and returns whether they did. close() is roughly shutdown() followed by awaitTermination in a loop.
Exceptions inside tasks
An exception thrown inside a task doesn’t reach the code that submitted it right away. Where it goes depends on how you handed the task in. This program gives its pool a thread factory that installs an uncaught exception handler, so we can see what reaches it:
void main() throws InterruptedException {
var handlerSaw = new LinkedBlockingQueue<String>();
var threadsMade = new AtomicInteger();
ThreadFactory withHandler = task -> {
threadsMade.incrementAndGet();
var thread = new Thread(task);
thread.setUncaughtExceptionHandler((t, e) -> handlerSaw.add(e.getMessage()));
return thread;
};
try (var pool = Executors.newFixedThreadPool(1, withHandler)) {
Future<Integer> parsed = pool.submit(() -> Integer.parseInt("forty-two"));
try {
parsed.get();
} catch (ExecutionException e) {
IO.println("submit: get() threw " + e.getClass().getSimpleName());
IO.println(" cause: " + e.getCause());
}
pool.execute(() -> Integer.parseInt("seven"));
IO.println("execute: the handler got " + handlerSaw.take());
IO.println(" threads the pool has made: " + threadsMade.get());
Future<?> forgotten = pool.submit(() -> Integer.parseInt("nine"));
while (!forgotten.isDone()) {
Thread.sleep(1);
}
IO.println("submit, no get(): state is " + forgotten.state());
IO.println(" and the handler got " + handlerSaw.poll());
}
}It prints:
submit: get() threw ExecutionException
cause: java.lang.NumberFormatException: For input string: "forty-two"
execute: the handler got For input string: "seven"
threads the pool has made: 2
submit, no get(): state is FAILED
and the handler got nullWith submit, the Future catches the exception and stores it. get() throws ExecutionException, and getCause() is the original NumberFormatException.
With execute, there’s no Future to hold it. The exception escapes the task and kills the worker thread, so the thread’s uncaught exception handler gets it, and the pool makes a replacement thread. Without a custom handler, the default one prints the stack trace to standard error, and main never hears about it.
The third case is the one that loses errors. A task sent with submit failed, and nobody called get(). The handler got nothing, and nothing was printed anywhere. Future.state(), added in Java 19, shows FAILED, but only if someone asks. When you submit a task, keep its Future and call get() on it.
Timeouts and cancellation
get has a version with a timeout, and cancel(true) interrupts a task that’s already running. Here the task waits on a latch that never opens, so the 50 ms timeout is certain to expire:
void main() throws Exception {
var neverOpens = new CountDownLatch(1);
var started = new CountDownLatch(1);
var sawInterrupt = new CountDownLatch(1);
try (var pool = Executors.newFixedThreadPool(1)) {
Future<String> slow = pool.submit(() -> {
started.countDown();
try {
neverOpens.await();
return "finished";
} catch (InterruptedException e) {
sawInterrupt.countDown();
throw e;
}
});
started.await();
try {
slow.get(50, TimeUnit.MILLISECONDS);
} catch (TimeoutException e) {
IO.println("get(50 ms) threw TimeoutException");
}
IO.println("the task is still running: " + !slow.isDone());
boolean cancelled = slow.cancel(true);
sawInterrupt.await();
IO.println("cancel(true) returned " + cancelled);
IO.println("the task saw the interrupt and stopped waiting");
IO.println("state: " + slow.state());
try {
slow.get();
} catch (CancellationException e) {
IO.println("get() now throws CancellationException");
}
}
}It prints:
get(50 ms) threw TimeoutException
the task is still running: true
cancel(true) returned true
the task saw the interrupt and stopped waiting
state: CANCELLED
get() now throws CancellationExceptionTimeoutException means only that you stopped waiting. The task kept running. cancel(true) then interrupted its thread, await() threw InterruptedException, and the task ended. After that, get() throws CancellationException.
cancel(false) doesn’t interrupt. It still marks the future cancelled, but a task that’s already running carries on to the end. Either kind of cancel stops a queued task from ever starting.
Interruption is a request
An interrupt doesn’t stop a thread. It sets a flag on the thread, and blocking methods such as await, sleep and BlockingQueue.take notice the flag and throw InterruptedException. When they throw, they clear the flag. So a task that catches the exception and carries on has erased the request.
These two tasks are identical except for one line in the catch:
void main() throws InterruptedException {
var neverOpens = new CountDownLatch(1);
var bothStarted = new CountDownLatch(2);
var politeStopped = new CountDownLatch(1);
var rudeIgnoredIt = new CountDownLatch(1);
var pool = Executors.newFixedThreadPool(2, Thread.ofPlatform().daemon().factory());
pool.execute(() -> {
bothStarted.countDown();
while (!Thread.currentThread().isInterrupted()) {
try {
neverOpens.await();
} catch (InterruptedException e) {
Thread.currentThread().interrupt(); // put the flag back
}
}
politeStopped.countDown();
});
pool.execute(() -> {
bothStarted.countDown();
while (!Thread.currentThread().isInterrupted()) {
try {
neverOpens.await();
} catch (InterruptedException e) {
rudeIgnoredIt.countDown(); // swallowed: the flag stays clear
}
}
});
bothStarted.await();
pool.shutdownNow(); // interrupts both tasks
politeStopped.await();
rudeIgnoredIt.await();
IO.println("the task that restored the flag stopped");
IO.println("the task that swallowed the interrupt went back to waiting");
IO.println("pool terminated within 200 ms: "
+ pool.awaitTermination(200, TimeUnit.MILLISECONDS));
}It prints:
the task that restored the flag stopped
the task that swallowed the interrupt went back to waiting
pool terminated within 200 ms: falseThe first task put the flag back with Thread.currentThread().interrupt(), so its loop condition saw it and the task ended. The second swallowed the exception, the flag stayed clear, and the loop went straight back to waiting. shutdownNow sent one interrupt and has no second one to send, so the pool can’t finish. Its threads are daemon threads here only so the program can exit.
The rule: either let InterruptedException propagate, or catch it and call Thread.currentThread().interrupt(). A long loop with no blocking calls should check Thread.currentThread().isInterrupted() itself. The static Thread.interrupted() also reads the flag, but it clears it too, so use it only when you’re handling the interrupt there.
How many threads?
The right pool size depends on what the tasks spend their time doing, and there’s no exact answer to look up.
CPU-bound tasks, such as parsing, compressing or number crunching, need a core the whole time. More threads than cores means threads take turns, and switching between them costs time. Start near Runtime.getRuntime().availableProcessors().
IO-bound tasks, such as calls to a database or another service, spend most of their time waiting. A waiting thread uses no CPU, so more threads than cores can help.
The classic rule of thumb, from Java Concurrency in Practice, is:
threads = cores × (1 + wait time / compute time)
A task that waits 90 ms and computes 10 ms on an 8-core machine gives 8 × (1 + 9) = 80 threads. Treat that as a first guess, not an answer. The database may allow only 20 connections, and then 80 threads just queue in front of it. Measure under real load and adjust.
For IO-bound work, Java 21’s virtual threads usually remove the question. The next part covers them.
CompletableFuture: chaining steps
A Future gives you one way to use the result, which is blocking in get(). A CompletableFuture lets you attach the next step, and it runs when the value arrives:
record User(int id, String name) {}
void main() {
try (var pool = Executors.newFixedThreadPool(4)) {
CompletableFuture<User> user = CompletableFuture.supplyAsync(() -> findUser(7), pool);
CompletableFuture<String> name = user.thenApply(User::name);
IO.println("thenApply: " + name.join());
CompletableFuture<CompletableFuture<Integer>> nested =
user.thenApply(u -> cartTotal(u, pool));
CompletableFuture<Integer> total = user.thenCompose(u -> cartTotal(u, pool));
IO.println("thenApply, nested: " + nested.join().join());
IO.println("thenCompose: " + total.join());
CompletableFuture<Integer> shipping = CompletableFuture.supplyAsync(() -> 5, pool);
CompletableFuture<Integer> toPay = total.thenCombine(shipping, Integer::sum);
IO.println("thenCombine: " + toPay.join());
}
}
User findUser(int id) {
return new User(id, "Ana"); // stands in for a database call
}
CompletableFuture<Integer> cartTotal(User u, Executor pool) {
return CompletableFuture.supplyAsync(() -> 37, pool); // stands in for another service
}It prints:
thenApply: Ana
thenApply, nested: 37
thenCompose: 37
thenCombine: 42supplyAsync runs the supplier on the executor you pass. thenApply transforms the value when it’s ready, like map on a stream.
thenApply and thenCompose differ when the next step returns a future itself. With thenApply, you get a future of a future, CompletableFuture<CompletableFuture<Integer>>, and need two join() calls. thenCompose flattens it into one CompletableFuture<Integer>, like flatMap. thenCombine waits for two independent futures and merges their values.
join() waits like get(), but it throws no checked exceptions, so it fits inside lambdas.
Pass an executor. Without one, supplyAsync uses ForkJoinPool.commonPool(). That pool is shared with parallel streams and everything else in the JVM, and it’s sized for CPU work, one thread fewer than the machine has cores. Blocking calls there can starve unrelated code. Its threads are also daemon threads (checked: Thread.currentThread().isDaemon() returned true inside one), so the JVM can exit while your task is still running.
Waiting for several with allOf
CompletableFuture.allOf returns a future that completes when every future you pass has completed. It carries no values, so you read each one afterwards. Here each forecast waits for the one before it, so they finish in reverse:
void main() {
var finishOrder = new ConcurrentLinkedQueue<String>();
try (var pool = Executors.newFixedThreadPool(3)) {
CompletableFuture<?> nothing = CompletableFuture.completedFuture(null);
// each forecast waits for the one before it, so they finish in reverse
var beijing = forecast("Beijing", nothing, finishOrder, pool);
var madrid = forecast("Madrid", beijing, finishOrder, pool);
var lisbon = forecast("Lisbon", madrid, finishOrder, pool);
CompletableFuture.allOf(lisbon, madrid, beijing).join();
IO.println("finished: " + finishOrder);
for (var f : List.of(lisbon, madrid, beijing)) {
IO.println(f.join());
}
}
}
CompletableFuture<String> forecast(
String city, CompletableFuture<?> after, Queue<String> finishOrder, Executor pool) {
return CompletableFuture.supplyAsync(() -> {
after.join();
finishOrder.add(city);
int celsius = switch (city) {
case "Lisbon" -> 24;
case "Madrid" -> 31;
default -> 28;
};
return city + ": " + celsius + "C";
}, pool);
}It prints:
finished: [Beijing, Madrid, Lisbon]
Lisbon: 24C
Madrid: 31C
Beijing: 28CThe forecasts finished Beijing first. After allOf(...).join(), every future is done, so each join() in the loop returns at once. The loop reads them in the order you chose, so the output order is fixed.
When a stage fails
A CompletableFuture wraps a failure differently depending on how you wait for it:
void main() throws InterruptedException {
try (var pool = Executors.newFixedThreadPool(2)) {
CompletableFuture<Integer> port =
CompletableFuture.supplyAsync(() -> Integer.parseInt("eighty"), pool);
try {
port.join();
} catch (CompletionException e) {
IO.println("join() threw CompletionException");
IO.println(" cause: " + e.getCause());
}
try {
port.get();
} catch (ExecutionException e) {
IO.println("get() threw ExecutionException");
IO.println(" cause: " + e.getCause());
}
int withDefault = port.exceptionally(ex -> 8080).join();
IO.println("exceptionally: " + withDefault);
String report = port
.thenApply(p -> "listening on " + p)
.handle((value, ex) -> {
if (ex == null) {
return value;
}
Throwable real = ex instanceof CompletionException ? ex.getCause() : ex;
return "handle got " + ex.getClass().getSimpleName()
+ ", real problem: " + real.getMessage();
})
.join();
IO.println(report);
}
}It prints:
join() threw CompletionException
cause: java.lang.NumberFormatException: For input string: "eighty"
get() threw ExecutionException
cause: java.lang.NumberFormatException: For input string: "eighty"
exceptionally: 8080
handle got CompletionException, real problem: For input string: "eighty"join() wraps the failure in an unchecked CompletionException. get() wraps it in a checked ExecutionException, just like Future.get(). In both, getCause() is the real exception.
exceptionally runs only on failure and supplies a replacement value. handle runs either way and receives the value or the exception, one of them null.
This surprised us: handle received a CompletionException, not the NumberFormatException. A failure passed along from an earlier stage arrives wrapped. We also found that exceptionally called directly on the supplyAsync future got the wrapper too. So unwrap it before you look at it, as the program does.
Timeouts on a CompletableFuture
orTimeout fails the future after a delay, and completeOnTimeout completes it with a fallback value instead. Both lookups here wait on a latch that never opens:
void main() {
var neverOpens = new CountDownLatch(1);
var slowTaskEnded = new CountDownLatch(2);
try (var pool = Executors.newFixedThreadPool(2)) {
Supplier<String> slowLookup = () -> {
try {
neverOpens.await();
return "live price";
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return "interrupted";
} finally {
slowTaskEnded.countDown();
}
};
CompletableFuture<String> strict = CompletableFuture
.supplyAsync(slowLookup, pool)
.orTimeout(50, TimeUnit.MILLISECONDS);
try {
strict.join();
} catch (CompletionException e) {
IO.println("orTimeout: " + e.getCause().getClass().getSimpleName());
}
String price = CompletableFuture
.supplyAsync(slowLookup, pool)
.completeOnTimeout("cached price", 50, TimeUnit.MILLISECONDS)
.join();
IO.println("completeOnTimeout: " + price);
IO.println("slow lookups that have ended: " + (2 - slowTaskEnded.getCount()));
neverOpens.countDown(); // let them finish, or close() would wait forever
}
IO.println("slow lookups that have ended: " + (2 - slowTaskEnded.getCount()));
}It prints:
orTimeout: TimeoutException
completeOnTimeout: cached price
slow lookups that have ended: 0
slow lookups that have ended: 2orTimeout failed the future with TimeoutException, wrapped in CompletionException by join(). completeOnTimeout gave the cached price.
Neither one stopped the work. After both timeouts, zero slow lookups had ended: their threads were still blocked. main had to open the latch, or close() would have waited forever. cancel(true) doesn’t help either. Its documentation says the interrupt flag has no effect, and we checked: the task never saw an interrupt. Once main opened the latch, both lookups ended, which is the last line. A timeout on a CompletableFuture stops the waiting, not the task.
Coordination helpers
java.util.concurrent also has small tools for making threads wait for each other. CountDownLatch, which the part on threads used, lets one thread wait until others have finished something:
void main() throws InterruptedException {
var services = List.of("search", "cache", "database");
var allReady = new CountDownLatch(services.size());
var ready = new ConcurrentSkipListSet<String>();
try (var pool = Executors.newFixedThreadPool(3)) {
for (String service : services) {
pool.execute(() -> {
ready.add(service); // stands in for slow start-up work
allReady.countDown();
});
}
allReady.await();
IO.println("every service is up: " + ready);
}
}It prints:
every service is up: [cache, database, search]The latch starts at 3, each service counts down once, and await() returns when it reaches 0. A ConcurrentSkipListSet keeps its elements sorted, so the printed order is fixed.
A Semaphore holds a number of permits. acquire() takes one, or waits if none are left, and release() gives it back. It limits how many threads do something at once, such as calling a service that accepts two connections. Six tasks share two permits here, and a latch makes sure two really are inside together:
void main() throws InterruptedException {
var permits = new Semaphore(2);
var inside = new AtomicInteger();
var mostInside = new AtomicInteger();
var twoInside = new CountDownLatch(2);
var gate = new CountDownLatch(1);
try (var pool = Executors.newFixedThreadPool(6)) {
for (int i = 0; i < 6; i++) {
pool.execute(() -> {
try {
permits.acquire();
try {
mostInside.accumulateAndGet(inside.incrementAndGet(), Math::max);
twoInside.countDown();
gate.await();
inside.decrementAndGet();
} finally {
permits.release();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
}
twoInside.await();
while (permits.getQueueLength() < 4) {
Thread.sleep(1);
}
IO.println("holding a permit: " + inside.get());
IO.println("waiting for one: " + permits.getQueueLength());
gate.countDown();
}
IO.println("most inside at once: " + mostInside.get());
}It prints:
holding a permit: 2
waiting for one: 4
most inside at once: 2The first two tasks took the permits and waited at the gate. The other four waited in acquire(). The high-water mark is 2. The twoInside latch forced two holders inside together, and the semaphore kept a third one out. Release in finally, as with a lock, or a failed task keeps its permit forever.
A CyclicBarrier makes a fixed number of threads wait until all of them reach the same point, then lets them all continue together.
Concurrent collections, revisited
The part on threads showed that a ConcurrentHashMap call is safe on its own, but check-then-act across two calls isn’t. The compute methods put the whole update inside one call:
void main() {
var stock = new ConcurrentHashMap<String, Integer>();
stock.put("tea", 20);
stock.put("cake", 3);
var requests = new ConcurrentHashMap<String, Integer>();
var basket = List.of("tea", "cake", "tea", "soup", "cake", "tea");
try (var pool = Executors.newFixedThreadPool(4)) {
for (int i = 0; i < 4; i++) {
pool.execute(() -> {
for (String item : basket) {
requests.merge(item, 1, Integer::sum);
// take one; the last one removes the entry
stock.computeIfPresent(item, (name, left) -> left == 1 ? null : left - 1);
}
});
}
}
IO.println("requests: " + new TreeMap<>(requests));
IO.println("stock: " + new TreeMap<>(stock));
}It prints:
requests: {cake=8, soup=4, tea=12}
stock: {tea=8}Four tasks made 24 requests. merge counted them with no lost updates. computeIfPresent took one item at a time, and returning null removed the entry, so cake sold out after three and disappeared. Soup was never in stock, so the function never ran for it.
A BlockingQueue passes work from one thread to another. put waits while the queue is full, and take waits while it’s empty:
void main() throws Exception {
var queue = new ArrayBlockingQueue<String>(2);
var received = new ArrayList<String>();
try (var pool = Executors.newFixedThreadPool(2)) {
Future<?> producer = pool.submit(() -> {
for (int i = 1; i <= 5; i++) {
queue.put("order-" + i); // waits while the queue is full
}
queue.put("DONE");
return null;
});
Future<?> consumer = pool.submit(() -> {
String order = queue.take(); // waits while the queue is empty
while (!order.equals("DONE")) {
received.add(order);
order = queue.take();
}
return null;
});
producer.get();
consumer.get();
}
IO.println("consumer received: " + received);
}It prints:
consumer received: [order-1, order-2, order-3, order-4, order-5]With a capacity of 2, the producer can’t race ahead of the consumer by more than two orders. The "DONE" value tells the consumer to stop. With one producer and one consumer, orders come out in the order they went in. With several of either, each queue is still safe, but the order across threads isn’t fixed.
Both tasks are Callable lambdas, since they end with return null. That lets put and take throw InterruptedException into the Future without a try.
What to remember
- An executor reuses a fixed set of threads and queues the extra tasks. Open it with try-with-resources, so
close()waits for the work to finish. submitreturns aFuture. Callget()on it, or a failed task’s exception is silently lost.invokeAllreturns futures in submission order.get()throwsExecutionException, andjoin()throwsCompletionException. The real exception is the cause.- A timeout stops the waiting, not the task. Interruption is a request, so restore the flag when you catch
InterruptedException. - Size a pool for CPU work near the core count. For IO work, use the wait-to-compute rule of thumb as a starting guess, then measure.
- Give
CompletableFutureyour own executor. UsethenComposewhen the next step returns a future, andallOfthenjoinin a fixed order. - A
Semaphorelimits how many threads do something at once. ABlockingQueuehands work between threads in order.
Hand tasks to a pool, keep every future, and decide up front what happens when a task fails or runs too long.