How to create a Generic Executor Service for asynchronous calls

Hello Everyone! 🙂

I am sure a lot of you know what are synchronous and asynchronous calls. Just in case you don’t:

Synchronous
If an API call is synchronous, it means that code execution will block (or wait) for the API call to return before continuing. This means that until a response is returned by the API, your application will not execute any further, which could be perceived by the user as latency or performance lag in your app. Making an API call synchronously can be beneficial, however, if there is code in your app that will only execute properly once the API response is received.

Asynchronous
Asynchronous calls do not block (or wait) for the API call to return from the server. Execution continues on in your program, and when the call returns from the server, a “callback” function is executed.

For more details, refer – https://docs.apigee.com/api-baas/get-started/asynchronous-vs-synchronous-calls

When we talk about an asynchronous call implementation – we are talking about ExecutorService with Callable interface at ground level. And then it’s quite obvious that we want a Future objects which hold the data with its type as the return value. And you don’t want to write same logic for Future objects for different data types. So instead, we are going to use Generics to create a generic or common ExecutorService class which can be used for any data type.

The code:

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class ExecutorServiceProvider<T> {

  private ThreadPoolExecutor threadPoolExecutor;
  
  public ExecutorService getExecutorService() {
    return threadPoolExecutor = (null == threadPoolExecutor || threadPoolExecutor.isShutdown()) ? (ThreadPoolExecutor) Executors.newFixedThreadPool(5) : threadPoolExecutor;
  }
  
  public ExecutorService getExecutorService(int threadCount) {
    return threadPoolExecutor = (null == threadPoolExecutor || threadPoolExecutor.isShutdown()) ? (ThreadPoolExecutor) Executors.newFixedThreadPool(threadCount) : threadPoolExecutor;
  }
  
  public void shutdownExecutorService() {
    if(null != threadPoolExecutor || !threadPoolExecutor.isShutdown()) {
      threadPoolExecutor.shutdown();
    }
  }
  
  public void shutdownNowExecutorService() {
    if(null != threadPoolExecutor || !threadPoolExecutor.isShutdown()) {
      threadPoolExecutor.shutdownNow();
    }
  }
  
  public Future<T> getResult(int size, final ExecutorCompletionService<T> executorCompletionService, final long timeout, final TimeUnit timeUnit) {
    Future<T> futureList = null;
    
    long globalWaitTime = timeUnit.toNanos(timeout);
    
    for(int i = 0; i < size; i++) {
      final long waitStart = System.nanoTime();
      
      try {
        futureList = executorCompletionService.take();
        if(futureList != null) {
          try {
            futureList.get(globalWaitTime, TimeUnit.NANOSECONDS);
          } catch (InterruptedException | ExecutionException | TimeoutException e) {
            futureList.cancel(true);
          }
        }
      } catch (InterruptedException e) {
        futureList.cancel(true);
      } finally {
        final long waitFinish = System.nanoTime() - waitStart;
        globalWaitTime = Math.max(globalWaitTime - waitFinish, 0);
      }
    }
    return futureList;
  }

  public void shutdownAndAwaitTermination() {
		threadPoolExecutor.shutdown();
	    try {
	        if (!threadPoolExecutor.awaitTermination(60, TimeUnit.SECONDS)) {
	        	threadPoolExecutor.shutdownNow();
	            if (!threadPoolExecutor.awaitTermination(60, TimeUnit.SECONDS))
	                System.err.println("Pool did not terminate");
	        }
	    } catch (InterruptedException ie) {
	    	threadPoolExecutor.shutdownNow();
	        Thread.currentThread().interrupt();
	    }
	}
}

This class gives you flexibility to change the time that you want to wait for Future objects to be returned.

And that’s it.

You can browse the code here: https://github.com/niteshapte/generic-executor-service-java

Please check out my next post where I have described how to use above code. It might solve a lot of your problem and make you happy.

Critics/feedbacks are very much welcome.

Have a nice day ahead!

Stay Awesome!