14.3 Thread Pools and Task Results

After learning about Thread Creation and Synchronization, Communication, and Deadlocks, spend about 90 minutes completing this Java 17 lab. You'll configure a bounded thread pool, observe task failures and rejection, and shut down the executor in stages.

Workstations and waiting areas are limited

A thread pool solves three things: reusing worker threads, limiting concurrent execution, and managing task backlog. Setting only the thread count while leaving an unbounded queue still fails to complete capacity design.

1. Moving from Threads to Executors

ExecutorService pool = Executors.newFixedThreadPool(4);
try {
    Future<String> future = pool.submit(() -> "catalog");
    String result = future.get();
} finally {
    pool.shutdown();
}

• Executor defines only execute(Runnable), separating submission from execution;
• ExecutorService adds submission, bulk tasks, shutdown, and termination waiting;
• ThreadPoolExecutor exposes capacity parameters such as work thread, queue, factory, and rejection strategy;
• ScheduledExecutorService handles delays and periodic tasks.

Business objects typically depend on Executor or ExecutorService interfaces; the application assembly layer is responsible for creating and shutting down specific thread pools.

2. ThreadPoolExecutor How to Receive Tasks

ThreadPoolExecutor pool = new ThreadPoolExecutor(
        2,                               // corePoolSize
        4,                               // maximumPoolSize
        30, TimeUnit.SECONDS,            // Non-core idle thread survival time
        new ArrayBlockingQueue<>(100),   // Bounded waiting area
        threadFactory,
        new ThreadPoolExecutor.AbortPolicy());

While the executor is running and its thread factory can create workers, submission follows this outline. A concurrent shutdown can also reject a task after an attempted enqueue:

1. The current number of worker threads is less than corePoolSize; create a core thread;
2. Otherwise, try adding the task to the work queue;
3. The queue is full and the number of threads is less than maximumPoolSize; then create a thread;
4. If the queue is full, the thread count has reached its limit, or the executor has been shut down, hand it over to the rejection strategy.

A common misconception is that "as soon as maximumPoolSize is set, the pool will expand." If the queue is unbounded, tasks typically just accumulate in the queue, and queue saturation does not normally trigger growth beyond the core size.

3. Thread Factory Makes Worker Threads Identifiable

AtomicInteger number = new AtomicInteger();
ThreadFactory factory = task -> {
    Thread thread = new Thread(task,
            "library-worker-" + number.incrementAndGet());
    thread.setDaemon(false);
    thread.setUncaughtExceptionHandler((t, error) ->
            System.err.println(t.getName() + " Uncaught exception:" + error));
    return thread;
};

Thread names appear in logs, monitoring tools, and thread dumps. Ordinary business pools typically use non-daemon threads and are explicitly shut down by their owner; do not rely on the JVM terminating to complete data flushing for daemon threads.

A thread factory creates workers. Install request context for each task and restore or clear it in finally. Setting security, logging, or ThreadLocal context only when a worker is created does not handle the different requests that worker will later execute.

4. Queue and Rejection Strategies for Overload Policies

Strategy	Behavior	Risk and Applicable Boundaries
AbortPolicy	Throw RejectedExecutionException	Default; caller must handle or report upward
CallerRunsPolicy	Run the task synchronously in the submitting thread	Can slow down the submit end but increases call latency; tasks are discarded after shutdown
DiscardPolicy	Silently discard new tasks	Consider only when business explicitly allows discarding and metrics are already available
DiscardOldestPolicy	Drop the front and retry submission	Disrupts order; the "oldest" semantics in a priority queue might be even more dangerous

Rejection is not a thread pool failure; it's the point at which capacity boundaries begin to take effect. The system must clearly choose:

• Return "busy, please retry" to the upstream;
• Limit or block submitters;
• Downgraded to a cheaper job;
• Write the task to an external queue with persistence and capacity governance;
• Count discarded tasks when dropping them is permitted.

With CallerRunsPolicy, maximumPoolSize limits workers inside the pool; submitting threads can execute additional tasks concurrently. Discard policies, including CallerRunsPolicy after shutdown, do not automatically cancel discarded FutureTask objects. submit() can return a Future that never completes, so cancellation or explicit rejection handling is required.

Do not hide overload with infinite waiting. Queue depth, active threads, rejection count, task wait time, and execution time should all be monitored.

5. Quick Factory Isn't Always Off-Limits

Factory	Implemented Features	Usage Conditions
newFixedThreadPool(n)	Fixed thread count, unbounded LinkedBlockingQueue	Submission volume externally strictly bounded
newSingleThreadExecutor()	Single-threaded, unbounded queue	Requires ordering, and backlog has an external upper limit
newCachedThreadPool()	SynchronousQueue, created on demand	Short tasks, concurrency already constrained externally
newScheduledThreadPool(n)	Delayed Job Queue	Scheduled tasks with failure, overlap, and shutdown strategies

"Production environments prohibit Executors" is too absolute. The real issue is whether the default capacity complies with system constraints. If it cannot be proven that backlog is bounded, explicitly use ThreadPoolExecutor and bounded queues.

6. Thread Count Starts from Measurement

CPU-intensive tasks can start with a thread count near Runtime.getRuntime().availableProcessors(); tasks involving heavy I/O waiting may allow for more threads. Both are just starting points for benchmarking.

When determining thread count, you also need to consider:

• Single-task CPU time and wait time;
• Throughput and tail latency;
• Per-thread stack and task object memory;
• Database connection pooling, remote rate limiting, and downstream capacity;
• Container CPU quota and JVM-visible cores;
• Will the task hold a lock, nest submissions, or block waiting on the same thread pool?

"CPU cores + 1" or "core count × 2" can't be a universal production rule. Set resource limits first, then measure using representative traffic.

7. Runnable, Callable, and Future

Runnable.run() returns no result and declares no checked exception. Callable<V>.call() returns V and may throw checked exceptions. This fragment assumes Book and catalog already exist and its enclosing method propagates InterruptedException; application code must also turn timeout and failure into explicit outcomes.

Future<Book> future = pool.submit(() -> catalog.load("BK-14"));

try {
    Book book = future.get(500, TimeUnit.MILLISECONDS);
} catch (TimeoutException e) {
    future.cancel(true);
} catch (ExecutionException e) {
    Throwable taskFailure = e.getCause();
}

• execute(Runnable) No result handle; uncaptured exception follows the worker thread's uncaught exception path, and the worker thread might be replaced by the pool;
• submit(...) returns Future; exceptions thrown by the task are reported by get() via ExecutionException;
• A successful get() establishes the visibility relationship from task completion to the caller;
• get(timeout, unit) A timeout ends this wait, it does not automatically stop the task;
• cancel(true) requests interruption of a running task; the task must cooperate by propagating or checking interruption;
• cancel(false) does not interrupt a running task. After successful cancellation, subsequent get() calls throw CancellationException, even if the underlying code has not yet exited.

With CallerRunsPolicy, submit() itself may run a task and block; a later get(timeout) does not bound that time.

Discarding the returned Future makes failures from submit() easy to miss. Tasks without return values should also have a consistent method for recording failures.

Experiment: Bounded Pool, Task Results, Exceptions, and Shutdown

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class ExecutorDemo {
    public static void main(String[] args) throws Exception {
        AtomicInteger number = new AtomicInteger();
        ThreadFactory factory = task -> {
            Thread thread = new Thread(task, "library-worker-" + number.incrementAndGet());
            thread.setDaemon(false);
            return thread;
        };
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                2,
                2,
                0L,
                TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(4),
                factory,
                new ThreadPoolExecutor.CallerRunsPolicy());

        boolean terminated;
        try {
            List<Future<String>> futures = List.of(
                    pool.submit(() -> "catalog"),
                    pool.submit(() -> "receipt"),
                    pool.submit(() -> "audit"));
            List<String> results = futures.stream().map(ExecutorDemo::get).toList();
            System.out.println("Task results: " + results);

            Future<String> failed = pool.submit(() -> {
                throw new IllegalStateException("Inventory snapshot corrupted");
            });
            try {
                failed.get();
                throw new AssertionError("Failed tasks should not return results");
            } catch (ExecutionException e) {
                System.out.println("Task exception: " + e.getCause().getClass().getSimpleName());
            }
        } finally {
            pool.shutdown();
            terminated = awaitTermination(pool);
        }

        System.out.println("Terminated: " + terminated);
        if (!terminated) {
            throw new AssertionError("The thread pool has not terminated");
        }
        System.out.println("Thread pool verification passed");
    }

    private static String get(Future<String> future) {
        try {
            return future.get();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("Interrupted while waiting for task", e);
        } catch (ExecutionException e) {
            throw new IllegalStateException("Task failed", e.getCause());
        }
    }

    private static boolean awaitTermination(ThreadPoolExecutor pool) {
        try {
            if (pool.awaitTermination(2, TimeUnit.SECONDS)) {
                return true;
            }
            pool.shutdownNow();
            return pool.awaitTermination(2, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            pool.shutdownNow();
            Thread.currentThread().interrupt();
            return false;
        }
    }
}

Run:

chapter14_executor_root=$(mktemp -d)
mkdir -p "$chapter14_executor_root/out"
# Save the above code as "$chapter14_executor_root/ExecutorDemo.java"
javac --release 17 -Xlint:all -Werror \
  -d "$chapter14_executor_root/out" \
  "$chapter14_executor_root/ExecutorDemo.java"
java -ea -cp "$chapter14_executor_root/out" ExecutorDemo
rm -rf -- "$chapter14_executor_root"

Output:

Task results: [catalog, receipt, audit]
Task exception: IllegalStateException
Terminated: true
Thread pool verification passed

Three tasks may run in any order, but the program takes values in the order specified by Future list, so the resulting list remains stable. Failed tasks must call get() to observe the wrapped error reason.

The queue capacity is 4, so the preceding experiment does not trigger rejection. Save this separate example as SaturationDemo.java and compile with the same Java 17 flags. Two workers remain held by a latch and one task fills the queue, so the fourth task must run in the caller. It also checks that submitting after shutdown leaves a Future requiring explicit cancellation:

import java.util.concurrent.*;

public class SaturationDemo {
    public static void main(String[] args) throws Exception {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(2, 2, 0, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1), new ThreadPoolExecutor.CallerRunsPolicy());
        CountDownLatch ready = new CountDownLatch(2);
        CountDownLatch release = new CountDownLatch(1);
        Runnable held = () -> {
            ready.countDown();
            try {
                release.await();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        };
        try {
            pool.execute(held);
            pool.execute(held);
            ready.await();
            pool.execute(() -> {}); // One queued task; both workers remain held.
            Thread caller = Thread.currentThread();
            Future<Thread> fourth = pool.submit(Thread::currentThread);
            if (fourth.get() != caller) throw new AssertionError("Expected caller execution");
            System.out.println("Caller executed fourth task: true");
        } finally {
            release.countDown();
            pool.shutdown();
            if (!pool.awaitTermination(2, TimeUnit.SECONDS)) {
                pool.shutdownNow();
                throw new AssertionError("Pool did not terminate");
            }
        }
        Future<?> discarded = pool.submit(() -> {});
        if (discarded.isDone()) throw new AssertionError("Unexpected completed future");
        discarded.cancel(false); // CallerRunsPolicy discarded this task after shutdown.
        System.out.println("Discarded future explicitly cancelled: " + discarded.isCancelled());
    }
}

Caller executed fourth task: true
Discarded future explicitly cancelled: true

8. Phased Shutdown

shutdown() no longer accepts new tasks but processes tasks already submitted. shutdownNow() attempts to interrupt running tasks and returns queued tasks that have not started; it does not guarantee that every thread disappears immediately.

Recommendation process:

1. Call shutdown();
2. Call awaitTermination() with a deadline the application can tolerate;
3. If it times out, call shutdownNow();
4. Wait for a short period and record that it has not terminated;
5. If the current thread is interrupted, call shutdownNow(), restore the interrupt status, and return promptly.

Queued tasks returned by shutdownNow() do not automatically become cancelled Futures. Their owner should cancel those Futures or notify waiters explicitly. In Java 17, ExecutorService is not yet AutoCloseable, so this lesson manages shutdown in finally.

The thread pool should be shut down by the component that created it. Don't let each borrowed service method individually shut down the shared pool, and don't create a new pool for every request.

9. Periodic Tasks and Fork/Join

Scheduling Task

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
scheduler.scheduleAtFixedRate(task, 0, 1, TimeUnit.MINUTES);

• scheduleAtFixedRate starts based on the planned start time; it prevents parallel re-entry of tasks in the same cycle if tasks are too slow, but may allow continuous catch-up;
• scheduleWithFixedDelay waits a fixed interval after the previous execution finishes;
• After a periodic task throws an uncaught exception, subsequent executions are suppressed; clear logging and recovery policies must be defined;
• The scheduler still needs to be shut down.

Fork/Join

ForkJoinPool uses work-stealing and is well-suited for CPU computations that can be recursively broken down into smaller subtasks. It's not a faster thread pool for all tasks.

Common OpenJDK implementations use the common pool for parallel streams, but the Stream API does not promise a particular executor. Long blocking tasks in a shared pool may affect other parallel streams and default CompletableFuture asynchronous tasks. For isolation or capacity bounds, submit tasks explicitly to an executor; do not assume a parallel stream belongs to a custom pool.

10. Operations Maintenance Checklist

Dimension	Questions Answered
Ownership	Who creates and who closes this pool?
Capacity	What are the maximum number of concurrent threads and maximum queued tasks?
Overload	When the queue is full, reject, degrade, block, or persist?
Delay	What are the task queue time and execution time?
Failure	Where are the exceptions from execute and submit recorded?
Cancel	Does the task respond to cancellation? Does a blocking call have a timeout?
Downstream	How many concurrent connections or remote services can the database and remote services handle?
Context	How to clean up ThreadLocal, logs, and security context?
Close	What are graceful timeout and forced cancellation strategies?

11. Common Failures

Phenomenon	Possible Causes
maximumPoolSize has never taken effect	Unbounded queue, has never entered scaling steps
Memory usage continues to rise	Accumulated queue tasks and their captured objects
Thread count increases rapidly	cached pool or custom upper limit is too large
submit() task failed but logs show no errors	Discarded Future, did not call get()
Main method ends, process still running	Non-daemon threads still alive, pool not closed
After shutdownNow(), the task continues to run	The task swallows interruption or calls an uninterruptible API
CallerRunsPolicy causes request slowdown	The submission thread is handling an overloaded task, which is its backpressure mechanism
Tasks wait for one another inside the pool	Thread starvation deadlock: tasks synchronously wait for subtasks in the same small pool

12. Practice

1. Change the experiment queue capacity to 1, use a latch to stabilize two worker threads and one queue slot, verify that the next submission executes CallerRunsPolicy. Assert the execution thread name, rather than relying on timing.
2. Add a 100-millisecond timeout to a task. After the timeout, invoke cancel(true), have the task restore interrupt status after catching InterruptedException and then exit.
3. Run a task that throws an exception using execute() and submit(), and record where the exception is caught.
4. Library Capacity Statement: Peak submission rate, single-task duration, queue limit, response rejection, downstream connection cap, and shutdown timeout.

Verify capacity agreement before closing

Change the experiment to two worker threads, one queue slot, use a latch to stabilize occupancy of the worker and queue positions, then submit the fourth task. If using CallerRunsPolicy, the fourth task should be executed in the submitting thread; test the thread name, don't use sleep to guess that the thread pool is saturated.

Run failed tasks separately via execute and submit: the former follows the uncaught exception path in the worker thread, while the latter stores the failure in a Future and wraps it into get() as ExecutionException. Discarding the Future after submission effectively abandons this failure-reporting path.

Final staged shutdown: shutdown, time-limited wait, necessary shutdownNow, then wait again. If the current thread is interrupted, first request cancellation, then restore the interrupt status. The thread count, queue capacity, rejection policy, downstream capacity, and shutdown deadline together form a complete executor contract.

Check the official contract for Java 17

• ThreadPoolExecutor: Task reception order, queue, thread factory, rejection, and hooks.
• ExecutorService: Submit, close, terminate, and memory consistency effects.
• Future: Guarantees visibility of results, exceptions, cancellation, and get.
• ScheduledThreadPoolExecutor: Rules for executing delayed and periodic tasks.

Continue Learning: I/O and Files

A thread pool manages worker count, queued tasks, rejection, and shutdown deadlines together. The next chapter covers files and directories; I/O is blocking and thus most needs the principles of capacity, timeouts, interruption, and shutdown here.

Continue learning about I/O streams and persistence boundaries