Why we need CustomThreadPoolExecutor when we can use ExecutorService Framework to create and manage Threads?

ExecutorService executor = Executors.newFixedThreadPool(20);

is nothing but

return new ThreadPoolExecutor(20, 20,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());

under the hoods.

ThreadPoolExecutor would be more effective if you have customized many or all of below parameters.

ThreadPoolExecutor(int corePoolSize, 
               int maximumPoolSize, 
               long keepAliveTime, 
               TimeUnit unit, 
               BlockingQueue<Runnable> workQueue, 
               ThreadFactory threadFactory,
               RejectedExecutionHandler handler)

How it works?

1. If fewer than corePoolSize threads are running, try to start a new thread with the given command as its first task. The call to addWorker atomically checks runState and workerCount, and so prevents false alarms that would add threads when it shouldn’t, by returning false.
2. If a task can be successfully queued, then we still need to double-check whether we should have added a thread (because existing ones died since last checking) or that the pool shut down since entry into this method. So we recheck state and if necessary roll back the enqueuing if stopped, or start a new thread if there are none.
3. If we cannot queue task, then we try to add a new thread. If it fails, we know we are shut down or saturated and so reject the task.

CustomThreadPoolExecutor.java

import java.util.concurrent.*;

public class CustomThreadPoolExecutor extends ThreadPoolExecutor {
    public CustomThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
                                    TimeUnit unit, BlockingQueue<Runnable> workQueue,
                                    ThreadFactory threadFactory,
                                    RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
    }
}

CustomThreadFactory.java

import java.util.concurrent.ThreadFactory;

public class CustomThreadFactory implements ThreadFactory {
    @Override
    public Thread newThread(Runnable r) {
        Thread th = new Thread(r);
        th.setPriority(Thread.NORM_PRIORITY);
        th.setDaemon(false);
        return th;
    }
}

CustomRejectHandler.java

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

public class CustomRejectHandler implements RejectedExecutionHandler {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        System.out.println("Task Rejected: "+ r.toString());
    }
}

Task.java

import java.util.concurrent.ThreadPoolExecutor;

public class Task implements Runnable{

    String taskName;
    ThreadPoolExecutor executor;
    Long timeInMilliSeconds;


    public Task(String taskName, ThreadPoolExecutor executor, Long timeInMilliSeconds) {
        this.taskName = taskName;
        this.executor = executor;
        this.timeInMilliSeconds = timeInMilliSeconds;
    }

    @Override
    public void run() {
        try{
            Thread.sleep(this.timeInMilliSeconds);

            System.out.print("Tasks in Blocking Queue "+ this.executor.getQueue().stream().toList() + ", ");
            System.out.print(this.taskName + " completed by " +  Thread.currentThread().getName() + " after running "+ timeInMilliSeconds +"ms" );
            System.out.println(", Active Threads available "+ executor.getPoolSize());
        }catch (Exception e){
        }
    }

    @Override
    public String toString() {
        return "Task{" +
                "taskName='" + taskName + '\'' +
                '}';
    }
}

BatchProcessor.java

public class BatchProcessor {
    public static void main(String[] args) throws InterruptedException {

        ThreadPoolExecutor executor = new CustomThreadPoolExecutor(2,4, 10, TimeUnit.MINUTES,
                                                                    new ArrayBlockingQueue<>(2),
                                                                    new CustomThreadFactory(),
                                                                    new CustomRejectHandler());

        System.out.println("Active Threads  available for processing at start "+  executor.getPoolSize());

        executor.submit( new Task("task1", executor, 2500L)); //Directly dealt by CorePool Thread
        executor.submit( new Task("task2", executor, 500L)); //Directly dealt by CorePool Thread
        Thread.sleep(2000L);
        System.out.println("Slept for 2000 Millisecond");
        executor.submit( new Task("task3", executor, 200L)); //Directly dealt by CorePool Thread
        executor.submit( new Task("task4", executor, 1000L)); //Directly dealt by CorePool Thread
        executor.submit( new Task("task5", executor, 300L)); //Dealt by extra thread within Maximum Pool Size
        executor.submit( new Task("task6",executor, 300L)); //Directly dealt by CorePool Thread

        executor.shutdown();
    }
}

Output

Active Threads  available for processing at start 0
Tasks in Blocking Queue [], task2 completed by Thread-1 after running 500ms, Active Threads available 2
Slept for 2000 Millisecond
Tasks in Blocking Queue Task{'task4','task6'}, task3 completed by Thread-1 after running 200ms, Active Threads available 3
Tasks in Blocking Queue Task{'task6'}, task5 completed by Thread-2 after running 300ms, Active Threads available 3
Tasks in Blocking Queue [], task1 completed by Thread-0 after running 2500ms, Active Threads available 3
Tasks in Blocking Queue [], task6 completed by Thread-2 after running 300ms, Active Threads available 2
Tasks in Blocking Queue [], task4 completed by Thread-1 after running 1000ms, Active Threads available 1