Java ConcurrencyTopic 8 of 16

LanguageJavaIntermediateAsked Often

ExecutorService & Thread Pools

In one line

ExecutorService manages a pool of threads + a work queue. Use ThreadPoolExecutor with bounded queue + named threads + sensible rejection policy in production. newFixedThreadPool/newCachedThreadPool have unbounded queues, production hazards.

Why ExecutorService

new Thread() + .start() is for examples. ExecutorService is for production. It manages thread lifecycle (creation, reuse, shutdown), bounds concurrency (queue + max threads), routes failures (rejection policy), and provides Futures for results.

Don't use Executors.newXxx factories in production, they have hidden defaults that bite under load. Use ThreadPoolExecutor directly with all parameters spelled out.

Important

The four production essentials

Bounded work queue, ArrayBlockingQueue(N). Limits memory under burst.
Named threads, ThreadFactoryBuilder().setNameFormat("..."). Critical for thread dumps.
UncaughtExceptionHandler, log uncaught exceptions. Default behavior silently drops them.
CallerRunsPolicy, natural backpressure when queue is full.

Three patterns of pool sizing

// CPU-bound, match cores
int cores = Runtime.getRuntime().availableProcessors();
ExecutorService cpuPool = newBoundedPool(cores, cores, 1000);

// I/O-bound, much higher
// Goetz's formula: cores * targetUtil * (1 + waitTime/computeTime)
// 8 cores * 1.0 * (1 + 50ms/5ms) ≈ 88
ExecutorService ioPool = newBoundedPool(88, 88, 1000);

// Java 21 virtual threads, request-per-thread, unsized
ExecutorService virtual = Executors.newVirtualThreadPerTaskExecutor();

The shutdown story

pool.shutdown();                         // no new tasks; finish current
if (!pool.awaitTermination(10, SECONDS)) {
    pool.shutdownNow();                  // interrupt running tasks
    if (!pool.awaitTermination(5, SECONDS)) {
        log.error("did not terminate");
    }
}

Two phases: graceful (let in-flight finish) and forceful (interrupt + drop). Always pair with timeouts. Always log failures. Test it, many production services have never been gracefully shut down because no one tested it.

Java 21 changes the game

Executors.newVirtualThreadPerTaskExecutor() is a request-per-thread executor with effectively no thread limit. For I/O-heavy services this is the right answer in Java 21+, simpler than reactive, scales better than fixed pools.

Warning

Don't blindly migrate to virtual threads

Synchronized blocks pin to carriers; I/O inside synchronized stalls the carrier. Switch to ReentrantLock.
ThreadLocal proliferation, each virtual thread has its own; with many ThreadLocals, memory growth scales with task count.
Bounded concurrency must come from elsewhere, virtual thread pool doesn't bound; add a Semaphore around external service calls.

Primitives by language

ExecutorService / ThreadPoolExecutor
Executors.newFixedThreadPool / newCachedThreadPool / newSingleThreadExecutor
Executors.newVirtualThreadPerTaskExecutor (Java 21+)
BlockingQueue (work queue)
ThreadFactory (named threads, daemon flags)
RejectedExecutionHandler (overflow policy)

Implementation

Production-grade ThreadPoolExecutor

Spell out everything: core/max threads, keep-alive, bounded queue, named threads, rejection policy. CallerRunsPolicy provides natural backpressure when queue fills.

 1  import java.util.concurrent.*;
 2  import com.google.common.util.concurrent.ThreadFactoryBuilder;
 3  
 4  ExecutorService pool = new ThreadPoolExecutor(
 5      8, 8,                                          // core, max threads
 6      0L, TimeUnit.MILLISECONDS,                     // keep-alive (irrelevant for fixed)
 7      new ArrayBlockingQueue<>(1000),                // BOUNDED queue, backpressure
 8      new ThreadFactoryBuilder()
 9          .setNameFormat("worker-%d")
10          .setUncaughtExceptionHandler((t, e) ->
11              log.error("uncaught in {}", t.getName(), e))
12          .build(),
13      new ThreadPoolExecutor.CallerRunsPolicy()      // overflow → submitter runs
14  );

Graceful shutdown, two-phase

shutdown() rejects new tasks; existing tasks finish. shutdownNow() interrupts running tasks + drains queue. Always pair with awaitTermination and a timeout. Never forget to shutdown.

 1  ExecutorService pool = Executors.newFixedThreadPool(8);
 2  // ... submit tasks ...
 3  
 4  pool.shutdown();
 5  try {
 6      if (!pool.awaitTermination(10, TimeUnit.SECONDS)) {
 7          List<Runnable> dropped = pool.shutdownNow();
 8          log.warn("{} tasks dropped at shutdown", dropped.size());
 9          if (!pool.awaitTermination(5, TimeUnit.SECONDS)) {
10              log.error("Pool did not terminate cleanly");
11          }
12      }
13  } catch (InterruptedException e) {
14      pool.shutdownNow();
15      Thread.currentThread().interrupt();
16  }

Submit + Future.get with timeout

submit(Callable) returns a Future. get() blocks for the result; get(timeout, unit) blocks with a deadline. The latter is essential for tail-latency-bounded services.

 1  Future<String> future = pool.submit(() -> httpClient.get(url));
 2  try {
 3      String result = future.get(2, TimeUnit.SECONDS);
 4      return result;
 5  } catch (TimeoutException e) {
 6      future.cancel(true);                       // interrupt the worker
 7      return fallback();
 8  } catch (ExecutionException e) {
 9      // task threw, e.getCause() is the original exception
10      throw new ServiceException("fetch failed", e.getCause());
11  }

invokeAll, parallel fan-out with deadlines

Submit a list of Callables; await all results. With timeout, any unfinished tasks are cancelled. Cleaner than managing N Futures manually.

 1  List<Callable<String>> tasks = urls.stream()
 2      .map(url -> (Callable<String>) () -> httpClient.get(url))
 3      .toList();
 4  
 5  List<Future<String>> futures = pool.invokeAll(tasks, 5, TimeUnit.SECONDS);
 6  List<String> results = new ArrayList<>();
 7  for (Future<String> f : futures) {
 8      if (f.isCancelled()) {
 9          results.add(null);                     // timed out
10      } else {
11          try { results.add(f.get()); }
12          catch (Exception e) { results.add(null); }
13      }
14  }

Java 21, virtual thread per task executor

Each submitted task gets its own virtual thread. Effectively unbounded concurrency. Pair with bounded semaphores for backpressure on downstream services.

1  try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
2      List<Future<String>> futures = urls.stream()
3          .map(url -> executor.submit(() -> httpClient.get(url)))
4          .toList();
5      for (Future<String> f : futures) {
6          process(f.get());
7      }
8  } // auto-close awaits all submitted tasks

ScheduledExecutorService, for delayed/periodic tasks

schedule runs once after delay. scheduleAtFixedRate runs periodically (start-to-start). scheduleWithFixedDelay runs periodically (end-to-start).

 1  ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
 2  
 3  // Run once after 5s
 4  scheduler.schedule(() -> log.info("delayed"), 5, TimeUnit.SECONDS);
 5  
 6  // Run every 10s, regardless of how long previous took
 7  scheduler.scheduleAtFixedRate(this::checkHealth, 0, 10, TimeUnit.SECONDS);
 8  
 9  // Run with 10s GAP between end of one and start of next
10  scheduler.scheduleWithFixedDelay(this::poll, 0, 10, TimeUnit.SECONDS);

Key points

•ThreadPoolExecutor (the actual class), full control
•Executors.newXxx, convenience but with footguns (unbounded queues)
•Always: bounded queue, named threads, explicit rejection policy
•shutdown() vs shutdownNow(), graceful vs forceful
•Java 21+ virtual thread executor, request-per-thread reborn

Follow-up questions

▸Why is newFixedThreadPool dangerous in production?

It uses an UNBOUNDED LinkedBlockingQueue. Burst load → queue grows without bound → OOM. Always use ThreadPoolExecutor explicitly with ArrayBlockingQueue(N) or another bounded queue.

▸What's the right rejection policy?

Default (AbortPolicy) throws RejectedExecutionException. CallerRunsPolicy makes the submitter run the task, natural backpressure. DiscardPolicy silently drops. DiscardOldestPolicy drops the oldest queued. CallerRuns is the most-used in production.

▸How to size a thread pool?

CPU-bound: pool = NumCPU. I/O-bound: Goetz's formula, cores × (1 + waitTime/computeTime). For Java 21 virtual threads: don't size at all; use newVirtualThreadPerTaskExecutor with semaphores for downstream-protection.

▸Is it necessary to shutdown the executor?

Yes. Without shutdown, the JVM doesn't exit because non-daemon worker threads stay alive. Always call shutdown() (or use try-with-resources in Java 19+). For tests, shutdown in @AfterEach to avoid thread leaks.

Gotchas

!Executors.newFixedThreadPool: unbounded queue → potential OOM
!Executors.newCachedThreadPool: unbounded thread count → potential OS thread exhaustion
!Forgetting to call shutdown() → JVM doesn't exit
!Future.get() without timeout → can wait forever if task hangs
!Worker thread throwing uncaught exception → killed silently; pool spawns replacement; bug invisible without UncaughtExceptionHandler

Related reading

Why ExecutorService

Don't use Executors.newXxx factories in production, they have hidden defaults that bite under load. Use ThreadPoolExecutor directly with all parameters spelled out.

Important

The four production essentials

Bounded work queue, ArrayBlockingQueue(N). Limits memory under burst.
Named threads, ThreadFactoryBuilder().setNameFormat("..."). Critical for thread dumps.
UncaughtExceptionHandler, log uncaught exceptions. Default behavior silently drops them.
CallerRunsPolicy, natural backpressure when queue is full.

Three patterns of pool sizing

// CPU-bound, match cores int cores = Runtime.getRuntime().availableProcessors(); ExecutorService cpuPool = newBoundedPool(cores, cores, 1000); // I/O-bound, much higher // Goetz's formula: cores * targetUtil * (1 + waitTime/computeTime) // 8 cores * 1.0 * (1 + 50ms/5ms) ≈ 88 ExecutorService ioPool = newBoundedPool(88, 88, 1000); // Java 21 virtual threads, request-per-thread, unsized ExecutorService virtual = Executors.newVirtualThreadPerTaskExecutor();

The shutdown story

pool.shutdown(); // no new tasks; finish current if (!pool.awaitTermination(10, SECONDS)) { pool.shutdownNow(); // interrupt running tasks if (!pool.awaitTermination(5, SECONDS)) { log.error("did not terminate"); } }

Java 21 changes the game

Warning

Don't blindly migrate to virtual threads

Synchronized blocks pin to carriers; I/O inside synchronized stalls the carrier. Switch to ReentrantLock.
ThreadLocal proliferation, each virtual thread has its own; with many ThreadLocals, memory growth scales with task count.
Bounded concurrency must come from elsewhere, virtual thread pool doesn't bound; add a Semaphore around external service calls.

1 import java.util.concurrent.*; 2 import com.google.common.util.concurrent.ThreadFactoryBuilder; 3 4 ExecutorService pool = new ThreadPoolExecutor( 5 8, 8, // core, max threads 6 0L, TimeUnit.MILLISECONDS, // keep-alive (irrelevant for fixed) 7 new ArrayBlockingQueue<>(1000), // BOUNDED queue, backpressure 8 new ThreadFactoryBuilder() 9 .setNameFormat("worker-%d") 10 .setUncaughtExceptionHandler((t, e) -> 11 log.error("uncaught in {}", t.getName(), e)) 12 .build(), 13 new ThreadPoolExecutor.CallerRunsPolicy() // overflow → submitter runs 14 );

1 ExecutorService pool = Executors.newFixedThreadPool(8); 2 // ... submit tasks ... 3 4 pool.shutdown(); 5 try { 6 if (!pool.awaitTermination(10, TimeUnit.SECONDS)) { 7 List<Runnable> dropped = pool.shutdownNow(); 8 log.warn("{} tasks dropped at shutdown", dropped.size()); 9 if (!pool.awaitTermination(5, TimeUnit.SECONDS)) { 10 log.error("Pool did not terminate cleanly"); 11 } 12 } 13 } catch (InterruptedException e) { 14 pool.shutdownNow(); 15 Thread.currentThread().interrupt(); 16 }

1 Future<String> future = pool.submit(() -> httpClient.get(url)); 2 try { 3 String result = future.get(2, TimeUnit.SECONDS); 4 return result; 5 } catch (TimeoutException e) { 6 future.cancel(true); // interrupt the worker 7 return fallback(); 8 } catch (ExecutionException e) { 9 // task threw, e.getCause() is the original exception 10 throw new ServiceException("fetch failed", e.getCause()); 11 }

1 List<Callable<String>> tasks = urls.stream() 2 .map(url -> (Callable<String>) () -> httpClient.get(url)) 3 .toList(); 4 5 List<Future<String>> futures = pool.invokeAll(tasks, 5, TimeUnit.SECONDS); 6 List<String> results = new ArrayList<>(); 7 for (Future<String> f : futures) { 8 if (f.isCancelled()) { 9 results.add(null); // timed out 10 } else { 11 try { results.add(f.get()); } 12 catch (Exception e) { results.add(null); } 13 } 14 }

1 try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) { 2 List<Future<String>> futures = urls.stream() 3 .map(url -> executor.submit(() -> httpClient.get(url))) 4 .toList(); 5 for (Future<String> f : futures) { 6 process(f.get()); 7 } 8 } // auto-close awaits all submitted tasks

1 ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2); 2 3 // Run once after 5s 4 scheduler.schedule(() -> log.info("delayed"), 5, TimeUnit.SECONDS); 5 6 // Run every 10s, regardless of how long previous took 7 scheduler.scheduleAtFixedRate(this::checkHealth, 0, 10, TimeUnit.SECONDS); 8 9 // Run with 10s GAP between end of one and start of next 10 scheduler.scheduleWithFixedDelay(this::poll, 0, 10, TimeUnit.SECONDS);