ThreadPoolExecutor: Custom Thread Pool Configuration

// Executors factory methods and their hidden pitfalls:

// newFixedThreadPool(n) — n threads, UNBOUNDED queue
ExecutorService fixed = Executors.newFixedThreadPool(10);
// Risk: tasks pile up in queue indefinitely → OutOfMemoryError under load

// newCachedThreadPool() — UNLIMITED threads, SynchronousQueue
ExecutorService cached = Executors.newCachedThreadPool();
// Risk: creates thousands of threads under burst load → OOM or context-switch storm

// newSingleThreadExecutor() — 1 thread, UNBOUNDED queue
ExecutorService single = Executors.newSingleThreadExecutor();
// Risk: same unbounded queue issue as fixed pool

// ThreadPoolExecutor gives you full control over EVERY parameter:
ThreadPoolExecutor pool = new ThreadPoolExecutor(
    4,                              // corePoolSize
    8,                              // maximumPoolSize
    60L, TimeUnit.SECONDS,          // keepAliveTime for idle threads above core
    new ArrayBlockingQueue<>(200),  // BOUNDED queue — tasks rejected when full
    new ThreadFactory() {           // custom thread factory
        private int n = 0;
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r, "worker-" + n++);
            t.setDaemon(false);
            return t;
        }
    },
    new ThreadPoolExecutor.CallerRunsPolicy() // backpressure policy
);

// Rule of thumb for sizing:
// CPU-bound tasks: corePoolSize = maxPoolSize = number of CPU cores
// I/O-bound tasks: corePoolSize = 2 * cores, maxPoolSize = 4 * cores (tune empirically)
int cores = Runtime.getRuntime().availableProcessors();
System.out.println("Available CPUs: " + cores);

// Full constructor signature:
// ThreadPoolExecutor(int corePoolSize,
//                    int maximumPoolSize,
//                    long keepAliveTime,
//                    TimeUnit unit,
//                    BlockingQueue<Runnable> workQueue,
//                    ThreadFactory threadFactory,
//                    RejectedExecutionHandler handler)

// corePoolSize — threads kept alive even when idle
// - new tasks go to idle core threads first
// - if all core threads busy AND queue not full → task goes to queue
// - if queue full AND thread count < max → spawn extra thread
// - if queue full AND thread count == max → rejection policy fires

// maximumPoolSize — max threads (including core threads)
// - extra threads (above core) are reclaimed after keepAliveTime idle

// keepAliveTime + unit — how long idle non-core threads survive
// - call allowCoreThreadTimeOut(true) to also reclaim idle CORE threads

// workQueue — buffered tasks waiting for a thread
// - determines how tasks queue when all core threads are busy

// threadFactory — creates new threads (name, priority, daemon flag, UncaughtExceptionHandler)
ThreadFactory namedFactory = r -> {
    Thread t = new Thread(r);
    t.setName("order-processor-" + System.nanoTime());
    t.setDaemon(true); // JVM exits even if these threads are running
    t.setUncaughtExceptionHandler((thread, ex) ->
        System.err.println(thread.getName() + " threw: " + ex));
    return t;
};

// Typical CPU-bound pool:
int n = Runtime.getRuntime().availableProcessors();
ThreadPoolExecutor cpuPool = new ThreadPoolExecutor(
    n, n,               // core == max — no dynamic growth
    0L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(1000), // bounded buffer
    namedFactory,
    new ThreadPoolExecutor.AbortPolicy()
);

// Typical I/O-bound pool:
ThreadPoolExecutor ioPool = new ThreadPoolExecutor(
    n * 2, n * 4,       // core threads, burst up to 4x cores
    30L, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(500),
    namedFactory,
    new ThreadPoolExecutor.CallerRunsPolicy()
);

// Dynamic resizing at runtime (rare but possible):
cpuPool.setCorePoolSize(n + 2);
cpuPool.setMaximumPoolSize(n + 4);
cpuPool.setKeepAliveTime(120, TimeUnit.SECONDS);

import java.util.concurrent.*;

// 1. LinkedBlockingQueue — optionally bounded linked list
BlockingQueue<Runnable> unbounded = new LinkedBlockingQueue<>();       // Integer.MAX_VALUE capacity
BlockingQueue<Runnable> bounded   = new LinkedBlockingQueue<>(1000);   // bounded — preferred!
// Behaviour in TPE: core threads fill first, then queue, then extra threads (up to max)
// With unbounded: maximumPoolSize is NEVER reached (queue absorbs everything)

// 2. ArrayBlockingQueue — fixed-size array, one lock, fair/nonfair
BlockingQueue<Runnable> arr = new ArrayBlockingQueue<>(200);          // nonfair (default)
BlockingQueue<Runnable> fair = new ArrayBlockingQueue<>(200, true);   // fair — FIFO ordering
// Slightly higher memory efficiency than LinkedBlockingQueue (no node objects)
// Use when queue size is known upfront and you want bounded allocation

// 3. SynchronousQueue — no storage; each submit needs a waiting thread
BlockingQueue<Runnable> sync = new SynchronousQueue<>();
// With SynchronousQueue + unbounded maxPoolSize → Executors.newCachedThreadPool() behaviour
// Each submitted task MUST be handed directly to a thread — no buffering at all
// Good when latency matters more than throughput and tasks are short-lived

// 4. PriorityBlockingQueue — heap-ordered by task priority
PriorityBlockingQueue<Runnable> pq = new PriorityBlockingQueue<>();
// Tasks must implement Comparable or you must supply a Comparator
class PriorityTask implements Runnable, Comparable<PriorityTask> {
    final int priority;
    final Runnable work;
    PriorityTask(int p, Runnable r) { this.priority = p; this.work = r; }
    public void run() { work.run(); }
    // high first
    public int compareTo(PriorityTask o) { return Integer.compare(o.priority, priority); }
}

// 5. DelayQueue — tasks become available only after their delay expires
// Used in scheduled executors — not typical for custom TPE

// Rejection fires when: pool is at maximumPoolSize AND queue is full
// Four built-in RejectedExecutionHandler implementations:

// 1. AbortPolicy (DEFAULT) — throws RejectedExecutionException
ThreadPoolExecutor p1 = new ThreadPoolExecutor(2, 4, 60, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(10), new ThreadPoolExecutor.AbortPolicy());
try {
    p1.submit(() -> doWork());
} catch (RejectedExecutionException e) {
    // Handle: log, return error, drop task, etc.
    System.err.println("Task rejected: " + e.getMessage());
}

// 2. DiscardPolicy — silently drops the new task (no exception!)
ThreadPoolExecutor p2 = new ThreadPoolExecutor(2, 4, 60, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(10), new ThreadPoolExecutor.DiscardPolicy());
// Danger: you lose tasks silently — use only for best-effort/fire-and-forget work

// 3. DiscardOldestPolicy — drops the OLDEST waiting task and retries submit
ThreadPoolExecutor p3 = new ThreadPoolExecutor(2, 4, 60, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(10), new ThreadPoolExecutor.DiscardOldestPolicy());
// Makes sense when recent data is more valuable than old (sensor readings, events)

// 4. CallerRunsPolicy — submitting thread runs the task itself (see section below)
ThreadPoolExecutor p4 = new ThreadPoolExecutor(2, 4, 60, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(10), new ThreadPoolExecutor.CallerRunsPolicy());

// 5. Custom rejection policy — implement RejectedExecutionHandler
class LogAndQueuePolicy implements RejectedExecutionHandler {
    private final BlockingQueue<Runnable> overflow;

    LogAndQueuePolicy(BlockingQueue<Runnable> overflow) {
        this.overflow = overflow;
    }

    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        if (executor.isShutdown()) return; // pool shut down — discard
        System.err.println("[WARN] Pool saturated — task queued to overflow buffer");
        if (!overflow.offer(r)) {
            System.err.println("[ERROR] Overflow buffer also full — task dropped!");
        }
    }
}

BlockingQueue<Runnable> overflowBuf = new LinkedBlockingQueue<>(5000);
ThreadPoolExecutor p5 = new ThreadPoolExecutor(2, 4, 60, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(200), new LogAndQueuePolicy(overflowBuf));

// ThreadPoolExecutor exposes real-time metrics — no JMX needed

ThreadPoolExecutor pool = new ThreadPoolExecutor(4, 8, 60, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(200));

// Thread counts:
int active    = pool.getActiveCount();       // threads currently running tasks
int poolSize  = pool.getPoolSize();          // current number of threads (may be < core)
int coreSize  = pool.getCorePoolSize();      // configured core threads
int maxSize   = pool.getMaximumPoolSize();   // configured max threads
int largest   = pool.getLargestPoolSize();   // peak thread count since pool creation

// Task counts:
long submitted  = pool.getTaskCount();          // tasks submitted (approx)
long completed  = pool.getCompletedTaskCount(); // tasks finished
int queued      = pool.getQueue().size();       // tasks waiting in queue
int remaining   = pool.getQueue().remainingCapacity(); // space left in queue

// Compute pending (submitted but not yet completed):
long pending = pool.getTaskCount() - pool.getCompletedTaskCount();

System.out.printf("Threads: %d active / %d pool / %d max%n", active, poolSize, maxSize);
System.out.printf("Queue: %d waiting / %d remaining%n", queued, remaining);
System.out.printf("Tasks: %d completed / %d pending%n", completed, pending);

// Periodic monitoring thread:
ScheduledExecutorService monitor = Executors.newSingleThreadScheduledExecutor();
monitor.scheduleAtFixedRate(() -> {
    System.out.printf("[POOL] active=%d pool=%d queue=%d completed=%d%n",
        pool.getActiveCount(), pool.getPoolSize(),
        pool.getQueue().size(), pool.getCompletedTaskCount());
}, 0, 5, TimeUnit.SECONDS);

// Hook into task lifecycle by subclassing:
class InstrumentedPool extends ThreadPoolExecutor {
    InstrumentedPool(int core, int max, BlockingQueue<Runnable> q) {
        super(core, max, 60, TimeUnit.SECONDS, q);
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        System.out.println(t.getName() + " starting task");
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        if (t != null) System.err.println("Task threw: " + t);
    }

    @Override
    protected void terminated() {
        super.terminated();
        System.out.println("Pool fully shut down");
    }
}

// Two shutdown modes:

// shutdown() — orderly shutdown
// - Rejects NEW tasks immediately
// - Allows already-queued and currently-running tasks to complete
pool.shutdown();

// shutdownNow() — immediate shutdown
// - Attempts to cancel all queued tasks (removes them from queue)
// - Sends interrupt to running threads (they must handle it!)
// - Returns list of tasks that were queued but never started
List<Runnable> unstarted = pool.shutdownNow();
System.out.println("Discarded " + unstarted.size() + " queued tasks");

// Waiting for completion after shutdown():
pool.shutdown();
try {
    if (!pool.awaitTermination(30, TimeUnit.SECONDS)) {
        // Timeout — force stop
        pool.shutdownNow();
        if (!pool.awaitTermination(10, TimeUnit.SECONDS)) {
            System.err.println("Pool did not terminate");
        }
    }
} catch (InterruptedException e) {
    pool.shutdownNow();
    Thread.currentThread().interrupt();
}

// Registration as JVM shutdown hook for server applications:
class ManagedPool {
    private final ThreadPoolExecutor pool = new ThreadPoolExecutor(
        4, 8, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(500));

    ManagedPool() {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.out.println("Shutdown hook: draining pool...");
            pool.shutdown();
            try {
                pool.awaitTermination(15, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                pool.shutdownNow();
            }
            System.out.println("Pool drained.");
        }, "pool-shutdown-hook"));
    }

    Future<?> submit(Runnable r) { return pool.submit(r); }
}

// State query:
pool.isShutdown();      // true after shutdown() or shutdownNow()
pool.isTerminating();   // true between shutdown() and full termination
pool.isTerminated();    // true once all tasks have finished after shutdown

// CallerRunsPolicy — when the pool is saturated, the SUBMITTING thread
// runs the task itself instead of queuing it or throwing.

// Effect: submitting thread is blocked doing work → it submits fewer new tasks
// → natural backpressure: the system slows down instead of crashing

// Example: web server processing requests
ThreadPoolExecutor handler = new ThreadPoolExecutor(
    8, 16, 60, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(100),
    Executors.defaultThreadFactory(),
    new ThreadPoolExecutor.CallerRunsPolicy() // <--- backpressure
);

// Simulated request handler:
void handleRequest(HttpRequest request) {
    handler.submit(() -> {
        processRequest(request); // normally runs in pool thread
    });
    // If pool is saturated:
    //   CallerRunsPolicy → THIS thread (the HTTP acceptor) runs processRequest()
    //   HTTP acceptor is busy → accepts no new connections → backpressure to client
    //   Client sees slower responses → backs off naturally
}

// Compare with AbortPolicy:
//   Pool saturated → RejectedExecutionException
//   Server must catch it and return HTTP 503
//   Client must implement retry logic
//   Code is more complex; client must understand overload signal

// CallerRunsPolicy pitfall — Netty / async I/O event loop:
// If your submitting thread is a non-blocking event loop, CallerRunsPolicy
// will block it while running the task → all I/O on that loop stalls.
// In this case: prefer AbortPolicy + return 503 from the async handler.

// Custom CallerRunsPolicy with logging:
class LoggingCallerRunsPolicy implements RejectedExecutionHandler {
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        if (!executor.isShutdown()) {
            System.err.println("[BACKPRESSURE] Pool saturated; caller running task: "
                + Thread.currentThread().getName());
            long start = System.nanoTime();
            r.run();
            long ms = (System.nanoTime() - start) / 1_000_000;
            System.err.println("[BACKPRESSURE] Caller-run task finished in " + ms + " ms");
        }
    }
}