Scalable Task Execution with Thread Pools and Fork/Join

Alongside the safe coordination of shared resources, concurrent applications must address an equally critical concern — the management of the threads that perform the work itself. Regardless of how carefully locking is handled, every task must ultimately run on a thread — and the way those threads are created, reused, and controlled has a direct impact on the performance, stability, and scalability of the entire application.

To address the complexity of thread management, Java introduced the ExecutorService — a high-level abstraction that decouples the task submission from the task definition and how it is executed. Rather than dealing with Thread directly, software engineers define the task and delegate its execution to an executor; depending on the concrete implementation, it handles the tasks' queuing, thread creation, reuse, and termination. From performance perspective, instead of creating a thread for each task and destroying it on completion, the execution defines a set of threads, which are reused across many tasks.

Beyond the performance, the framework brings a substantial support to reliability and usability. It introduces a well-defined lifecycle to executors, which enables the starting and shutting down of task execution gracefully, a standard model for returning result of task execution and propagating the exceptions through Future, and build-in support for advanced scenarios such as scheduled and recurring tasks. By abstracting away the low-level mechanics of thread management, the Executor framework allows engineers to concentrate on the logic of their tasks rather than the infrastructure required to run them — embodying the same modern design principles that guide the rest of the concurrency utilities.

Deep Control Over Concurrent Execution with ExecutorService

At the top of the hierarchy, the Executor interface is defined, but because the more advanced and commonly used methods are defined by the ExecutorService, this article will treat the ExecutorService as the framework's high-level entry point. Overall, theExecutorService exposes a set of methods that capture its core capabilities: submitting a single task or invoking a list of tasks, waiting for task termination, shutting down the executor along with its running and pending tasks, and utility methods to verify whether tasks have terminated after a shutdown is triggered and whether the executor itself has been shut down.

ThreadPoolExecutor

At the heart of the executor service implementation sit the ThreadPoolExecutor, the most flexible and widely used. It's flexibility lies in the configuration the executor exposes: the number of task that run concurrently, reuse of threads from the pool, the core pre-initialised thread pool size, and several other parameters that enable software engineers to fine-tune behaviour without losing focus on business logic when building highly concurrent applications.

To understand where the flexibility comes from, let's take a closer look at the parameters that drive it.

• corePoolSize: defines the number of threads to keep alive in the pool, remaining even after long periods of inactivity
• maximumPoolSize: defines the maximum number of threads allowed by the thread pool, a less value than core pool size is not allowed
• keepAliveTime and unit: define how long idle threads beyond the core pool size are kept alive before being terminated
• workQueue: a java.util.concurrent.BlockingQueue<Runnable> that holds the tasks when no thread is immediately available to pick
• threadFactory: the java.util.concurrent.ThreadFactory used to create new threads when the pool requires additional workers for the task execution
• handler: handle the tasks that cannot be executed because threads in pool are busy and the queue has reached its capacity

Now that we covered the parameters that make the executor service flexible, it's time to dive into the internal flow of the ThreadPoolExecutor and examine the nuances that shapes its behaviour. We'll now turn to task submission and examine the internal mechanisms enables the thread pool's behaviour.

Worker plays a central role in the task execution within the ThreadPoolExecutor, being responsible for running the task and manage the thread, and retrieving pending tasks from the work queue. The Worker is build on top of AQS framework and leverages the locking mechanism to coordinate the execution task execution and manage its internal state.

When new task is submitted for execution, the thread pool check whether the current number of workers is below the configured core pool size. If so, a new worker is created immediately to execute the task, enabling lazy initialisation of the core thread pool. If core pool size has already been reached, the executor attempts to place the task into work queue. When the queue cannot accept the task, the executor checks whether the configured maximum pool size allows an additional worker to be created. If no further worker can be added, the task is handled according the configured rejection policy. For this reason, the difference between maximum pool size and core pool size should not be viewed as a flexible buffer that immediately is available for task execution.

package org.course.concurrency;

import java.util.Random;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class MultithreadingApplication {

  private static final Random RANDOM = new Random();

  public static void main(String[] args) {
    try (var executor =
        new ThreadPoolExecutor(5, 15, 10L, TimeUnit.SECONDS, new LinkedBlockingQueue<>())) {
      for (int i = 0; i < 100; i++) {
        executor.submit(
            () -> {
              System.out.println("Running thread: " + Thread.currentThread().getName());
              try {
                TimeUnit.SECONDS.sleep(RANDOM.nextInt(5));
              } catch (InterruptedException e) {
                throw new RuntimeException(e);
              }
            });
      }
    }
  }
}

After the worker thread is started, it repeatedly received tasks from the work queue and execute them, where the implementation is shared by ThreadPoolExecutor#runWorker. Because the executor relies on a BlockingQueue, the worker thread blocks when no task is available. Internally, the queue's waiting mechanism parks the thread until new task arrives in the work queue. There is a slight difference between core and non-core workers. Non-core worker poll the tasks with a timeout units, if no task becomes available within the keep-alive time configured during the executor service initialisation, getTask() returns null, cause the worker loop to complete and the worker thread to terminate. This design shows the strength of ThreadPoolExecutor: it does not manage every low-level operation directly, but composes several concurrency mechanisms into a flexible execution model. Thread creation is delegated to a ThreadFactory, task waiting is handled by a BlockingQueue, and worker execution is coordinated through the internal Worker abstraction.

In addition to submitting individual tasks, the executor service provides operation for executing a collection of tasks: invokeAll() and invokeAny()

package org.course.concurrency;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class MultithreadingApplication {

  private static final Random RANDOM = new Random();

  public static void main(String[] args) {
    try (ExecutorService executor = Executors.newFixedThreadPool(5)) {
      var tasks = new ArrayList<Callable<String>>();
      for (int i = 0; i < 100; i++) {
        tasks.add(
            () -> {
              var thread = "Running thread: " + Thread.currentThread().getName();
              System.out.println(thread);
              try {
                TimeUnit.SECONDS.sleep(RANDOM.nextInt(5));
              } catch (InterruptedException e) {
                throw new RuntimeException(e);
              }
              return thread;
            });
      }
      System.out.println("Submitting " + tasks.size() + " tasks at " + LocalDateTime.now());
      executor.invokeAll(tasks);
      System.out.println("Completing application execution at: " + LocalDateTime.now());
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
  }
}

package org.course.concurrency;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class MultithreadingApplication {

  private static final Random RANDOM = new Random();

  public static void main(String[] args) {
    try (ExecutorService executor = Executors.newFixedThreadPool(5)) {
      var tasks = new ArrayList<Callable<String>>();
      for (int i = 0; i < 100; i++) {
        tasks.add(
            () -> {
              var thread = "Running thread: " + Thread.currentThread().getName();
              try {
                TimeUnit.SECONDS.sleep(RANDOM.nextInt(5));
              } catch (InterruptedException e) {
                throw new RuntimeException(e);
              }
              System.out.println(thread);
              return thread;
            });
      }
      System.out.println("Submitting " + tasks.size() + " tasks at " + LocalDateTime.now());
      var result = executor.invokeAny(tasks);
      System.out.println(
          "Completing application execution with " + result + " at: " + LocalDateTime.now());
    } catch (InterruptedException | ExecutionException e) {
      throw new RuntimeException(e);
    }
  }
}

Managing the lifecycle of thread pool in a concurrent application is as important as executing the task, and the executor service exposes dedicated methods.

shutdown() gracefully shut down the thread pool where transition the state to SHUTDOWN and stopping new task submission to the executor service. At its core, the executor interrupts idle workers and attempts to signal termination when worker count reaches ZERO. If workers count are still active, it indicates that tasks are still in progress—either currently running or waiting in the queue—and the shutdown operation completes, leaving the final termination to be triggered by the remaining workers as they finish. When a worker exists the task execution loop, it attempts to terminate the thread pool. This mechanism ensures that the last worker to complete the execution is responsible for signalling the final termination of the thread pool executor service and transition to TERMINATED state.

 shutdownNow() transition the thread pool to STOP state and initiate an immediate shutdown of executor service without waiting for running tasks of queued tasks to gracefully complete the execution. The worker shutdown performs thread interruption, which means immediate thread termination is not guaranteed. Interruption act as a signal, and whether the task stops depends on how its implementation responds to that signal. Any tasks waiting in work queue are retuned by the method and have to notice that interrupted tasks are not included the in the returned list.

awaitTermination(timeout, unit) is a blocking operation that waits for a specific timeout to receive the terminate signal produced by the executor workers.

isTerminated and isShutdown check the thread pool state and translate to a boolean indicating if the executor has completed the pending tasks or shutting down has been initiated. These methods are especially useful after calling shutdown, when the application has to determine whether operation completed.

For analytics the ThreadPoolExecutor provides several methods to get approximate number of threads that currently execute the tasks, approximate number of tasks that have completed execution within the executor, approximate number of tasks that has been scheduled within the executor, largest number of threads that has been simultaneously executed by the executor.

ScheduledExecutorService / ScheduledThreadPoolExecutor

The ScheduledExecutorService is a specialised variation of ExecutorService that introduces time-based task execution, supporting delayed execution along with recurring execution based on fixed rate or fixed delay. Because the ScheduledThreadPoolExecutor is the primary implementation of ScheduledExecutorService, it provides the right place to examine how scheduled task are represented, queued, and executed. At its core, the implementation is build on top of ThreadPoolExecutor, reusing its worker management and task execution model. Workers are initialised lazily through method ensurePrestart() when a task is scheduled through methods exposed by ThreadPoolExecutor.

• ensurePrestart(): ensure that at least one worker is running, create new core worker does not reach the configured core pool size; when the core pool size is not configured (is zero) then launch a non-core worker
• prestartCoreThread(): starts a core worker is maximum core size is not reached and return true is new idle core worker has been launched or false if all core threads already has been started
• prestartAllCoreThreads(): starts core worker to fulfil the core pool size and return the number or worker launched in idle mode, a small number than core pool size or even zero could be returned when core pool already runs core workers

Beyond its foundation on ThreadPoolExecutor, the schedule execution model of ScheduledThreadPoolExecutor is largely driven by two nested implementations: ScheduledFutureTask and DelayedWorkQueue, which define the behaviour of scheduled executor service. Task execution is not guaranteed to occur at exact time the delay expires, as it depends on the worker availability. Instead, the implementation guarantees that the task becomes eligible for execution one the configured delay has been elapsed.

Internally, the DelayedWorkQueue maintains the tasks ordered by their scheduled execution time, using the ordering defined by ScheduledFutureTask#compareTo. The ordering is presented through sift operations as tasks are inserted into or removed from the queue. Although tasks are managed within an array, every queue operation is protected by a ReentrantLock, ensuring that only one thread can modify the queue at a time and guaranteeing thread safety. The following section examines the key methods used by the core thread pool implementation.

take() access the first task in the queue before returning it to worker for execution, it passes through several paths designed to handle different scheduling scenarios. When the queue head is null, meaning no tasks are scheduled, the current worker waits for available signal, through available condition indicating that a new task has been scheduled. When the queue contains tasks but the scheduled execution time of head task has not yet been reached, the worker waits for remaining delay. Once the task exists and scheduled delay expires, the head task is removed from the queue, all the tasks are sifted and the head task is returned to worker for execution.

An important detail is that receiving the available signal or waking up after the delay does not mean that the worker receives the scheduled task for execution. Instead, the worker repeats the evaluation by checking whether a task is present at the head of the queue and whether it's scheduled execution time has elapsed. This repeated validation ensures that newly scheduled tasks with an earlier execution time are handled correctly before any previously waiting task and no other worker has already taken it for execution.

poll(timeout, unit) behaves similarly to take(), but returns null if poll timeout expired which cause the non-worker to complete the thread and waiting delay for signal is configured based on left time between scheduled delay and polling delay.

offer(runnableTask) inserts the current runnable task into the queue according to its scheduled execution time once the offering thread acquires the lock. During insertion, the task is sifted until it reaches appropriate position, ensuring that tasks scheduled to run earlier remain ahead of it, while tasks with later execution times are shifted accordingly to assure the queue ordering. When the newly inserted task becomes the queue header, the queue signals the available condition.

The next component driving the behaviour of ScheduledThreadPoolExecutor is the ScheduledFutureTask. The ScheduledFutureTask is responsible encapsulate the task to execute, the initial delay of scheduled task and period where positive value indicates a fixed-rate execution, negative indicates fixed-delay execution and zero indicates non-repetitive task. Also provides support to compare the tasks based on their delay through Comparable#compareTo . The period field cannot be fully understood in isolation, as its value is determined by the scheduling method used to create the task and defines the scheduled task behaviour. Having briefly explored the internal flows and implementations, the focus returns to the primary subject of this section—describe the capabilities of ScheduledExecutorService through the interface methods.

schedule(command, delay, unit) the method schedule a non-repetitive task with an initial delay. In practice the method create a ScheduledFutureTask with an initial delay so the DelayedWorkQueue waits this time before make the task ready for execution. For non-repetitive tasks, the ScheduledFutureTask just execute the scheduled task.

package org.course.concurrency;

import java.time.LocalDateTime;
import java.util.Random;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class MultithreadingApplication {

  private static final Random RANDOM = new Random();

  public static void main(String[] args) {
    try (ScheduledThreadPoolExecutor scheduledExecutorService =
        new ScheduledThreadPoolExecutor(3)) {
      System.out.println("Start scheduled tasks at: " + LocalDateTime.now());
      for (int i = 0; i < 10; i++) {
        int delay = RANDOM.nextInt(3, 10);
        int taskName = i;
        System.out.printf(
            "Scheduling the task %s with delay %s at %s%n", taskName, delay, LocalDateTime.now());
        scheduledExecutorService.schedule(
            () -> {
              System.out.printf("Running scheduled task %s at %s%n", taskName, LocalDateTime.now());
              try {
                TimeUnit.MILLISECONDS.sleep(RANDOM.nextInt(500));
              } catch (InterruptedException e) {
                throw new RuntimeException(e);
              }
            },
            delay,
            TimeUnit.SECONDS);
      }
    }
  }
}

scheduleAtFixedRate(command, initialDelay, period, unit) the method schedule a periodic task with an initial delay. Under the fixed-rate execution, the ScheduledFutureTask calculates the next execution time from configured period and previous configured execution time, rather than time when the task actually started or completed. For repetitive task, the ScheduledFutureTask executes the task, updates its next scheduled time, re-enqueues itself into the work queue, ensuring that at least one worker is available in the thread pool.

package org.course.concurrency;

import java.time.LocalDateTime;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class MultithreadingApplication {

  public static void main(String[] args) throws InterruptedException {
    try (ScheduledThreadPoolExecutor scheduledExecutorService =
        new ScheduledThreadPoolExecutor(10)) {
      System.out.println("Start scheduled tasks at: " + LocalDateTime.now());
      for (int i = 0; i < 5; i++) {
        int taskName = i;
        scheduledExecutorService.scheduleAtFixedRate(
            () -> {
              System.out.printf("Running scheduled task %s at %s%n", taskName, LocalDateTime.now());
              try {
                TimeUnit.SECONDS.sleep(3);
              } catch (InterruptedException e) {
                throw new RuntimeException(e);
              }
            },
            1,
            5,
            TimeUnit.SECONDS);
      }
      // waiting to execute a specific number of tasks
      while (scheduledExecutorService.getCompletedTaskCount() < 20) {
        TimeUnit.MILLISECONDS.sleep(500);
      }
      scheduledExecutorService.shutdown();
    }
  }
}

scheduleWithFixedDelay(command, initialDelay, period, unit) the method schedule a periodic task with an initial delay. Under fixed-delay execution, the ScheduledFutureTask calculates the next execution time from the configured period and time when the previous schedule completed execution. Similar to fixed-rate, updates its next scheduled time, re-enqueues itself and ensure that at least one worker is available in the thread pool.

package org.course.concurrency;

import java.time.LocalDateTime;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class MultithreadingApplication {

  public static void main(String[] args) throws InterruptedException {
    try (ScheduledThreadPoolExecutor scheduledExecutorService =
        new ScheduledThreadPoolExecutor(10)) {
      System.out.println("Start scheduled tasks at: " + LocalDateTime.now());
      for (int i = 0; i < 5; i++) {
        int taskName = i;
        scheduledExecutorService.scheduleWithFixedDelay(
            () -> {
              System.out.printf("Running scheduled task %s at %s%n", taskName, LocalDateTime.now());
              try {
                TimeUnit.SECONDS.sleep(3);
              } catch (InterruptedException e) {
                throw new RuntimeException(e);
              }
            },
            1,
            5,
            TimeUnit.SECONDS);
      }
      // waiting to execute a specific number of tasks
      while (scheduledExecutorService.getCompletedTaskCount() < 20) {
        TimeUnit.MILLISECONDS.sleep(500);
      }
      scheduledExecutorService.shutdown();
    }
  }
}

ForkJoinPool: Parallel Recursive Task Processing 

ForkJoinPool is a specialised executor service implementation optimised for CPU-bound tasks that follow a fork/join execution model, allowing computations to be decomposed into smaller, independent subtasks, executed concurrently, and later joined into a final result. This execution model is particularly effective for large computations that can be divided into many small independent subtasks, but does not provide benefit for workloads dominated by blocking operations such as network or database access.

The parallelism parameter of ForkJoinPool defines the number of worker threads available to execute the tasks and subtasks. When a worker becomes idle, steal pending subtasks from busy workers to maximise processor utilisation.

ForkJoinWorkerThread is the worker of ForkJoinPool, an extension of Thread responsible for executing tasks submitted to ForkJoinPool. Each worker owns a local queue of ForkJoinTask and once started, delegates to the owning ForkJoinPool to scan its local queue or steal tasks from other worker when idle (its local queue does not have a task at this point in time). Each ForkJoinPool instance manages the work queues associated with its worker thread. This should not be confused with ForkJoinPool.commonPool(), which is a separate shared pool used only when tasks are submitted to the common pool.

The ForkJoinPool instance is managing queues of all workers but this should not be confused with the ForkJoinPool.commonPool(), which is a separate shared pool used only when tasks are submitted to the common pool.

The ForkJoinPool worker ForkJoinWorkerThread is an extension of Thread and is responsible to manage the own queue of tasks and when run delegates to ForkJoinPool to scan own and work queue and await for task from managed work queue of ForkJoinPool instance, important to not confuse with common pool.

A complex task can be decomposed into smaller, independent subtasks that are executed concurrently within a ForkJoinPool only the subtask execution is encapsulated as a ForkJoinTask. The ForkJoinTask is abstract and most appropriate representation to use in practice are RecursiveTask and RecursiveAction. Following will demonstrate how to use them in practice.

package org.course.concurrency;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;

public class MultithreadingApplication {

  private static final Random RANDOM = new Random();

  public static void main(String[] args) throws InterruptedException {
    try (ForkJoinPool forkJoinPool = new ForkJoinPool(15)) {
      List<OrderLine> novaMartOrderLines = new ArrayList<>(100_000);
      for (int i = 0; i < 10_000_000; i++) {
        novaMartOrderLines.add(
            new OrderLine(
                RANDOM.nextInt(50, 150),
                new BigDecimal(RANDOM.nextInt(5, 50)),
                new BigDecimal(RANDOM.nextInt(10))));
      }
      List<OrderLine> urbanCoOrderLines = new ArrayList<>(150_000);
      for (int i = 0; i < 10_500_000; i++) {
        urbanCoOrderLines.add(
            new OrderLine(
                RANDOM.nextInt(100, 550),
                new BigDecimal(RANDOM.nextInt(10, 30)),
                new BigDecimal(RANDOM.nextInt(5))));
      }
      System.out.println("Starting revenue calculation: " + LocalDateTime.now());
      ForkJoinTask<BigDecimal> novaMartRevenueCalculationTask =
          forkJoinPool.submit(new RevenueCalculationTask(novaMartOrderLines));
      ForkJoinTask<BigDecimal> urbanCoRevenueCalculationTask =
          forkJoinPool.submit(new RevenueCalculationTask(urbanCoOrderLines));
      System.out.println("Completed revenue calculation: " + LocalDateTime.now());
      System.out.println("Nova Mart revenue: " + novaMartRevenueCalculationTask.get());
      System.out.println("Urban Co revenue: " + urbanCoRevenueCalculationTask.get());
    } catch (ExecutionException e) {
      throw new RuntimeException(e);
    }
  }

  private static class RevenueCalculationTask extends RecursiveTask<BigDecimal> {

    private static final int THRESHOLD = 5_000;

    private final List<OrderLine> orderLines;

    public RevenueCalculationTask(List<OrderLine> orderLines) {
      this.orderLines = Objects.requireNonNull(orderLines);
    }

    @Override
    protected BigDecimal compute() {
      if (orderLines.size() < THRESHOLD) {
        return calculateDirectly();
      }
      int middle = orderLines.size() / 2;
      RevenueCalculationTask leftCalculationTask =
          new RevenueCalculationTask(orderLines.subList(0, middle));
      RevenueCalculationTask rightCalculationTask =
          new RevenueCalculationTask(orderLines.subList(middle, orderLines.size()));
      // fork left task, enqueue it and let fork join worker to steal the execution
      leftCalculationTask.fork();
      // right task will be executed by current worker,
      // tiny decoupling does not necessary mean better performance
      // compute will execute the calculation on direct worker
      BigDecimal rightResult = rightCalculationTask.compute();
      BigDecimal leftResult = leftCalculationTask.join();
      return rightResult.add(leftResult);
    }

    private BigDecimal calculateDirectly() {
      System.out.println(
          "Running revenue calculation by thread: " + Thread.currentThread().getName());
      return orderLines.stream().map(OrderLine::revenue).reduce(BigDecimal.ZERO, BigDecimal::add);
    }
  }

  private record OrderLine(int quantity, BigDecimal unitPrice, BigDecimal discount) {
    public BigDecimal revenue() {
      return unitPrice.multiply(BigDecimal.valueOf(quantity)).subtract(discount);
    }
  }
}

Now that we have the general characteristics of the fork/join framework and shown a practical application, we can proceed to consider the methods that matters to control the scheduling, execution and waiting for task completion.

The core operations provided by ForkJoinPool define how computations are submitted for execution, whether they execute synchronously or asynchronously, and how their results are obtained.

• ForkJoinPool#submit(), asynchronously submit a task of Runnable, Callable or ForkJoinTask type for execution and immediately return a ForkJoinTask representing the submitted computation
• ForkJoinPool#invoke(), allows to submit a ForkJoinTask for execution and return the computation result when task completes execution. This operation blocks the calling thread until the submitted task completes and its result becomes available

The methods exposed by ForkJoinTask define how subtasks are scheduled for concurrent execution, synchronized upon completion, and coordinated to complete the overall computation.

• ForkJoinTask#exec() represents the task to execute
• ForkJoinTask#fork() places the current task into the worker’s local queue, allowing the current worker to continue executing its computation while the forked task can be executed later or stolen by another idle worker.
• ForkJoinTask#join() returns the task result once the task completes. When invoked from a ForkJoinWorkerThread, the waiting worker may execute other available ForkJoinTasks while awaiting completion of the joined task
• ForkJoinTask#get() implements the Future#get() contract and differs from join() in how it reports completion. If the waiting thread is interrupted ForkJoinTask#get() throws InterruptedException, if the task completes exceptionally, the exception is wrapped in ExecutionException.
• ForkJoinTask#invoke() starts the task execution flow directly in the current thread and return the result once the computation completes. It is useful when the current flow needs the task result immediately rather than scheduling the task asynchronously trough fork()

CompletionService: Another Way to Submit and Handle Completed Tasks

The java.util.concurrent.CompletionService provides an alternative execution model for submitting tasks and retrieving their results in completion order. At its core, ExecutorCompletionService is the main implementation of CompletionService where coordinate the task execution with an Executor and manage task completion within a BlockingQueue<Future<V>>.

In general, the CompletionService aims to decouple task submission and the consumption of task result. The model is useful when task submission and result processing operate independently. One component may submit tasks for execution, while another consumes results from the queue. Unlike ExecutorService#invokeAll(), which waits for the entire collection to complete, or manually iterating over submitted Future instances and potentially waiting on unfinished task, CompletionService makes each completed task available to the requester as soon as its result is ready. This allow results to be processed in completion order rather than submission order.

On submission, ExecutorCompletionService wraps the task in a QueueingFuture, its internal extension of FutureTask where overrides the FutureTask#done(), and delegates the execution to underlying Executor. The task execution itself is entirely driven by the executor implementation. Once the task completes, QueueingFuture automatically place the completed Future into the underlying completion queue, making available for retrieval only when task completed.

For completed task fetching, the CompletionService#pool(timeout, unit), CompletionService#pool() and CompletionService#take() is completely handled by the BlockingQueue implementation.

package org.course.concurrency;

import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

public class MultithreadingApplication {

  public static void main(String[] args) {
    // use enough number of workers to ensure all task runs concurrently
    try (ExecutorService executorService = Executors.newFixedThreadPool(20)) {
      CompletionService<String> completionService =
          new ExecutorCompletionService<>(executorService);
      executorService.invokeAll(
          List.of(new Producer(completionService), new Consumer(completionService)));
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
  }

  private static class Producer implements Callable<Void> {

    private final CompletionService<String> completionService;

    public Producer(CompletionService<String> completionService) {
      this.completionService = completionService;
    }

    @Override
    public Void call() {
      for (int i = 15; i > 0; i--) {
        int sleepTimeout = i;
        completionService.submit(
            () -> {
              System.out.printf("Starting task expected to complete in %s seconds\n", sleepTimeout);
              TimeUnit.SECONDS.sleep(sleepTimeout);
              return "Task %s completed at %s".formatted(sleepTimeout, LocalDateTime.now());
            });
      }
      return null;
    }
  }

  private static class Consumer implements Callable<Void> {

    private final CompletionService<String> completionService;

    private Consumer(CompletionService<String> completionService) {
      this.completionService = completionService;
    }

    @Override
    public Void call() {
      try {
        Future<String> future;
        while ((future = completionService.poll(5, TimeUnit.SECONDS)) != null) {
          System.out.println(future.get());
        }
      } catch (InterruptedException | ExecutionException e) {
        throw new RuntimeException(e);
      }
      return null;
    }
  }
}

Managing Asynchronous Results: From Future to CompletableFuture

The previous sections have already introduced Future, RunnableFuture, and related implementation while we examine task submission and completion. With that foundation established, it is time to take a closer look at the abstractions and implementations Java provides for representing, retrieving, and composing asynchronous results. The following section explorers the traditional Future model through RunnableFuture and FutureTask, followed by the more flexible completion model defined by CompletableStage and implemented by CompletableFuture.

Future represents the result of an asynchronous computation. It provides the methods to check if computation completed (computation completed or cancelation completed), wait for task completion and retrieve the result of computation. At first glance, Future appears to encapsulate the computation result and expose it through get(). In reality, the result is available after the computation completes, causing the calling thread to block until the result becomes available. Technical aspects behind this behaviour we will analyse later together with concrete implementations FutureTask and CompletableFuture.

RunnableFuture extends both Runnable and Future, combining task execution with asynchronous result retrieval into a single abstraction. This model, enables RunnableTask to execute the task and park the waiter until the result is available.

FutureTask being the standard implementation of RunnableFuture, encapsulates task execution and asynchronous result management. Understanding its implementation provides valuable insight into how Java coordinates task execution, completion, and result retrieval. The following section examines the methods that coordinate lifecycle, state transition, completion and result retrieval. FutureTask maintains an internal simple linked list nodes to record threads waiting the computation result and the thread that is running the task.

run() executes the encapsulated computation, set the outcome result and unpack waiting threads. Once the computation completes, the finishCompletion unpark the waiting threads present in FutureTask#waiters.

runAndReset is an alternative method that let the subtask like ScheduledFutureTask to execute the computation without setting the outcome result and without reseting the future task state.

get() retrieves the result of the asynchronous computation, blocking the calling thread if the computation has not yet completed. When the result is not available, the thread getting the result is parked and added to waiters stack nodes.

cancel() is an operation that tries to cancel the task execution. When task is in new state, on cancel the task is marked with CANCELLED when does not require to interrupt the executing thread or INTERRUPTING when should interrupt the executing thread. When task is currently running then interrupt the runner thread and unpark waiter threads.

isDone() indicates if the task completed. Important to note that done may indicate completing, normal completion, exception or cancellation.

isCancelled() indicates if the task is canceled before it's normal completion, the state could indicate canceled, interrupting or interrupted

resultNow() immediately return computation result. The method is intended to for cases where the result is guaranteed to be available. If task is not completed then the thread becomes busy-waiting and enter retry storm problem until state transition to NORMAL or EXCEPTIONAL. When task does not completes successfully IllegalStateException is thrown.

exceptionNow() similar to resultNow(), the FutureTask immediately return exception that caused task to failure. Important to note that the method also is intended for cases when guarantee the task completed with an exception because may enter retry storm problem until state transition to state or shows exception if task has been canceled or completed with success.

CompletionStage: Orchestrating Asynchronous Stages

With CompletionStage, the asynchronous programming model moves beyond the traditional Future approach by allowing computations to be composed instead of simply awaited. Instead of representing individual asynchronous computation result, each stage defines a continuation that is executed when the pending stage completes, enabling asynchronous computation to progress through a chain of dependent operations.

CompletionStage and CompletableFuture should not be confused with executor implementations. Although they provide asynchronous execution method, they neither create not manage worker threads, instead provide an execution model allowing to run either synchronously in the thread completing the previous stage or asynchronously relying on ForkJoinPool#commonPool() or a user supplier Executor. Following will explore how to create CompletionStage, apply computation to stage result and build a different result, consume the stage result and run a task when previous stage completes the execution without a result.

thenApply and thenApplyAsync transform the result of a completed stage into new value, potentially of a different type, and encapsulate it within a new CompletionStage

package org.course.concurrency;

import java.time.LocalDateTime;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Function;
import java.util.function.Supplier;

public class MultithreadingApplication {

  public static void main(String[] args) {
    try (ExecutorService executorService = Executors.newFixedThreadPool(5)) {
      System.out.println(
          "Executing CompletableFuture pipeline stage by thread: "
              + Thread.currentThread().getName());
      CompletableFuture<LocalDateTime> completableStage =
          CompletableFuture.supplyAsync(new ValueSupplier(), executorService)
              .thenApplyAsync(new ValueMapper(), executorService);
      System.out.printf(
          "Consuming completion stage result %s by thread %s%n",
          completableStage.join(), Thread.currentThread().getName());
    }
  }

  private static class ValueSupplier implements Supplier<String> {

    @Override
    public String get() {
      System.out.println("Running supplier by thread: " + Thread.currentThread().getName());
      return "supplier value: " + Thread.currentThread().getName();
    }
  }

  private static class ValueMapper implements Function<String, LocalDateTime> {

    @Override
    public LocalDateTime apply(String supplierValue) {
      System.out.printf(
          "Transforming value '%s' by thread: %s%n",
          supplierValue, Thread.currentThread().getName());
      return LocalDateTime.now();
    }
  }
}

thenAccept and thenAcceptAsync consumes the result of completed stage without producing a new value. The operation returns a CompletionStage<Void> representing the successful completion of all preceding stages including the accept consumer stage.

package org.course.concurrency;

import java.time.LocalDateTime;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Supplier;

public class MultithreadingApplication {

  public static void main(String[] args) {
    try (ExecutorService executorService = Executors.newFixedThreadPool(5)) {
      System.out.printf(
          "Executing CompletableFuture pipeline stage at %s by thread: %s%n",
          LocalDateTime.now(), Thread.currentThread().getName());
      CompletableFuture<Void> completableStage =
          CompletableFuture.supplyAsync(new ValueSupplier(), executorService)
              .thenAcceptAsync(new ValueConsumer(), executorService);
      completableStage.join();
      System.out.printf(
          "Completion stage completed at %s by thread %s%n",
          LocalDateTime.now(), Thread.currentThread().getName());
    }
  }

  private static class ValueSupplier implements Supplier<String> {

    @Override
    public String get() {
      try {
        System.out.println("Running supplier by thread: " + Thread.currentThread().getName());
        TimeUnit.SECONDS.sleep(1);
        return "supplier value: " + Thread.currentThread().getName();
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
    }
  }

  private static class ValueConsumer implements Consumer<String> {

    @Override
    public void accept(String supplierValue) {
      try {
        TimeUnit.SECONDS.sleep(2);
        System.out.printf(
            "Consuming value '%s' by thread: %s%n",
            supplierValue, Thread.currentThread().getName());
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
    }
  }
}

thenRun and thenRunAsync trigger a new action after the preceding stage completes, without consuming its result or producing a new value. The operation is invoked on empty and returns empty CompletionStage<Void> whose completion indicates that the previous stage and the subsequent Runnable execution have completed successfully.

package org.course.concurrency;

import java.time.LocalDateTime;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Supplier;

public class MultithreadingApplication {

  public static void main(String[] args) {
    try (ExecutorService executorService = Executors.newFixedThreadPool(5)) {
      System.out.printf(
          "Executing CompletableFuture pipeline stage at %s by thread: %s%n",
          LocalDateTime.now(), Thread.currentThread().getName());
      CompletableFuture<Void> completableStage =
          CompletableFuture.supplyAsync(new ValueSupplier(), executorService)
              .thenAcceptAsync(new ValueConsumer(), executorService)
              .thenRunAsync(
                  () ->
                      System.out.printf(
                          "Completing pipeline execution at %s by thread %s%n",
                          LocalDateTime.now(), Thread.currentThread().getName()),
                  executorService);
      completableStage.join();
      System.out.printf(
          "Completion stage completed at %s by thread %s%n",
          LocalDateTime.now(), Thread.currentThread().getName());
    }
  }

  private static class ValueSupplier implements Supplier<String> {

    @Override
    public String get() {
      try {
        System.out.println("Running supplier by thread: " + Thread.currentThread().getName());
        TimeUnit.SECONDS.sleep(1);
        return "supplier value: " + Thread.currentThread().getName();
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
    }
  }

  private static class ValueConsumer implements Consumer<String> {

    @Override
    public void accept(String supplierValue) {
      try {
        TimeUnit.SECONDS.sleep(2);
        System.out.printf(
            "Consuming value '%s' by thread: %s%n",
            supplierValue, Thread.currentThread().getName());
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
    }
  }
}

Summary

The Java concurrency framework provides a rich set of abstractions for executing and coordinating asynchronous computations. ExecutorService separates task submission from thread management, allowing applications to execute workloads efficiently without manually creating and controlling threads. Different executor implementations address different execution models, from fixed and cached thread pools to scheduled execution, work-stealing, and completion-driven task processing.

Building on top of executors, the Future abstraction introduces asynchronous result management by representing computations whose outcome becomes available at a later time. Its implementation through FutureTask demonstrates how task execution, state transitions, waiting threads, and completion are coordinated internally. While Future successfully decouples task execution from result retrieval, obtaining the result still requires synchronization with the asynchronous computation.

CompletionStage and CompletableFuture extend this model by allowing asynchronous computations to be expressed as a chain of dependent stages. Instead of explicitly waiting for intermediate results, subsequent stages are triggered automatically as previous stages complete, enabling more expressive and non-blocking asynchronous workflows. Although these abstractions provide methods for asynchronous execution, they delegate the actual execution to an underlying Executor, separating computation orchestration from thread management.

Understanding these implementations reveals that Java’s concurrency framework is built on a set of complementary abstractions rather than isolated APIs. Executors focus on task execution, Future manages asynchronous results, and CompletionStage coordinates dependent computations. Together, they form the foundation for building scalable, responsive, and maintainable concurrent applications while hiding much of the complexity involved in scheduling, synchronization, and asynchronous execution.