Decouple Task Submission from Execution with Java's Executor API
Decouple Task Submission from Execution with Java’s Executor API
Every service that does work concurrently faces the same question: who runs what, when, and how many times should it spin up a new thread? The naive answer — new Thread(runnable).start() everywhere — works until you need to understand pool sizing, task queuing, result collection, or graceful shutdown. By then, your code is littered with thread handles and join() calls.
Java’s Executor API solves this by treating thread pools as a first-class abstraction. You submit work to an ExecutorService, which decides how that work gets executed — from a bounded pool, an unbounded cache, or a scheduled dispatcher. The submission code doesn’t need to know what happens next.
This post walks through four concrete patterns: creating a fixed-size pool, collecting structured results with futures, batching task execution, and racing multiple implementations against each other.
The Code
The program below shows both the old pattern and the executor-backed replacement side by side. The DirectThreadWorker class demonstrates what manual thread management looks like — every caller creates threads directly, names them manually, and coordinates their lifecycle with join().
The PoolBackedWorker class replaces all of that with an injected ExecutorService. It wraps three factory patterns (newFixedThreadPool, newCachedThreadPool, newScheduledThreadPool) behind static constructors. Each public method maps to a distinct submission pattern covered in the demos below.
import java.util.concurrent.*;
import java.util.*;
import java.util.stream.*;
class DirectThreadWorker {
void process(List<String> items) {
Thread[] threads = new Thread[Math.min(items.size(), 5)];
for (int i = 0; i < Math.min(items.size(), 5); i++) {
final int idx = i;
threads[i] = new Thread(() -> {
System.out.printf("[%s] Processing '%s'%n",
Thread.currentThread().getName(), items.get(idx));
}, "worker-" + idx);
threads[i].start();
}
for (Thread t : threads) {
try { t.join(); } catch (InterruptedException e) { break; }
}
}
}
class PoolBackedWorker {
private final ExecutorService pool;
static PoolBackedWorker ofFixed(int nThreads) {
return new PoolBackedWorker(Executors.newFixedThreadPool(nThreads));
}
CompletableFuture<String> enqueueWithResult(String taskName, String item) {
return CompletableFuture.supplyAsync(() -> {
System.out.printf("[%s] [future] '%s' → %s%n",
Thread.currentThread().getName(), taskName, item);
return "done:" + item;
}, pool);
}
List<Future<String>> enqueueBatch(List<String> items) {
List<Callable<String>> tasks = items.stream()
.map(item -> (Callable<String>) () -> {
System.out.printf("[%s] [batch] '%s'%n",
Thread.currentThread().getName(), item);
return "result:" + item;
})
.collect(Collectors.toList());
try {
return pool.invokeAll(tasks, 5, TimeUnit.SECONDS);
} catch (InterruptedException e) {
pool.shutdownNow();
throw new RuntimeException("batch interrupted", e);
}
}
String enqueueRace(List<String> items) throws Exception {
List<Callable<String>> tasks = items.stream()
.map(item -> (Callable<String>) () -> {
System.out.printf("[%s] [race] '%s' → %s%n",
Thread.currentThread().getName(), item, "fast");
return "won:" + item;
})
.collect(Collectors.toList());
return pool.invokeAny(tasks);
}
}
public class ExecutorDemo {
static void demoDirectVsExecutor() throws Exception {
var worker = new DirectThreadWorker();
worker.process(Arrays.asList("alpha", "beta", "gamma"));
}
static void demoFixedPoolFutures() throws Exception {
PoolBackedWorker worker = PoolBackedWorker.ofFixed(3);
for (int i = 1; i <= 6; i++) {
worker.enqueueWithResult("compute", "task-" + i)
.thenAccept(result -> System.out.printf("[main] Got: %s%n", result));
}
Thread.sleep(2000);
worker.shutdown();
}
static void demoInvokeAll() throws Exception {
PoolBackedWorker worker = PoolBackedWorker.ofFixed(4);
List<String> inputs = Arrays.asList("file-A", "file-B", "file-C", "file-D", "file-E");
List<Future<String>> futures = worker.enqueueBatch(inputs);
for (int i = 0; i < futures.size(); i++) {
System.out.printf("[main] Collected: %s%n", futures.get(i).get());
}
worker.shutdown();
}
static void demoInvokeAny() throws Exception {
PoolBackedWorker worker = PoolBackedWorker.ofFixed(5);
List<String> inputs = Arrays.asList("slow-task", "fast-task", "medium-task");
String winner = worker.enqueueRace(inputs);
System.out.printf("[main] Winner: %s%n", winner);
Thread.sleep(1000);
worker.shutdown();
}
public static void main(String[] args) throws Exception {
demoDirectVsExecutor();
demoFixedPoolFutures();
demoInvokeAll();
demoInvokeAny();
}
}Running It
The first demo runs 3 items through DirectThreadWorker, which creates a new Thread per item with manual names and coordinates them with join(). This is the tight coupling: every caller knows about thread naming, lifecycle management, and blocking waits.
In the second demo, we replace that pattern with ofFixed(3) — a pool of exactly three threads. We submit six tasks, each returning a CompletableFuture<String>. The key observation here is reuse: notice how pool-1-thread-1, pool-1-thread-2, and pool-1-thread-3 handle all six submissions. Tasks 4 through 6 didn’t create new threads — they slot into the queue and execute when a thread becomes available.
=== DIRECT THREAD APPROACH ===
[worker-0] Processing 'alpha'
[worker-1] Processing 'beta'
[worker-2] Processing 'gamma'
=== FIXED POOL — submit + CompletableFuture ===
[pool-1-thread-1] [future] 'compute' → task-1
[main] Got: done:task-1
[pool-1-thread-2] [future] 'compute' → task-2
[pool-1-thread-3] [future] 'compute' → task-3
[pool-1-thread-1] [future] 'compute' → task-4
[main] Got: done:task-2
[main] Got: done:task-3
[main] Got: done:task-4
[pool-1-thread-2] [future] 'compute' → task-5
[pool-1-thread-3] [future] 'compute' → task-6
[main] Got: done:task-5
[main] Got: done:task-6The results come back to the main thread through thenAccept() chains. The execution order of tasks 2 and 3 (after 1) is non-deterministic — whichever of pool-1-thread-2 or pool-1-thread-3 finishes first takes task 4. That’s by design: you submit work, you don’t control scheduling.
The third demo shows invokeAll for batch processing. Five items go into a four-thread pool. Again, the fifth item runs on thread-3 because it was already freed — only four threads ever get created despite five tasks being submitted:
=== INVOKE ALL — batch + collect results ===
[pool-2-thread-2] [batch] 'file-B'
[pool-2-thread-3] [batch] 'file-C'
[pool-2-thread-3] [batch] 'file-E'
[pool-2-thread-1] [batch] 'file-A'
[pool-2-thread-4] [batch] 'file-D'
[main] Collected: result:file-A
[main] Collected: result:file-B
[main] Collected: result:file-C
[main] Collected: result:file-D
[main] Collected: result:file-EThe returned List<Future<String>> preserves insertion order — you collect results in the same order as input regardless of which thread processed what. The caller blocks until all tasks complete (or the timeout expires).
The fourth demo demonstrates invokeAny for racing tasks. Three items — labeled “slow-task”, “fast-task”, and “medium-task” — start running simultaneously on three different threads. Only the first to return its result is delivered to the caller; the others are cancelled:
=== INVOKE ANY — race to finish first ===
[pool-3-thread-1] [race] 'slow-task' → fast
[pool-3-thread-2] [race] 'fast-task' → fast
[pool-3-thread-3] [race] 'medium-task' → fast
[main] Winner: won:slow-taskThe winner was “slow-task” in this run, which seems counterintuitive given the name. Thread scheduling is non-deterministic — if you actually need a deterministic winner, race conditions require real work differentiation (e.g., one task hitting a fast cache, another making a network call). The point isn’t that naming controls execution speed; it’s that invokeAny lets the pool pick whichever implementation finishes first, and cancels the rest.
Takeaway
The Executor API decouples what gets done from how it’s done: you write code that calls submit(), execute(), or invokeAll() on a pool, and the pool manages thread creation, queuing, reuse, and lifecycle. The three patterns — bounded pools for controlled concurrency (newFixedThreadPool), unbounded caches for bursty work (newCachedThreadPool), and scheduled dispatchers for recurring jobs (newScheduledThreadPool) — cover most use cases without ever touching Thread or join() directly.