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?
- 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.
- 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.
- 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