Deep dive into Java Executor service framework - Part 1

Suhas Bharadwaj — Fri, 21 Dec 2018 13:40:54 +0000

Introduction:

Hello folks! This is my first technical blog that am starting to write under my domain. I used to be an avid consumer of blog posts before, but it was time to move on to my own sweet space on the internet as a producer of content as I feel it is important to share my learnings with the community.

Here is the link to the original article !

In this blog post, I'd like to cover in detail, the implementation of Executor Service framework released with the JDK 5. It is used to run the Runnable objects using concepts such as ThreadPool and ThreadFactory.

If you are planning to implement some kind of concurrency in your project, I'd recommend for you to use higher level frameworks like this one, instead of going at it from scratch since that would ensure that your codebase is less affected by runtime datarace kind of situations and leave most of the heavy lifting of managing threads and synchronization to something that is more deterministic

1. Executors.java Factory

The first thing that we do when we are using the Executor Service framework is to use the Executors factory class to return an instance of an object that implements all the methods of the ExecutorService interface. We call the public static threadpool instance creation methods on the constructor to get that specific wrapped ExecutorService implementation back. Once we get an instance of that service, we are in a positioned to submit tasks to the service and track the progress on each of those tasks. You can find this class under package java.util.concurrent. Below is a sample snippet of achieving what I just discussed:

ExecutorService executorService = Executors.newFixedThreadPool(10);

try{
    executorService.execute(new Runnable() {
        public void run() {
            System.out.println("Asynchronous task");
        }
    });
}

finally {
    executorService.shutdown(); 
}

Different thread pools implementations that are instantiable from the Executors class are :

Fixed Thread Pool : Creates a thread pool that reuses a fixed number of threads operating off a shared unbounded queue. At any point, at most nThreads will be active processing tasks. I'll be explaining in depth, in Part 2 of the series, the double click on this implementation
Single Thread Executor : Creates a single-threaded executor that can schedule commands to run after a given delay, or to execute periodically. (Note however that if this single thread terminates due to a failure during execution prior to shutdown, a new one will take its place if needed to execute subsequent tasks.) Tasks are guaranteed to execute sequentially, and no more than one task will be active at any given time. The returned executor is guaranteed not to be reconfigurable to use additional threads.
Cached Thread Pool : Creates a thread pool that creates new threads as needed, but will reuse previously constructed threads when they are available. These pools will typically improve the performance of programs that execute many short-lived asynchronous tasks.

2. ExecutorService Interface

ExecutorService interface extends Executor interface which is an object that executes submitted Runnable tasks. This interface provides a way of decoupling task submission from the mechanics of how each task will be run, including details of thread use, scheduling, etc. An Executor is normally used instead of explicitly creating threads. For example, rather than invoking new Thread(new(RunnableTask())).start() for each of a set of tasks, you might use. Here is a concrete class that implements this interface :

class ThreadPerTaskExecutor implements Executor {
   public void execute(Runnable r) {
     new Thread(r).start();
   }
}

Sample of snippet of its invocation :

Executor executor = new ThreadPerTaskExecutor();
executor.execute(new RunnableTask1());
executor.execute(new RunnableTask2());

Some of main methods from the ExecutorService interface that need to be implemented by the implementation classes are:

void shutdown(); : Initiates an orderly shutdown in which previously submitted tasks are executed, but no new tasks will be accepted. Invocation has no additional effect if already shut down.
List<Runnable> shutdownNow(); : Attempts to stop all actively executing tasks, halts the processing of waiting tasks, and returns a list of the tasks that were awaiting execution.
boolean isShutdown(); : Returns true if this executor has been shut down.
boolean isTerminated(); : Returns true if all tasks have completed following shut down.
boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException; : Blocks until all tasks have completed execution after a shutdown * request, or the timeout occurs, or the current thread is interrupted, whichever happens first.
<T> Future<T> submit(Callable<T> task); : Submits a value-returning task for execution and returns a Future representing the pending results of the task. The Future's get method will return the task's result upon successful completion.
<T> Future<T> submit(Runnable task, T result); : Submits a Runnable task for execution and returns a Future representing that task. The Future's get method will return the given result upon successful completion.
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException; : Executes the given tasks, returning a list of Futures holding their status and results when all complete. Future#isDone is true for each element of the returned list. Note that a completed task could have terminated either normally or by throwing an exception. The results of this method are undefined if the given collection is modified while this operation is in progress.

3. Thread Pool Executor

An implementation of the ExecutorService that executes each submitted task using one of possibly several pooled threads, normally configured using Executors factory methods. Thread pools address two different problems: they usually provide improved performance when executing large numbers of asynchronous tasks, due to reduced per-task invocation overhead, and they provide a means of bounding and managing the resources, including threads, consumed when executing a collection of tasks. Each ThreadPoolExecutor also maintains some basic statistics, such as the number of completed tasks.

The ThreadPookExecutor extents the AbstractExecutorService that again implements the ExecutorService interface.

The construction of the ThreadPoolExecutor takes in a few mandatory params to initialize in the right manner with the right strategy:

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

You might be wondering what the corePoolSize, maximumPoolSize and 'workQueue' are doing in this context. The corePoolSize tells the ThreadPoolExecutor to initialize a default set of threads ready to pick up the next task that is submitted. The workQueue is a BlockingQueue implementation data structure to queue the requests that exceed the corePoolSize limit. Once corePoolSize is reached, it starts to poll/take this queue for tasks and then schedule one of the existing threads to execute the task. This way, the thread pool executor manages to handle the thead resources in an efficient manner

Closing note

This article is a part 1 of the 3 part series that I intend to publish. This gives a good understanding of the composition of the Executor Service framework implementation and some of the core ideas that make this framework easy and effective to use.

If you have any queries or any constructive criticism to offer, leave it out in the comments below!