ThreadPoolExecutor is the foundational class for thread pools in Java's concurrency utilities. It efficiently manages thread lifecycle and task execution based on configurable parameters. The constructor defines its core behavior:

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

Task submission follows a specific sequence: core threads, then the work queue, then non-core threads up to the maximum, and finally the rejection handler.

public void execute(Runnable job) {
    if (job == null) throw new NullPointerException();
    int currentState = controller.get();
    // Try to use a core thread first
    if (getActiveThreads(currentState) < corePoolSize) {
        if (createWorker(job, true)) return;
        currentState = controller.get();
    }
    // If pool is running, try to enqueue the task
    if (isOperational(currentState) && taskQueue.offer(job)) {
        int recheckedState = controller.get();
        if (!isOperational(recheckedState) && taskQueue.remove(job)) {
            handleRejection(job);
        } else if (getActiveThreads(recheckedState) == 0) {
            createWorker(null, false);
        }
    }
    // If queue is full, try to create a non-core thread
    else if (!createWorker(job, false)) {
        handleRejection(job); // Max threads reached, reject
    }
}

The controller variable is an AtomicInteger that combines the pool's run state and the current worker count using bitwise operations for efficient, lock-free updates.

private final AtomicInteger controller = new AtomicInteger(combineStateAndCount(RUNNING, 0));
private static final int THREAD_COUNT_BITS = 29;
private static final int MAX_THREAD_COUNT = (1 << THREAD_COUNT_BITS) - 1;

// Pool states
private static final int RUNNING    = -1 << THREAD_COUNT_BITS;
private static final int SHUTDOWN   =  0 << THREAD_COUNT_BITS;
private static final int STOP       =  1 << THREAD_COUNT_BITS;
private static final int TIDYING    =  2 << THREAD_COUNT_BITS;
private static final int TERMINATED =  3 << THREAD_COUNT_BITS;

// Helper methods
private static int extractState(int c)      { return c & ~MAX_THREAD_COUNT; }
private static int extractThreadCount(int c) { return c & MAX_THREAD_COUNT; }
private static int combineStateAndCount(int rs, int wc) { return rs | wc; }

The createWorker method is responsible for safely incrementing the worker count and instantiating a new worker thread.

private boolean createWorker(Runnable initialJob, boolean isCore) {
    // ... loops for CAS to increment worker count ...
    Worker newWorker = null;
    try {
        newWorker = new Worker(initialJob);
        final Thread t = newWorker.assignedThread;
        if (t != null) {
            final ReentrantLock poolLock = this.primaryLock;
            poolLock.lock();
            try {
                // Re-check state and add worker to set
                workerSet.add(newWorker);
                // ... update pool size tracking ...
            } finally {
                poolLock.unlock();
            }
            t.start(); // Start the thread
            return true;
        }
    } finally {
        if (!workerStarted) {
            rollbackWorkerCreation(newWorker);
        }
    }
    return false;
}

Each worker, an inner class, ecnapsulates a thread and executes tasks in a loop.

private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
    final Thread assignedThread;
    Runnable firstJob;
    Worker(Runnable firstJob) {
        setState(-1);
        this.firstJob = firstJob;
        this.assignedThread = getThreadFactory().newThread(this);
    }
    public void run() {
        performWork(this);
    }
}

final void performWork(Worker w) {
    Thread currentThread = Thread.currentThread();
    Runnable job = w.firstJob;
    w.firstJob = null;
    w.unlock();
    boolean abnormalExit = true;
    try {
        while (job != null || (job = fetchTask()) != null) {
            w.lock();
            // Check if thread should be interrupted
            if ((getPoolState(controller.get()) >= STOP ||
                 (Thread.interrupted() && getPoolState(controller.get()) >= STOP)) &&
                !currentThread.isInterrupted()) {
                currentThread.interrupt();
            }
            try {
                preExecute(currentThread, job);
                Throwable error = null;
                try {
                    job.run(); // Execute the actual task
                } catch (Throwable t) {
                    error = t; throw t;
                } finally {
                    postExecute(job, error);
                }
            } finally {
                job = null;
                w.tasksCompleted++;
                w.unlock();
            }
        }
        abnormalExit = false;
    } finally {
        cleanupWorker(w, abnormalExit);
    }
}

The fetchTask method is central to thread reuse, determining how threads retrieve work from the queue.

private Runnable fetchTask() {
    boolean pollTimedOut = false;
    for (;;) {
        int c = controller.get();
        int state = extractState(c);
        // Check if pool is shutting down
        if (state >= SHUTDOWN && (state >= STOP || taskQueue.isEmpty())) {
            decrementWorkerCount();
            return null;
        }
        int workerNum = extractThreadCount(c);
        // Determine if this thread should time out
        boolean canTimeOut = allowCoreThreadTimeOut || workerNum > corePoolSize;
        if ((workerNum > maximumPoolSize || (canTimeOut && pollTimedOut))
            && (workerNum > 1 || taskQueue.isEmpty())) {
            if (compareAndDecrementWorkerCount(c)) return null;
            continue;
        }
        try {
            // Core threads block indefinitely; others poll with timeout
            Runnable r = canTimeOut ?
                taskQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                taskQueue.take();
            if (r != null) return r;
            pollTimedOut = true;
        } catch (InterruptedException retry) {
            pollTimedOut = false;
        }
    }
}

When a task cannot be accepted, the RejectedExecutionHandler is invoked. Built-in policies include:

• AbortPolicy: Throws a RejectedExecutionException.
• CallerRunsPolicy: Executes the task in the caller's thread.
• DiscardPolicy: Silently discards the task.
• DiscardOldestPolicy: Discards the oldest queued task and retries submission.

The pool transitions through states: RUNNING -> SHUTDOWN/STOP -> TIDYING -> TERMINATED. The shutdown() method initiates an order shutdown, while shutdownNow() attempts to stop all executing tasks immediately.

import java.util.concurrent.*;

public class PoolDemo {
    public static void main(String[] args) {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
            2, 5, 10, TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(3),
            new ThreadPoolExecutor.AbortPolicy()
        );
        // Submit tasks to demonstrate the execution flow
        for (int i = 1; i <= 8; i++) {
            int id = i;
            pool.execute(() -> {
                try {
                    System.out.println("Task " + id + " executed by " + Thread.currentThread().getName());
                    Thread.sleep(1000);
                } catch (InterruptedException e) { /* handle */ }
            });
        }
        // This 9th task will be rejected
        try {
            pool.execute(() -> System.out.println("Task 9"));
        } catch (RejectedExecutionException e) {
            System.out.println("Task rejected: " + e.getMessage());
        }
        pool.shutdown();
    }
}