Deep Dive into Java ThreadPoolExecutor: Source Code Analysis (Part 6)

Core Design of Thread Pools

Thread pools reduce the high cost of creating a new thread, allow limiting maximum concurrency, and provide a task queue for unified management.

Creation overhead : creating a thread requires a system call; a pool reuses threads after a one‑time creation.

Resource control : without a pool the number of threads is unbounded; a pool caps the maximum number of concurrent threads.

Task management : a pool can cache tasks in a queue; raw threads cannot.

Lifecycle : a pool offers unified start/stop management.

Seven Core Parameters

public ThreadPoolExecutor(
    int corePoolSize,          // core thread count
    int maximumPoolSize,       // max thread count
    long keepAliveTime,        // idle thread keep‑alive time
    TimeUnit unit,            // time unit
    BlockingQueue<Runnable> workQueue, // task queue
    ThreadFactory threadFactory,       // thread factory
    RejectedExecutionHandler handler   // rejection policy
)

Parameter Decision Flow

New task submitted
    │
    ▼
Current thread count < corePoolSize?
    │
    ├── Yes → create a new core thread
    │
    └── No → try to enqueue
                │
                ├── Success → wait for an idle thread
                │
                └── Failure → current thread count < maximumPoolSize?
                        │
                        ├── Yes → create a non‑core thread
                        │
                        └── No → execute the configured rejection policy

Queue Selection

LinkedBlockingQueue – unbounded (default). Suitable when task volume is unknown and spikes are high.

ArrayBlockingQueue – bounded. Suitable when task volume is controllable and OOM must be avoided.

SynchronousQueue – no storage, hand‑off directly. Suitable for high‑throughput scenarios where each task needs its own thread.

DelayQueue – delayed execution. Suitable for scheduled tasks.

Source Analysis of execute()

public void execute(Runnable command) {
    if (command == null) throw new NullPointerException();
    int c = ctl.get();
    // 1. workerCount < corePoolSize → create core thread
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true)) return;
        c = ctl.get();
    }
    // 2. pool running → try to enqueue
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (!isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    } else {
        // 3. queue full → try to create non‑core thread
        if (!addWorker(command, false))
            reject(command); // apply rejection policy
    }
}

addWorker Core Logic

private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get();
        int rs = runStateOf(c);
        // check pool state
        if (rs >= SHUTDOWN && !(rs == SHUTDOWN && firstTask == null && !workQueue.isEmpty()))
            return false;
        for (;;) {
            int wc = workerCountOf(c);
            if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get();
            if (runStateOf(c) != rs) continue retry;
        }
    }
    // create and start Worker
    Worker w = null;
    boolean workerStarted = false, workerAdded = false;
    try {
        w = new Worker(firstTask);
        Thread t = w.thread;
        if (t != null) {
            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                int rs = runStateOf(ctl.get());
                if (rs < SHUTDOWN || (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive())
                        throw new IllegalThreadStateException();
                    workers.add(w);
                    int s = workers.size();
                    if (s > largestPoolSize) largestPoolSize = s;
                    workerAdded = true;
                }
            } finally { mainLock.unlock(); }
            if (workerAdded) {
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (!workerStarted) addWorkerFailed(w);
    }
    return workerStarted;
}

Worker Design

private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
    final Thread thread;          // executing thread
    Runnable firstTask;           // first task (may be null)
    volatile long completedTasks; // number of completed tasks
    Worker(Runnable firstTask) {
        setState(-1); // prevent interruption until runWorker
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }
    public void run() { runWorker(this); }
}

runWorker Core Loop

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interruption
    boolean completedAbruptly = true;
    try {
        while (task != null || (task = getTask()) != null) {
            w.lock();
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                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);
    }
}

Rejection Policies

AbortPolicy (default) – throws RejectedExecutionException. Use when the caller must handle rejected tasks explicitly.

CallerRunsPolicy – the calling thread runs the task. Use as a graceful degradation to avoid task loss.

DiscardPolicy – silently discards the task. Acceptable when loss is tolerable.

DiscardOldestPolicy – discards the oldest queued task and retries submission. Use when the newest data is preferred.

Common Thread‑Pool Variants

// FixedThreadPool: fixed number of threads (core = max, unbounded queue)
// Risk: task backlog may cause OOM
Executors.newFixedThreadPool(10);

// CachedThreadPool: elastic pool (core = 0, max = Integer.MAX_VALUE, SynchronousQueue)
// Risk: burst traffic may create many threads
Executors.newCachedThreadPool();

// SingleThreadExecutor: single thread (core = max = 1, unbounded queue)
// Guarantees task order
Executors.newSingleThreadExecutor();

// ScheduledThreadPool: supports delayed and periodic execution
Executors.newScheduledThreadPool(5);

Interview‑Ready Answer for execute() Flow

Core thread priority : if current thread count < corePoolSize, create a new core thread. Queue caching : otherwise try to enqueue the task. Expansion : if the queue is full and thread count < maximumPoolSize, create a non‑core thread. Rejection : if the queue is full and thread count ≥ maximumPoolSize, apply the configured rejection policy. The whole process is coordinated by the ctl field (CAS updates) to guarantee atomicity, addWorker creates threads, and runWorker repeatedly pulls tasks from the queue for execution.