{"id":5487,"date":"2025-03-16T11:22:30","date_gmt":"2025-03-16T11:22:30","guid":{"rendered":"https:\/\/codethataint.com\/blog\/?p=5487"},"modified":"2025-03-17T14:09:36","modified_gmt":"2025-03-17T14:09:36","slug":"how-threadpoolexecutor-works","status":"publish","type":"post","link":"https:\/\/codethataint.com\/blog\/how-threadpoolexecutor-works\/","title":{"rendered":"CustomThreadPoolExecutor – Why, What, How?"},"content":{"rendered":"

Why we need CustomThreadPoolExecutor when we can use ExecutorService Framework to create and manage Threads?<\/strong><\/p>\n

\r\nExecutorService executor = Executors.newFixedThreadPool(20);\r\n<\/pre>\n
is nothing but <\/p>\n
\r\nreturn new ThreadPoolExecutor(20, 20,\r\n                                  0L, TimeUnit.MILLISECONDS,\r\n                                  new LinkedBlockingQueue<Runnable>());\r\n<\/pre>\n
under the hoods.<\/p>\n
ThreadPoolExecutor would be more effective if you have customized many or all of below parameters.<\/p>\n
\r\nThreadPoolExecutor(int corePoolSize, \r\n               int maximumPoolSize, \r\n               long keepAliveTime, \r\n               TimeUnit unit, \r\n               BlockingQueue<Runnable> workQueue, \r\n               ThreadFactory threadFactory,\r\n               RejectedExecutionHandler handler)\r\n<\/pre>\n
How it works?<\/strong><\/p>\n
    \n
  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.<\/li>\n
  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.<\/li>\n
  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.<\/li>\n<\/ol>\n
    CustomThreadPoolExecutor.java<\/strong><\/p>\n
    \r\nimport java.util.concurrent.*;\r\n\r\npublic class CustomThreadPoolExecutor extends ThreadPoolExecutor {\r\n    public CustomThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,\r\n                                    TimeUnit unit, BlockingQueue<Runnable> workQueue,\r\n                                    ThreadFactory threadFactory,\r\n                                    RejectedExecutionHandler handler) {\r\n        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);\r\n    }\r\n}\r\n<\/pre>\n
    CustomThreadFactory.java<\/strong><\/p>\n
    \r\nimport java.util.concurrent.ThreadFactory;\r\n\r\npublic class CustomThreadFactory implements ThreadFactory {\r\n    @Override\r\n    public Thread newThread(Runnable r) {\r\n        Thread th = new Thread(r);\r\n        th.setPriority(Thread.NORM_PRIORITY);\r\n        th.setDaemon(false);\r\n        return th;\r\n    }\r\n}\r\n<\/pre>\n
    CustomRejectHandler.java<\/strong><\/p>\n
    \r\nimport java.util.concurrent.RejectedExecutionHandler;\r\nimport java.util.concurrent.ThreadPoolExecutor;\r\n\r\npublic class CustomRejectHandler implements RejectedExecutionHandler {\r\n    @Override\r\n    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {\r\n        System.out.println("Task Rejected: "+ r.toString());\r\n    }\r\n}\r\n<\/pre>\n
    Task.java<\/strong><\/p>\n
    \r\nimport java.util.concurrent.ThreadPoolExecutor;\r\n\r\npublic class Task implements Runnable{\r\n\r\n    String taskName;\r\n    ThreadPoolExecutor executor;\r\n    Long timeInMilliSeconds;\r\n\r\n\r\n    public Task(String taskName, ThreadPoolExecutor executor, Long timeInMilliSeconds) {\r\n        this.taskName = taskName;\r\n        this.executor = executor;\r\n        this.timeInMilliSeconds = timeInMilliSeconds;\r\n    }\r\n\r\n    @Override\r\n    public void run() {\r\n        try{\r\n            Thread.sleep(this.timeInMilliSeconds);\r\n\r\n            System.out.print("Tasks in Blocking Queue "+ this.executor.getQueue().stream().toList() + ", ");\r\n            System.out.print(this.taskName + " completed by " +  Thread.currentThread().getName() + " after running "+ timeInMilliSeconds +"ms" );\r\n            System.out.println(", Active Threads available "+ executor.getPoolSize());\r\n        }catch (Exception e){\r\n        }\r\n    }\r\n\r\n    @Override\r\n    public String toString() {\r\n        return "Task{" +\r\n                "taskName='" + taskName + '\\'' +\r\n                '}';\r\n    }\r\n}\r\n<\/pre>\n
    BatchProcessor.java<\/strong><\/p>\n
    \r\npublic class BatchProcessor {\r\n    public static void main(String[] args) throws InterruptedException {\r\n\r\n        ThreadPoolExecutor executor = new CustomThreadPoolExecutor(2,4, 10, TimeUnit.MINUTES,\r\n                                                                    new ArrayBlockingQueue<>(2),\r\n                                                                    new CustomThreadFactory(),\r\n                                                                    new CustomRejectHandler());\r\n\r\n        System.out.println("Active Threads  available for processing at start "+  executor.getPoolSize());\r\n\r\n        executor.submit( new Task("task1", executor, 2500L)); \/\/Directly dealt by CorePool Thread\r\n        executor.submit( new Task("task2", executor, 500L)); \/\/Directly dealt by CorePool Thread\r\n        Thread.sleep(2000L);\r\n        System.out.println("Slept for 2000 Millisecond");\r\n        executor.submit( new Task("task3", executor, 200L)); \/\/Directly dealt by CorePool Thread\r\n        executor.submit( new Task("task4", executor, 1000L)); \/\/Directly dealt by CorePool Thread\r\n        executor.submit( new Task("task5", executor, 300L)); \/\/Dealt by extra thread within Maximum Pool Size\r\n        executor.submit( new Task("task6",executor, 300L)); \/\/Directly dealt by CorePool Thread\r\n\r\n        executor.shutdown();\r\n    }\r\n}\r\n<\/pre>\n
    Output<\/strong><\/p>\n
    \r\nActive Threads  available for processing at start 0\r\nTasks in Blocking Queue [], task2 completed by Thread-1 after running 500ms, Active Threads available 2\r\nSlept for 2000 Millisecond\r\nTasks in Blocking Queue Task{'task4','task6'}, task3 completed by Thread-1 after running 200ms, Active Threads available 3\r\nTasks in Blocking Queue Task{'task6'}, task5 completed by Thread-2 after running 300ms, Active Threads available 3\r\nTasks in Blocking Queue [], task1 completed by Thread-0 after running 2500ms, Active Threads available 3\r\nTasks in Blocking Queue [], task6 completed by Thread-2 after running 300ms, Active Threads available 2\r\nTasks in Blocking Queue [], task4 completed by Thread-1 after running 1000ms, Active Threads available 1\r\n<\/pre>\n","protected":false},"excerpt":{"rendered":"
    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… Continue reading →<\/span><\/a><\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"closed","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[268],"tags":[],"class_list":["post-5487","post","type-post","status-publish","format-standard","hentry","category-concepts-threads"],"_links":{"self":[{"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/posts\/5487","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/comments?post=5487"}],"version-history":[{"count":5,"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/posts\/5487\/revisions"}],"predecessor-version":[{"id":5498,"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/posts\/5487\/revisions\/5498"}],"wp:attachment":[{"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/media?parent=5487"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/categories?post=5487"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/codethataint.com\/blog\/wp-json\/wp\/v2\/tags?post=5487"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}