Spring batch - how to dynamically increase and decrease threads using ThreadPoolTaskExecutor

Question

I'm fairly new to Spring-Batch so this may be a lack of understanding on my part. I'm wanting to understand how to dynamically increase and decrease threads using the ThreadPoolTaskExecutor in conjunction with the ThreadPoolExecutor while my job is running. I've tried to subclass both the ThreadPoolTaskExecutor and the ThreadPoolExecutor so I can gain access to the beforeExecute() and afterExecute() which would allow me to terminate threads if the corepoolsize was decreased using an approach that is listed on this site.

What I seem to be not understanding is that when I override the initializeExecutor() method which returns an ExecutorService, it apparently does not set the (private internal) threadPoolExecutor variable in the parent class (ThreadPoolTaskExecutor). It sets the private ExecutorService executor; (from the ExecutorConfigurationSupport class)

Since the threadPoolExecutor is not a protected member I cannot gain access to it. Without that being set, when I run I obviously end up getting a "ThreadPoolExecutor not initialized" error within the Spring Framework when I examine what's wrong under the covers.

public class MyThreadPoolTaskExecutor extends ThreadPoolTaskExecutor {

@Override
protected ExecutorService initializeExecutor(ThreadFactory tf, RejectedExecutionHandler reh)
{

    BlockingQueue <Runnable> queue = createQueue(Integer.MAX_VALUE);

    MyThreadPoolExecutor tp_executor = new MyThreadPoolExecutor( this.getCorePoolSize(), this.getMaxPoolSize(),     this.getKeepAliveSeconds(), TimeUnit.SECONDS, queue);

    // if you look at the parent class(ThreadPoolTaskExecutor) it performs this call next.
    //  this.threadPoolExecutor = executor;
    // that is a private member with no ability to set via any methods.

    return tp_executor;
}

}

public class MyThreadPoolExecutor extends ThreadPoolExecutor {

public MyThreadPoolExecutor(int corePoolSize, int maxPoolSize, long keepAliveTimeout, TimeUnit timeunit, BlockingQueue<Runnable> workQueue, ThreadFactory tf, RejectedExecutionHandler reh) 
{
    super(corePoolSize, maxPoolSize, keepAliveTimeout, timeunit, workQueue, tf, reh);
}

protected void beforeExecute (final Thread thread, final Runnable job) 
{   
  ...
}

}

Can someone explain what I am missing in my approach?

Why are you trying to dynamically change the amount of threads you're running? — Ishmael
We have a multitude of jobs running on our production boxes. If capacity were to free up on a box for a window of time our prod support may want to throw more threads at a job for some period of time. The ability to dynamically increase and/or decrease threads for a period of time can be a useful tool for prod support when there is a lot of activity on a box. — Ed T
This sounds like a solution for dynamic resource management. How are you determining that more resources are available or are you guys trying to change amount of threads on the fly? — Ishmael
Our prod support team utilizes tooling and if available cpu's become available they would want to increase the # of threads for this job on the fly to achieve maximum throughput. — Ed T

featuredpeow featuredpeow · Accepted Answer · 2016-08-12T22:43:40

I assume you want to use one number of threads in one job step and another number of threads in another job step. Simple way to achieve that would be to declare two separate executors with necessary number of threads, zero corePoolSize (to not create threads when this is not necessary) and zero keepAliveSeconds (to not keep threads when this is no longer necessary). Then just inject first executor in one step and second executor in another step.

@Configuration
public class Conf {

    @Bean
    public TaskExecutor executorA(@Value("${first.number.of.threads}") int numberOfThreads) {
        return executor(numberOfThreads);
    }

    @Bean
    public TaskExecutor executorB(@Value("${second.number.of.threads}") int numberOfThreads) {
        return executor(numberOfThreads);
    }

    private TaskExecutor executor(int numberOfThreads) {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(0);
        executor.setMaxPoolSize(numberOfThreads);
        executor.setAllowCoreThreadTimeOut(true);
        executor.setKeepAliveSeconds(0);
        return executor;
    }
}

Spring batch - how to dynamically increase and decrease threads using ThreadPoolTaskExecutor

1 Answers