Concurrency in Java: The Executor Framework

Introduction

With the increase in the number of the cores available in the processors nowadays, coupled with the ever increasing need to achieve more throughput, multi-threading APIs are getting quite popular. Java provides its own multi-threading framework called the Executor Framework.

What is the Executor Framework?

The Executor Framework contains a bunch of components that are used to efficiently manage worker threads. The Executor API de-couples the execution of task from the actual task to be executed via Executors. This design is one of the implementations of the Producer-Consumer pattern.

The java.util.concurrent.Executors provide factory methods which are to be used to create ThreadPools of worker threads.

To use the Executor Framework we need to create one such thread pool and submit the task to it for execution. It is the job of the Executor Framework to schedule and execute the submitted tasks and return the results from the thread pool.

A basic question that comes to mind is why do we need such thread pools when we can create objects of java.lang.Thread or implement Runnable/Callable interfaces to achieve parallelism?

The answer comes down to two basic facts:

1. Creating a new thread for a new task leads to overhead of thread creation and tear-down. Managing this thread life-cycle significantly adds to the execution time.
2. Adding a new thread for each process without any throttling leads to the creation of a large number of threads. These threads occupy memory and cause wastage of resources. The CPU starts to spend too much time switching contexts when each thread is swapped out and another thread comes in for execution.

All these factors reduce the throughput of the system. Thread pools overcome this issue by keeping the threads alive and reusing the threads. Any excess tasks flowing in that the threads in the pool can handle are held in a Queue. Once any of the threads get free, they pick up the next task from this queue. This task queue is essentially unbounded for the out-of-box executors provided by the JDK.

Types of Executors

Now that we have a good idea of what an executor is, let's also take a look at the different kinds of executors.

SingleThreadExecutor

This thread pool executor has only a single thread. It is used to execute tasks in a sequential manner. If the thread dies due to an exception while executing a task, a new thread is created to replace the old thread and the subsequent tasks are executed in the new one.

ExecutorService executorService = Executors.newSingleThreadExecutor()

FixedThreadPool(n)

As the name indicates, it is a thread pool of a fixed number of threads. The tasks submitted to the executor are executed by the n threads and if there are more tasks they are stored on a LinkedBlockingQueue. This number is usually the total number of the threads supported by the underlying processor.

ExecutorService executorService = Executors.newFixedThreadPool(4);

CachedThreadPool

This thread pool is mostly used where there are lots of short-lived parallel tasks to be executed. Unlike the fixed thread pool, the number of threads of this executor pool is not bounded. If all the threads are busy executing some tasks and a new task comes, the pool will create and add a new thread to the executor. As soon as one of the threads becomes free, it will take up the execution of the new tasks. If a thread remains idle for sixty seconds, they are terminated and removed from cache.

However, if not managed correctly, or the tasks are not short-lived, the thread pool will have lots of live threads. This may lead to resource thrashing and hence performance drop.

ExecutorService executorService = Executors.newCachedThreadPool();

ScheduledExecutor

This executor is used when we have a task that needs to be run at regular intervals or if we wish to delay a certain task.

ScheduledExecutorService scheduledExecService = Executors.newScheduledThreadPool(1);

The tasks can be scheduled in ScheduledExecutor using either of the two methods scheduleAtFixedRate or scheduleWithFixedDelay.

scheduledExecService.scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)

scheduledExecService.scheduleWithFixedDelay(Runnable command, long initialDelay, long period, TimeUnit unit)

The main difference between the two methods is their interpretation of the delay between consecutive executions of a scheduled job.

scheduleAtFixedRate executes the task with a fixed interval, irrespective of when the previous task ended.

scheduleWithFixedDelay will start the delay countdown only after the current task completes.

Understanding the Future Object

The result of the task submitted for execution to an executor can be accessed using the java.util.concurrent.Future object returned by the executor. Future can be thought of as a promise made to the caller by the executor.

Future<String> result = executorService.submit(callableTask);

A task submitted to the executor, like above, is asynchronous i.e. the program execution does not wait for the completion of task execution to proceed to the next step. Instead, whenever the task execution is completed, it is set in this Future object by the executor.

The caller can continue executing the main program and when the result of the submitted task is needed he can call .get() on this Future object. If the task is complete the result is immediately returned to the caller or else the caller is blocked until the execution of this is completed by the executor and the result is computed.

If the caller cannot afford to wait indefinitely before retrieving the result, this wait can be timed as well. This is achieved by the Future.get(long timeout, TimeUnit unit) method which throws a TimeoutException if the result is not returned in the stipulated time frame. The caller can handle this exception and continue with the further execution of the program.

If there is an exception when executing the task, the call to get method will throw an ExecutionException.

An important thing with respect to result being returned by Future.get() method is that it is returned only if the submitted task implements java.util.concurrent.Callable. If the task implements the Runnable interface, the call to .get() will return null once the task is complete.

Another important method is the Future.cancel(boolean mayInterruptIfRunning) method. This method is used to cancel the execution of a submitted task. If the task is already executing, the executor will attempt to interrupt the task execution if the mayInterruptIfRunning flag is passed as true.

Example: Creating and Executing a Simple Executor

We will now create a task and try to execute it in a fixed pool executor:

public class Task implements Callable<String> {

    private String message;

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

    @Override
    public String call() throws Exception {
        return "Hello " + message + "!";
    }
}

The Task class implements Callable and is parameterized to String type. It is also declared to throw Exception. This ability to throw an exception to the executor and executor returning this exception back to the caller is of great importance because it helps the caller know the status of task execution.

Now let's execute this task:

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

        Task task = new Task("World");

        ExecutorService executorService = Executors.newFixedThreadPool(4);
        Future<String> result = executorService.submit(task);

        try {
            System.out.println(result.get());
        } catch (InterruptedException | ExecutionException e) {
            System.out.println("Error occurred while executing the submitted task");
            e.printStackTrace();
        }

        executorService.shutdown();
    }
}

Here we have created a FixedThreadPool executor with a count of 4 threads since this demo is developed on a quad-core processor. The thread count can be more than the processor cores if the tasks being executed perform considerable I/O operations or spend time waiting for external resources.

We have instantiated the Task class and are passing it to the executor for execution. The result is returned by the Future object, which we then print on the screen.

Let's run the ExecutorExample and check its output:

Hello World!

As expected, the task appends the greeting "Hello" and returns the result via the Future object.

Lastly, we call the shutdown on the executorService object to terminate all the threads and return the resources back to the OS.

The .shutdown() method waits for the completion of currently submitted tasks to the executor. However, if the requirement is to immediately shut down the executor without waiting then we can use the .shutdownNow() method instead.

Any tasks pending for execution will be returned back in a java.util.List object.

We can also create this same task by implementing the Runnable interface:

public class Task implements Runnable{

    private String message;

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

    public void run() {
        System.out.println("Hello " + message + "!");
    }
}

There are a couple of important changes here when we implement runnable.

1. The result of task execution cannot be returned from the run() method. Hence, we are printing directly from here.
2. The run() method is not configured to throw any checked exceptions.

Conclusion

Multi-threading is getting increasingly mainstream as the processor clock-speed is difficult to increase. However, handling the lifecycle of each thread is very difficult due to the complexity involved.

In this article, we demonstrated an efficient yet simple multithreading framework, the Executor Framework, and explained its different components. We also took a look at different examples of creating, submitting and executing tasks in an executor.

As always, the code for this example can be found on GitHub.