Working With ThreadPoolTaskExecutor of Spring

Photo Courtesy Unsplash

ThreadPoolTaskExecutor is a java bean that allows for configuring a ThreadPoolExecutor in a bean style by setting up the values for the instance variables like corePoolSize, maxPoolSize, keepAliveSeconds, queueCapacity and exposing it as a Spring TaskExecutor.

One of the added Advantage of using ThreadPoolTaskExecutor of Spring is that it is well suited for management and monitoring via JMX.

The default configuration of core pool size is 1, max pool size and queue capacity as 2147483647.

This is roughly equivalent to Executors.newSingleThreadExecutor(), sharing a single thread for all tasks. And setting queueCapacity to 0 mimics Executors.newCachedThreadPool(), with immediate scaling of threads in the pool to a very high number.

If you are using XML file for configuring the bean, you can setup the ThreadPoolTaskExecutor like below:

<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
   <property name="corePoolSize" value="5" />
   <property name="maxPoolSize" value="10" />
   <property name="WaitForTasksToCompleteOnShutdown" value="true" />
</bean>

If you are using Java Annotation to define the bean, you can setup like below:

@Bean
public TaskExecutor threadPoolTaskExecutor() {
   ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
   executor.setCorePoolSize(5);
   executor.setMaxPoolSize(10);
   executor.initialize();
   return executor;
}

CorePoolSize

Is the minimum number of threads that remain active at any given point of time. If you don’t provide a value explicitly then it will have default value as 1. The TaskExecutor delegates the value to the underlying class ThreadPoolExecutor.

MaxPoolSize

Is the maximum number of threads that can be created. The TaskExecutor delegates the value to the underlying ThreadPoolExecutor. The maxPoolSize relies on queueCapacity because ThreadPoolTaskExecutor creates a new thread only if the number of items in the queue exceeds queue capacity.

Let’s test out the theory with some code so that it will be more clear for all of us. Here is a code snippet that will create new thread from TaskExecutor.

@Bean
public TaskExecutor threadPoolTaskExecutor() {
   ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
   executor.setCorePoolSize(5);public void createTasks(ThreadPoolTaskExecutor taskExecutor, int numTasks) {

    for (int i=0; i<numTasks; i++) {
       taskExecutor.execute(() -> {
       try {
          long sleepTime = ThreadLocalRandom.current().nextLong(1, 10) * 100;
          Thread.sleep(sleepTime);
       } 
       catch (InterruptedException e) {
          Thread.currentThread().interrupt();
       }
       });
   }
}

Now let’s create a main method where we can try out various use cases by setting values to the ThreadPoolTaskExecutor.

public static void main(String[] args) {
    // Creating ThreadPoolTaskExecutor
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.afterPropertiesSet();

    // Creating Tasks within ThreadPoolTaskExecutor
    createTasks(taskExecutor, 6);

    // Getting PoolSize
    System.out.println("Properties " +
        "- corePoolSize: " + taskExecutor.getCorePoolSize() +
        ", maxPoolSize: " + taskExecutor.getMaxPoolSize() +
        ", poolSize: " + taskExecutor.getPoolSize() +
        ", activeCount: " + taskExecutor.getActiveCount());

    // Shutting Down ThreadPoolTaskExecutor
    taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
    taskExecutor.shutdown();
}

If execute the above code snippet, you will the below output on the console. You can see that corePoolSize is 1 by default and maxPoolSize is 2147483647. Since the QueueCapacity is also 2147483647, you will see that it will make use of the one thread to process all 6 tasks.

17:37:21.696 [main] INFO org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor - Initializing ExecutorService

Properties - corePoolSize: 1, maxPoolSize: 2147483647, poolSize: 1, activeCount: 1

17:37:21.747 [main] INFO org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor - Shutting down ExecutorService

Let’s modify our code to setup some of the parameters so that we can see whether it will have any behavioural change in the TaskExecutor.

ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(1);
taskExecutor.setMaxPoolSize(4);
taskExecutor.afterPropertiesSet();

You will still observe that the TaskExecutor will still make use of one thread only though the max pool size is set to 4.

17:42:43.396 [main] INFO org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor - Initializing ExecutorService

Properties - corePoolSize: 1, maxPoolSize: 4, poolSize: 1, activeCount: 0

17:42:43.447 [main] INFO org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor - Shutting down ExecutorService

Let’s modify our code to setup queue capacity parameter so we see the behavioural change in the TaskExecutor.

ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(1);
taskExecutor.setMaxPoolSize(4);

taskExecutor.setQueueCapacity(2);
taskExecutor.afterPropertiesSet();

Now when you run the same piece of code, you will see that the pool size increases but it will not exceed max pool size.

17:47:29.986 [main] INFO org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor - Initializing ExecutorService

Properties - corePoolSize: 1, maxPoolSize: 4, poolSize: 4, activeCount: 4

17:47:30.036 [main] INFO org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor - Shutting down ExecutorService

TaskExecutor creates a new thread only if the number of items in the queue exceeds queue capacity. As a result, you will observe that pool size increases.

To conclude, it’s always a good practice to define the core pool size, max pool size and queue capacity for the TaskExecutor explicitly from our end instead of leaving it to use the default values.