Java’s Executor Framework for Starters

#java #programming #tutorial #development

Consider that you are working on a Web-app’s server. A client would make an HTTP/HTTPS request to your server and would expect some data in response. From your server’s perspective, each client request is a task and can be executed independent of another client request if required.

Once the tasks have been identified, you need to think of their execution policy. Should all the tasks be executed in a single thread? Or, should each task spawn a new thread, execute, and kill the thread? Or, should their be some different approach altogether?

Let us say, you ended up with the choice of executing all tasks in a single thread. This is how this code will look like.

class WebServer {
  public static void main(String[]args) {
    ServerSocket socket = new ServerSocket(80);
    while(true) {
      Socket connection = socket.accept();
      handleRequest(connection);
    }
  }
}

In the code above, each request is handled by the main thread itself and any new request would have to wait at the socket until earlier requests are all served.

You realised that this is not ideal and would highly impact the availability of your server. Thus, you decided to change the execution policy to ‘spawn a new thread per request’. This is how the code after the execution policy change looks.

class WebServer {
  public static void main(String[]args) {
    ServerSocket socket = new ServerSocket(80);
    while(true) {
      Socket connection = socket.accept();
      new Thread(() -> handleRequest(connection)).start(); 
    }
  }
}

Notice how a change in task execution policy required you to change the code that handled incoming requests.

Moving to a thread-per-request model might have given you improvement at first, but you realise that even this method is not free of issues. Creating too many threads is also not helpful since your machine can only process a certain number of threads concurrently at once. Thus, you require a more sophisticated execution policy.

Implementing such complicated execution policy right around your task handling code is not a good idea. Your task handling code will get convoluted and would be hard to maintain.

So, what is the solution? Executor Framework it is!

Executor Framework

Executor framework in Java provides an easy and intuitive way for asynchronous task execution. It helps you decouple the task creation logic from task execution logic.

The Executor interface sits at the core of this powerful framework. Here is how this interface is defined.

public interface Executor {
  void execute (Runnable command);
}

Despite being simple, this interface is really powerful. It helps you abstract out the task execution policy and provides you with a simple method to submit your tasks.

There are a number of standard implementations for Executor that are available in Java’s Executors class. Let us use one such implementation to improve our server class.

class WebServer {
  private static final Executor executor = Executors.newFixedThreadPool(100);
  public static void main(String[] args) {
    ServerSocket socket = new ServerSocket(80);
    while(True) {
      Socket connection = socket.accept();
      execute(() -> handleRequest(connection));
    }
  }
}

As evident from the code above, it now becomes extremely simple to change the task execution policy. Just replace the actual executor implementation used and you would have your existing tasks executed using the newer policy.

Okay. That is all good. But what is a newFixedThreadPool? Or, what is thread pool?

Thread Pools

Thread pools are just homogenous pools of threads that are bound to some sort of queue holding tasks to be executed. Threads in a thread pool waits for a new task to be assigned, execute a task if assigned, and goes back to waiting for another task.

There are a number of thread pool implementations provided in the class library. Some of them are as follows —

newFixedThreadPool: Create threads as task are submitted up to a fixed maximum. Once the maximum is reached, it tries to keep the number of threads stable by creating new threads for destroyed threads.
newCachedThreadPool: This pool is more flexible in terms of creating and deleting threads. When the tasks are relatively lesser than the number of threads, it downsizes the pool. On the other hand, when number of tasks grows, it upsizes the pool without any upper bound like fixed thread pool.
newSingleThreadExecutor: Pool with a single thread. All tasks are executed inside the same thread. Executor makes sure to replace the thread if it dies. It also ensures that tasks are executed in a certain order that is enforced by the work queue.
newScheduledThreadPool: A fixed-size thread pool that support delayed and periodic task execution. ## Executor Lifecycle Once we have created an executor, we would want to shut it down at some later point in time. Shutting down an executor would mean something like killing idle threads, signalling active threads to finish what they are doing and exit.

To address the issues around executor lifecycle, a new interface called ExecutorService extending the original Executor interface is added. This new interface contains a number of methods to manage an executor lifecycle.

public interface ExecutorService extends Exectuor {
  void shutdown();
  List<Runnable> shutdownNow();
  boolean isShutDown();
  boolean isTerminated();
  boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
}

An ExecutorService can be in one of the following three states at any time,

Running: Executor is in running state and will accept new tasks.
Shutting Down: Executor shut down has been initiated. No new tasks will be accepted.
Terminated: Executor has terminated. No new tasks can be submitted.

A graceful shutdown of an executor can be initiated using the shutdown method. A more abrupt shutdown can be requested using the shutdownNow method.

Let us add shutdown logic to our webserver.

class WebServer {
  private final ExecutorService executor = Executors.newCachedThreadPool();
  public static void main(String[]args) {
    ServerSocket socket = new ServerSocket(80);
    while(!executor.isShutdown()) {
      try {
        final Socket connection = socket.accept();
        executor.execute(() -> handleRequest(connection));
      } catch (RejectedExecutionException e) {
        if(!executor.isShutdown()) {
          log("Task submission Rejected");
        }
      }
    }
  }

  static void handleRequest(Socket connection) {
    if(isShutdownRequest(connection)) {
      executor.shutdown();
    } else {
      // Process Request
    }
  }
}