Concurrency in Java: Executor Service (Part 1)

Published in

Geek Culture

6 min readMar 24, 2022

Overview

The Java ExecutorService is a JDK API that allows the user to pass tasks to be executed by a pool of threads asynchronously. It can create and maintain a reusable pool of threads for executing submitted tasks. The service also manages a queue, which is used to queue up tasks in case the number of tasks exceeds the number of threads in the thread pool. The Java ExecutorService interface is present in the java.util.concurrent package.

Thread Pool

If we create a new thread to execute every new request it has several disadvantages. More time is spent in creation and destruction of threads than to process actual requests. Creation of too many threads can lead to system being ran out of memory.

Aforementioned reasons necessitates the creation of a pool of limited number of threads that are created beforehand, and can execute tasks as soon as they come. The delay introduced by thread creation is also eliminated.

To use thread pools, we first create an object of ExecutorService and pass a set of tasks to it. ThreadPoolExecutor class allows to set the core and maximum pool size.The runnables that are run by a particular thread are executed sequentially.

The number of threads that can be executed simultaneously is equal to the number of CPU threads in Java. If we have too many threads comparable to CPU cores we have a time splitting scheduling. So having a large number of threads in the thread pool is not optimal in case of CPU intensive operations.

On the other hand, in case we make IO calls from our threads (HTTP call, db calls etc.), we make a call and need to wait for the response from the resource. The thread is stuck in wait state and not dependent on CPU. So in these cases we only need to consider memory consumption as the parameter.

Blocking queue

A blocking queue is a queue that blocks when you try to dequeue from it and the queue is empty, or if you try to enqueue items to it and the queue is already full, similar to a Bounded Semaphore. A thread trying to dequeue from an empty queue is blocked until some other thread inserts an item into the queue. A thread trying to enqueue an item in a full queue is blocked until some other thread makes space in the queue, either by dequeuing one or more items or clearing the queue completely. it’s implemented in the java.util.concurrent package.

Types of Thread pools

Java supports four kinds of thread pools:

Fixed Thread Pool: We have fixed number of threads which pick up tasks assigned to it. All tasks are stored in a thread safe blocking queue.
CachedThreadPool: We do not have a fixed number of threads here. The blocking queue is replaced by a synchronous queue which only has space for one task. A new request is stored in the queue while it searches for any available thread. If no thread is available then it’ll create a new thread and add it to the pool. It also has the ability to kill threads which have been idle for more than 60 seconds.
ScheduledThreadPool: It is used for tasks which need to be scheduled after a delay. We can configure a one-off delay, or a periodic schedule, or even a schedule with a fixed delay. A delay queue is used here, because of which the order of tasks is not in order.
SingleThreadedExecutor : It is similar to a fixed thread executor, but the size of the blocking queue is 1. In this case if the thread is killed because of an exception, a new thread is created. It is used when we want to make sure of the order of execution (sequentially).

Instantiating the ExecutorService

We can instantiate the ExecutorService in two ways:

Implement Factory Methods of Executors class: The easiest way to create ExecutorService is to use one of the factory methods of the Executors class.For example, the following line of code will create a thread pool with 10 threads:

ExecutorService executor = Executors.newFixedThreadPool(10);

There are several other factory methods to create a predefined ExecutorService that meets specific use cases. To find the best method for your needs, consult Oracle’s official documentation.

Call the constructor of ExecutorService: Because ExecutorService is an interface, an instance of any its implementations can be used. There are several implementations to choose from in the java.util.concurrent package, or you can create your own.

For example, the ThreadPoolExecutor class has a few constructors that we can use to configure an executor service and its internal pool:

ExecutorService executorService = 
  new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,   
  new LinkedBlockingQueue<Runnable>());

There are several parameters in the constructor, which can be seen from the source code definition:

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          RejectedExecutionHandler handler) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), handler);

Following is the explanation of the parameters:

The values of corePoolSize, maxPoolSize and keepAliveTime for all four thread pool types are:

The keepAliveTime of 0 seconds means that it’s not applicable for those thread pool types. This is because we do not have a mechanism to kill threads in FixedThreadPool and SingleThreaded thread pools.

Similarly, the kind of queues for all these pool types are as followed:

The RejectionHandled comes into play when the queue is full and a new task is requested. We can define policies that will define what happens in these scenarios. The policies are:

Shutting Down a Thread Pool

In general, the ExecutorService will not be automatically destroyed when there is no task to process. It will stay alive and wait for new work to do. To shut the service down we have the shutdown() and shutdownNow() APIs.

The shutdown() method doesn’t cause immediate destruction of the ExecutorService. It will make the ExecutorService stop accepting new tasks and shut down after all running threads finish their current work:

executorService.shutdown();

The shutdownNow() method tries to destroy the ExecutorService immediately, but it doesn’t guarantee that all the running threads will be stopped at the same time:

List<Runnable> notExecutedTasks = executorService.shutDownNow();

This method returns a list of tasks that are waiting to be processed. It is up to the developer to decide what to do with these tasks.

The recommended method to shutdown the ExecutorService is a combination of the two methods. The ExecutorService can first stop taking new tasks and then wait up to a specified period of time for all tasks to be completed. If that time expires, the execution is stopped immediately. This is implemented using the awaitTermination() method.

Congratulations on making it to the end! Feel free to talk about tech or any cool projects on Twitter, GitHub, Medium, LinkedIn, or Instagram.

Thanks for reading!