Java Executor Framework: From Executor to ForkJoinPool

public interface Executor {
    /**
     * Executes the given command at some time in the future.
     * The command may execute in a new thread, in a pooled thread,
     * or in the calling thread, at the discretion of the Executor.
     */
    void execute(Runnable command);
}

// Usage — your code doesn't care HOW tasks run
public void processOrders(Executor executor, List<Order> orders) {
    orders.forEach(order ->
        executor.execute(() -> processOrder(order))
    );
}

// Caller decides the strategy — swap without changing business logic
processOrders(Executors.newFixedThreadPool(4), orders); // pool
processOrders(Runnable::run, orders);                        // inline (sync)
processOrders(r -> new Thread(r).start(), orders);      // new thread each

ExecutorService es = Executors.newFixedThreadPool(4);

// 1. submit(Callable) → Future
Future<String> future = es.submit(() -> fetchUserFromDb(userId));
try {
    String user = future.get(5, TimeUnit.SECONDS); // blocks up to 5s
} catch (TimeoutException e) {
    future.cancel(true); // interrupt the worker thread
}

// 2. invokeAll — parallel fan-out, wait for all
List<Callable<Report>> tasks = departments.stream()
    .map(d -> (Callable<Report>) () -> generateReport(d))
    .toList();
List<Future<Report>> results = es.invokeAll(tasks); // blocks until all done

// 3. invokeAny — race: first success wins
String fastest = es.invokeAny(List.of(
    () -> queryPrimary(),
    () -> queryReplica1(),
    () -> queryReplica2()
)); // whichever DB responds first; others cancelled

// 4. Proper shutdown — always in a finally / try-with-resources
es.shutdown(); // stop accepting new tasks
if (!es.awaitTermination(30, TimeUnit.SECONDS)) {
    es.shutdownNow(); // force-stop remaining tasks
}

ScheduledExecutorService scheduler =
    Executors.newScheduledThreadPool(2);

// Run once after 5 seconds
ScheduledFuture<?> future = scheduler.schedule(
    () -> sendReminderEmail(userId),
    5, TimeUnit.SECONDS
);

// Poll every 10s (fixed rate — wall-clock aligned)
scheduler.scheduleAtFixedRate(
    () -> syncInventoryFromWarehouse(),
    0,     // initial delay
    10,    // period
    TimeUnit.SECONDS
);

// Health-check: wait 30s AFTER each check completes
scheduler.scheduleWithFixedDelay(
    () -> runHealthCheck(),
    0, 30, TimeUnit.SECONDS
);

// Cancel a scheduled task
future.cancel(false); // false = don't interrupt if already running

ThreadPoolExecutor pool = new ThreadPoolExecutor(
    4,                                      // corePoolSize
    8,                                      // maximumPoolSize
    60L, TimeUnit.SECONDS,               // keepAliveTime for non-core
    new ArrayBlockingQueue<>(100),       // bounded queue — back pressure!
    new ThreadFactoryBuilder()             // Guava helper or custom
        .setNameFormat("order-worker-%d")
        .setDaemon(false)
        .build(),
    new ThreadPoolExecutor.CallerRunsPolicy() // slow down caller on overflow
);

// Monitoring hooks (override in subclass)
pool = new ThreadPoolExecutor(...) {
    protected void beforeExecute(Thread t, Runnable r) {
        MDC.put("threadName", t.getName()); // add to MDC for logging
    }
    protected void afterExecute(Runnable r, Throwable ex) {
        MDC.clear();
        if (ex != null) metrics.increment("task.failed");
    }
};

// Factory shortcuts and what they produce under the hood
Executors.newFixedThreadPool(n)
    // → TPE(n, n, 0, LinkedBlockingQueue UNBOUNDED)  ← queue grows forever!

Executors.newCachedThreadPool()
    // → TPE(0, MAX_INT, 60s, SynchronousQueue) ← can spawn millions of threads!

// ⚠ Always use explicit TPE constructor in production, never naked factories

// Prefer direct construction over Executors.newScheduledThreadPool()
// so you can control rejection policy and thread factory
ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(
    2,                              // corePoolSize (no max needed — DelayQueue is unbounded)
    new ThreadFactoryBuilder()
        .setNameFormat("scheduler-%d")
        .setDaemon(true)           // daemon = JVM can exit without waiting
        .build()
);

// Important tuning flags
scheduler.setRemoveOnCancelPolicy(true); // GC cancelled tasks from queue immediately
scheduler.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);

// Periodic cache refresh every minute
scheduler.scheduleAtFixedRate(
    cacheService::refresh, 0, 1, TimeUnit.MINUTES
);

// One-off delayed notification
ScheduledFuture<?> f = scheduler.schedule(
    () -> notificationService.sendPaymentReminder(invoiceId),
    48, TimeUnit.HOURS
);
// Cancel if payment received
paymentReceivedEvent.addListener(() -> f.cancel(false));

// RecursiveTask — fork/join divide-and-conquer
class ParallelSum extends RecursiveTask<Long> {
    private static final int THRESHOLD = 10_000;
    private final long[] array;
    private final int start, end;

    protected Long compute() {
        if (end - start <= THRESHOLD) {
            return sumDirectly(array, start, end); // base case
        }
        int mid = (start + end) / 2;
        ParallelSum left  = new ParallelSum(array, start, mid);
        ParallelSum right = new ParallelSum(array, mid, end);
        left.fork();              // submit left to pool asynchronously
        long rightResult = right.compute(); // run right in current thread
        long leftResult  = left.join();     // wait for left
        return leftResult + rightResult;
    }
}

// Execute
ForkJoinPool pool = new ForkJoinPool(
    Runtime.getRuntime().availableProcessors()
);
long total = pool.invoke(new ParallelSum(data, 0, data.length));

// ForkJoinPool.commonPool() is used by parallel streams automatically
long sum = Arrays.stream(data).parallel().sum(); // uses commonPool

// Custom pool for parallel streams (control parallelism level)
pool.submit(() ->
    orders.parallelStream()
          .filter(Order::isHighValue)
          .map(this::enrich)
          .collect(Collectors.toList())
).join();

// 1. Configuration
@Configuration
@EnableAsync
public class AsyncConfig implements AsyncConfigurer {

    @Bean(name = "taskExecutor")
    public ThreadPoolTaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor exec = new ThreadPoolTaskExecutor();
        exec.setCorePoolSize(4);
        exec.setMaxPoolSize(12);
        exec.setQueueCapacity(200);
        exec.setThreadNamePrefix("myApp-task-");
        exec.setKeepAliveSeconds(60);
        exec.setRejectedExecutionHandler(
            new ThreadPoolExecutor.CallerRunsPolicy()
        );
        exec.setWaitForTasksToCompleteOnShutdown(true);  // graceful stop
        exec.setAwaitTerminationSeconds(30);
        exec.initialize();
        return exec;
    }

    // Optional: set default executor for @Async without qualifier
    public Executor getAsyncExecutor() { return taskExecutor(); }
}

// 2. Usage — annotate service methods
@Service
public class NotificationService {

    @Async("taskExecutor")
    public CompletableFuture<Void> sendEmail(String to, String body) {
        emailClient.send(to, body); // runs in pool thread
        return CompletableFuture.completedFuture(null);
    }

    @Async("taskExecutor")
    public CompletableFuture<Report> generateReport(ReportRequest req) {
        Report r = heavyReportService.build(req);
        return CompletableFuture.completedFuture(r);
    }
}

// 3. Caller — non-blocking
@RestController
public class OrderController {
    public ResponseEntity<?> createOrder(OrderRequest req) {
        Order order = orderService.create(req);
        notificationService.sendEmail(req.getEmail(), ...); // fire and forget
        return ResponseEntity.ok(order); // returns immediately
    }
}