Mastering Java Thread Pools: Deep Dive into ThreadPoolExecutor Architecture

Introduction

We know that creating and destroying threads is expensive because it requires OS resources. To avoid frequent creation and to simplify management, a thread pool is introduced.

Advantages of Thread Pools

Reduced resource consumption : core threads are reused, avoiding repeated creation and destruction.

Improved response time : tasks can be executed immediately by existing alive threads.

Better manageability : threads can be centrally allocated, tuned and monitored.

Design Analogy

Thread pools can be compared to a factory: core workers are permanent staff, temporary workers handle peak load, the task queue is the warehouse, and the scheduler is the dispatcher.

ThreadPoolExecutor Constructors

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) { ... }

Four overloaded constructors allow specifying thread factory and rejection handler.

Key Parameters

corePoolSize : number of core threads that stay alive.

maximumPoolSize : maximum number of threads.

keepAliveTime and unit : idle timeout for non‑core threads.

workQueue : task queue implementation.

threadFactory (optional): creates new threads.

handler (optional): rejection policy.

Task Queues

Three main queue types are recommended:

SynchronousQueue : no capacity, hand‑off directly.

LinkedBlockingQueue : optionally unbounded, based on linked nodes.

ArrayBlockingQueue : bounded array‑based queue.

Additional queues include PriorityBlockingQueue, DelayQueue, LinkedBlockingDeque, and LinkedTransferQueue.

Rejection Policies

AbortPolicy (default): throws RejectedExecutionException.

CallerRunsPolicy : runs the task in the calling thread.

DiscardPolicy : silently discards the task.

DiscardOldestPolicy : removes the oldest queued task and retries.

Thread Factory

static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;
    DefaultThreadFactory() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-";
    }
    public Thread newThread(Runnable r) {
        Thread t = new Thread(group, r,
                namePrefix + threadNumber.getAndIncrement(), 0);
        if (t.isDaemon())
            t.setDaemon(false);
        if (t.getPriority() != Thread.NORM_PRIORITY)
            t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }
}

Thread Pool States

volatile int runState;
private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

Initialization, Capacity Adjustment and Shutdown

Methods prestartCoreThread() and prestartAllCoreThreads() create core threads eagerly. shutdown() stops accepting new tasks and waits for queued tasks to finish; shutdownNow() attempts immediate termination and returns pending tasks. Dynamic resizing is possible via setCorePoolSize and setMaximumPoolSize.

Using ThreadPoolExecutor Directly

ThreadPoolExecutor pool = new ThreadPoolExecutor(
        3, 5, 5, TimeUnit.SECONDS,
        new ArrayBlockingQueue<Runnable>(5));
for (int i = 0; i < pool.getCorePoolSize(); i++) {
    pool.execute(new Runnable() {
        public void run() {
            // task logic
        }
    });
}
pool.shutdown();

Executors Convenience Methods

Four common factories are provided:

newFixedThreadPool(int n) : fixed size, no keep‑alive.

newSingleThreadExecutor() : single worker thread.

newScheduledThreadPool(int n) : supports delayed and periodic tasks.

newCachedThreadPool() : creates threads as needed, reuses idle threads, uses SynchronousQueue.

Worker Class

private final class Worker extends AbstractQueuedSynchronizer
        implements Runnable {
    final Thread thread;
    Runnable firstTask;
    volatile long completedTasks;
    Worker(Runnable firstTask) {
        setState(-1); // inhibit interrupts
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }
    public void run() { runWorker(this); }
    // lock methods omitted for brevity
}

runWorker Method

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        while (task != null || (task = getTask()) != null) {
            w.lock();
            if (runStateAtLeast(ctl.get(), STOP) ||
                (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) {
                wt.interrupt();
            }
            try {
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    task.run();
                } catch (RuntimeException x) { thrown = x; throw x; }
                catch (Error x) { thrown = x; throw x; }
                catch (Throwable x) { thrown = x; throw new Error(x); }
                finally { afterExecute(task, thrown); }
            } finally {
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

getTask Method

private Runnable getTask() {
    boolean timedOut = false;
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
        if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
        int wc = workerCountOf(c);
        boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
        if ((wc > maximumPoolSize || (timed && timedOut)) &&
            (wc > 1 || workQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c))
                return null;
            continue;
        }
        try {
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
            if (r != null)
                return r;
            timedOut = true;
        } catch (InterruptedException retry) {
            timedOut = false;
        }
    }
}

processWorkerExit Method

private void processWorkerExit(Worker w, boolean completedAbruptly) {
    if (completedAbruptly)
        decrementWorkerCount();
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        completedTaskCount += w.completedTasks;
        workers.remove(w);
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
    int c = ctl.get();
    if (runStateLessThan(c, STOP)) {
        if (!completedAbruptly) {
            int min = allowCoreThreadTimeOut ? 0 : corePoolSize;
            if (min == 0 && !workQueue.isEmpty())
                min = 1;
            if (workerCountOf(c) >= min)
                return;
        }
        addWorker(null, false);
    }
}