Java Thread Pools and ThreadPoolExecutor

Why do we need a thread pool in Java? The answer is when we develop a simple, concurrent application in Java, we create some Runnable objects and then create the corresponding Thread objects to execute them. Creating a thread in Java is an expensive operation. And if you start creating a new thread instance everytime to accomplish a task, application performance will degrade.

1. How does a Thread Pool Works?

A thread pool is a collection of pre-initialized threads. Generally, the collection size is fixed, but it is not mandatory. It facilitates the execution of N number of tasks using the same threads. If there are more tasks than threads, then tasks need to wait in a queue like structure (FIFO – First in first out).

When any thread completes its execution, it can pickup a new task from the queue and execute it. When all tasks are completed, the threads remain active and wait for more tasks in the thread pool.

A watcher keeps watching the queue (usually BlockingQueue) for any new tasks. As soon as tasks come, threads start picking up tasks and executing them again.

2. ThreadPoolExecutor class

Since Java 5, the Java concurrency API provides a mechanism Executor framework. The main pieces are Executor interface, its sub-interface ExecutorService and the ThreadPoolExecutor class that implements both interfaces.

ThreadPoolExecutor separates the task creation and its execution. With ThreadPoolExecutor, we only have to implement the Runnable objects and send them to the executor. It is responsible for executing, instantiating, and running the tasks with necessary threads.

It goes beyond that and improves performance using a pool of threads. When you send a task to the executor, it tries to use a pooled thread to execute this task, to avoid the continuous spawning of threads.

3. Creating ThreadPoolExecutor

We can create the following 5 types of thread pool executors with pre-built methods in java.util.concurrent.Executors interface.

3.1. Fixed Sized Thread Pool Executor

Creates a thread pool that reuses a fixed number of threads to execute any number of tasks. If additional tasks are submitted when all threads are active, they will wait in the queue until a thread is available. It is the best fit for most of real-life use-cases.

ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(10);

3.2. Cached Thread Pool Executor

Creates a thread pool that creates new threads as needed, but will reuse previously constructed threads when they are available. DO NOT use this thread pool if tasks are long-running. It can bring down the system if the number of threads exceeds what the system can handle.

ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newCachedThreadPool();

3.3. Scheduled Thread Pool Executor

Creates a thread pool that can schedule commands to run after a given delay or to execute periodically.

ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newScheduledThreadPool(10);

3.4. Single Thread Pool Executor

Creates a single thread to execute all tasks. Use it when you have only one task to execute.

ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newSingleThreadExecutor();

3.5. Work Stealing Thread Pool Executor

Creates a thread pool that maintains enough threads to support the given parallelism level. Here, parallelism level means the maximum number of threads that will be used to execute a given task at a single point in multi-processor machines.

ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newWorkStealingPool(4);

4. ThreadPoolExecutor Example

4.1. Creating a Task

Let’s create a task that will take 2 seconds to complete, every time.

class Task implements Runnable {
  private final String name;

  public Task(String name) {
    this.name = name;
  }

  @SneakyThrows
  @Override
  public void run() {
    Thread.sleep(2000l);
    System.out.println("Task [" + name + "] executed on : " + LocalDateTime.now().toString());
  }
}

4.2. Execute Tasks with Thread Pool Executor

The given program creates 5 tasks and submits them to the executor queue. The executor uses a single thread to execute all tasks.

import java.time.LocalDateTime;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ThreadPoolExample {
  public static void main(String[] args) {

    ExecutorService executor = Executors.newSingleThreadExecutor();

    for (int i = 1; i <= 5; i++) {
      Task task = new Task("Task " + i);
      executor.execute(task);
    }

    shutdownAndAwaitTermination(executor);
  }

  static void shutdownAndAwaitTermination(ExecutorService pool) {
    // Disable new tasks from being submitted
    pool.shutdown();
    try {
      // Wait a while for existing tasks to terminate
      if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
        // Cancel currently executing tasks forcefully
        pool.shutdownNow();
        // Wait a while for tasks to respond to being cancelled
        if (!pool.awaitTermination(60, TimeUnit.SECONDS))
          System.err.println("Pool did not terminate");
      }
    } catch (InterruptedException ex) {
      // (Re-)Cancel if current thread also interrupted
      pool.shutdownNow();
      // Preserve interrupt status
      Thread.currentThread().interrupt();
    }
  }
}

Program output:

Task [Task 1] executed on : 2022-08-07T17:05:18.470589200
Task [Task 2] executed on : 2022-08-07T17:05:20.482150
Task [Task 3] executed on : 2022-08-07T17:05:22.482660
Task [Task 4] executed on : 2022-08-07T17:05:24.498243500
Task [Task 5] executed on : 2022-08-07T17:05:26.499919700

5. Using ScheduledThreadPoolExecutor

Fixed thread pools or cached thread pools are good when executing one unique task only once. When you need to execute a task, repeatedly N times, either N fixed number of times or infinitely after a fixed delay, you should use ScheduledThreadPoolExecutor.

5.1. Schedule Methods

ScheduledThreadPoolExecutor provides 4 methods that offer different capabilities to execute the tasks repeatedly.

1. schedule(Runnable command, long delay, TimeUnit unit) – Creates and executes a task that becomes enabled after the given delay.
2. schedule(Callable callable, long delay, TimeUnit unit) – Creates and executes a ScheduledFuture that becomes enabled after the given delay.
3. scheduleAtFixedRate(Runnable command, long initialDelay, long delay, TimeUnit unit) – Creates and executes a periodic action that becomes enabled first after the given initial delay, and subsequently with the given delay period. If any task execution takes longer than its period, subsequent executions may start late, but will not concurrently execute.
4. scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) – Creates and executes a periodic action that becomes enabled first after the given initial delay, and subsequently with the given delay period. No matter how much time a long-running task takes, there will be a fixed delay time gap between two executions.

5.2. ScheduledThreadPoolExecutor Example

In the following example, the task will be executed periodically until the task is canceled. There will always be a delay of 10 seconds between the completion time of the first task and the start time of the second task.

ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);

Task task = new Task("App-Task");

ScheduledFuture<?> result = executor.scheduleWithFixedDelay(task1, 0, 10, TimeUnit.SECONDS);

6. Custom Thread Pool Implementation

Though Java has very robust thread pool functionality through Executor framework. And it would help if you were not creating your own custom thread pool without executor. I will vehemently discourage any such attempt. Yet if you would like to create it for your learning, the given below is such thread pool implementation in Java.

public class CustomThreadPool {
  //Thread pool size
  private final int poolSize;

  //Internally pool is an array
  private final WorkerThread[] workers;

  // FIFO ordering
  private final LinkedBlockingQueue<Runnable> queue;

  public CustomThreadPool(int poolSize) {
    this.poolSize = poolSize;
    queue = new LinkedBlockingQueue<Runnable>();
    workers = new WorkerThread[poolSize];

    for (int i = 0; i < poolSize; i++) {
      workers[i] = new WorkerThread();
      workers[i].start();
    }
  }

  public void execute(Runnable task) {
    synchronized (queue) {
      queue.add(task);
      queue.notify();
    }
  }

  private class WorkerThread extends Thread {
    public void run() {
      Runnable task;

      while (true) {
        synchronized (queue) {
          while (queue.isEmpty()) {
            try {
              queue.wait();
            } catch (InterruptedException e) {
              System.out.println("An error occurred while queue is waiting: " + e.getMessage());
            }
          }
          task = (Runnable) queue.poll();
        }

        try {
          task.run();
        } catch (RuntimeException e) {
          System.out.println("Thread pool is interrupted due to an issue: " + e.getMessage());
        }
      }
    }
  }

  public void shutdown() {
    System.out.println("Shutting down thread pool");
    for (int i = 0; i < poolSize; i++) {
      workers[i] = null;
    }
  }
}

Execute the same task which we executed with CustomThreadPool and 2 worker threads.

public class CustomThreadPoolExample {
  public static void main(String[] args) {
    CustomThreadPool customThreadPool = new CustomThreadPool(2);

    for (int i = 1; i <= 5; i++) {
      Task task = new Task("Task " + i);
      System.out.println("Created : " + task.getName());

      customThreadPool.execute(task);
    }
  }
}

Program output. Notice that it executes two tasks at a time.

Created : Task 1
Created : Task 2
Created : Task 3
Created : Task 4
Created : Task 5
Task [Task 2] executed on : 2022-08-07T17:19:15.846912100
Task [Task 1] executed on : 2022-08-07T17:19:15.846912100

Task [Task 4] executed on : 2022-08-07T17:19:17.874728800
Task [Task 3] executed on : 2022-08-07T17:19:17.874728800

Task [Task 5] executed on : 2022-08-07T17:19:19.878018200

Above is a very raw thread pool implementation with a scope of lots of improvements. But still, rather than perfecting the above code, focus on learning Java executor framework.

Also, note that incorrect pooling or queue handling can result in deadlocks or resource thrashing. You can certainly avoid these problems with the Executor framework which is well tested by the Java community.

7. Conclusion

1. The ThreadPoolExecutor class has four different constructors but due to their complexity, the Java concurrency API provides the Executors class to construct executors and other related objects. Although we can create ThreadPoolExecutor directly using one of its constructors, it’s recommended to use the Executors class.
2. The cached thread pool creates new threads if needed to execute the new tasks and reuses the existing ones if they have finished executing the task they were running, which are now available. However, the cached thread pool has the disadvantage of constant lying threads for new tasks, so if you send too many tasks to this executor, you can overload the system. This can be overcome using a fixed thread pool, which we will learn in next tutorial.
3. One critical aspect of the ThreadPoolExecutor class, and of the executors in general, is that you have to end it explicitly. If you don’t do this, the executor will continue its execution, and the program won’t end. If the executor doesn’t have tasks to execute, it continues waiting for new tasks and it doesn’t end its execution. A Java application won’t end until all its non-daemon threads finish their execution, so your application will never end if you don’t terminate the executor.
4. To indicate to the executor that you want to finish it, you can use the shutdown() method of the ThreadPoolExecutor class. When the executor finishes executing all pending tasks, it completes its execution. After you call the shutdown() method, if you try to send another task to the executor, it will be rejected and the executor will throw a RejectedExecutionException exception.
5. The ThreadPoolExecutor class provides a lot of methods to obtain information about its status. We used in the example the getPoolSize(), getActiveCount(), and getCompletedTaskCount() methods to obtain information about the size of the pool, the number of threads, and the number of completed tasks of the executor. You can also use the getLargestPoolSize() method that returns the maximum number of threads that has been in the pool at a time.
6. The ThreadPoolExecutor class also provides other methods related with the finalization of the executor. These methods are:
   • shutdownNow(): This method shut downs the executor immediately. It doesn’t execute the pending tasks. It returns a list with all these pending tasks. The tasks that are running when you call this method continue with their execution, but the method doesn’t wait for their finalization.
   • isTerminated(): This method returns true if you have called the shutdown() or shutdownNow() methods and the executor finishes the process of shutting it down.
   • isShutdown(): This method returns true if you have called the shutdown() method of the executor.
   • awaitTermination(long timeout,TimeUnitunit): This method blocks the calling thread until the tasks of the executor have ended or the timeout occurs. The TimeUnit class is an enumeration with the following constants: DAYS, HOURS, MICROSECONDS etc.

Happy Learning !!

Source Code on Github