// WHAT YOU USED TO DO: a thread per task, complete manual management.
Thread[] threads = new Thread[200];
for (int i = 0; i < 200; i++) {
    threads[i] = new Thread(new NoticeTask(notices.get(i)), "bibliotech-notices-" + i);
    threads[i].start();
}
for (Thread t : threads) {
    t.join();
}
// And each task's result? volatile fields read by hand.
// And if one fails? An uncaught exception handler.
// And if I want to cancel everything? interrupt() on all 200, one by one.
// And if there are 200,000 notices? OutOfMemoryError.

// WHAT YOU WILL DO: describe tasks, the executor places them.
ExecutorService executor = Executors.newFixedThreadPool(8);
List<Future<NoticeResult>> futures = new ArrayList<>();

for (Notice n : notices) {
    futures.add(executor.submit(new NoticeTask(n)));   // Callable: returns a value
}

for (Future<NoticeResult> f : futures) {
    NoticeResult r = f.get();   // waits, and RETHROWS the exception if there was one
}

executor.shutdown();

// The complete interface. A single method.
public interface Executor {
    void execute(Runnable command);
}

// Three valid implementations of the SAME interface:
Executor onTheSameThread = task -> task.run();
Executor aNewThread      = task -> new Thread(task).start();
Executor aPool           = Executors.newFixedThreadPool(8);

// The client code does not change:
onTheSameThread.execute(() -> System.out.println("hello"));
aPool.execute(() -> System.out.println("hello"));

public interface ExecutorService extends Executor {

    // Submit tasks and get a Future
    Future<?> submit(Runnable task);
    <T> Future<T> submit(Callable<T> task);
    <T> Future<T> submit(Runnable task, T result);

    // Submit several
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks);
    <T> T invokeAny(Collection<? extends Callable<T>> tasks);

    // Shutdown
    void shutdown();
    List<Runnable> shutdownNow();
    boolean isShutdown();
    boolean isTerminated();
    boolean awaitTermination(long time, TimeUnit unit);
}

// execute(Runnable): returns nothing. If the task throws an exception,
// it goes to the thread's uncaught exception handler (08-02) and
// is normally printed on the console.
executor.execute(() -> { throw new RuntimeException("visible failure"); });

// submit(...): returns a Future. If the task throws an exception,
// it is STORED in the Future and is NOT printed anywhere.
// If nobody calls get(), the failure vanishes WITHOUT A TRACE.
executor.submit(() -> { throw new RuntimeException("INVISIBLE failure"); });

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

// 1. FIXED pool: N threads, always the same ones, unbounded queue.
ExecutorService fixed = Executors.newFixedThreadPool(8);

// 2. ELASTIC pool: creates threads on demand, recycles them after 60 s.
ExecutorService elastic = Executors.newCachedThreadPool();

// 3. A SINGLE THREAD: guarantees execution order.
ExecutorService single = Executors.newSingleThreadExecutor();

// 4. SCHEDULED: delayed and periodic tasks.
ScheduledExecutorService scheduled = Executors.newScheduledThreadPool(2);

// 5. WORK STEALING: a ForkJoinPool with parallelism = cores.
ExecutorService stealing = Executors.newWorkStealingPool();

// Sending notices: I/O, bounded and known load -> fixed pool.
ExecutorService notices = Executors.newFixedThreadPool(8);

// Writing the audit log: it MUST preserve ordering -> a single thread.
// Also, being a single thread, the file needs no synchronisation.
ExecutorService audit = Executors.newSingleThreadExecutor();

// Due-date notices every hour and a backup every night.
ScheduledExecutorService maintenance = Executors.newScheduledThreadPool(2);

// TIME BOMB
ExecutorService pool = Executors.newFixedThreadPool(4);

// Each task takes 100 ms; 1,000 arrive per second.
// The pool processes 40/s. 960 tasks pile up in memory every second.
// In 10 minutes: ~576,000 objects queued. OutOfMemoryError.
while (thereAreRequests()) {
    pool.submit(new SlowTask(nextRequest()));
}

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class BiblioTechExecutor {

    /**
     * A ThreadFactory that names the threads. Without it, the pool's threads
     * are called "pool-1-thread-1", which in a dump (08-03) says nothing.
     */
    static class NamedThreadFactory implements ThreadFactory {
        private final String prefix;
        private final boolean daemon;
        private final AtomicInteger counter = new AtomicInteger(1);

        NamedThreadFactory(String prefix, boolean daemon) {
            this.prefix = prefix;
            this.daemon = daemon;
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r, prefix + "-" + counter.getAndIncrement());
            t.setDaemon(daemon);
            // Safety net: any exception escaping a task submitted
            // with execute() gets logged (06-07, 08-02).
            t.setUncaughtExceptionHandler((thread, error) ->
                    System.err.println("[" + thread.getName() + "] uncaught failure: " + error));
            return t;
        }
    }

    public static ThreadPoolExecutor create() {
        return new ThreadPoolExecutor(
                4,                                    // 1. core size
                8,                                    // 2. maximum size
                60L, TimeUnit.SECONDS,                // 3. lifetime of extra threads
                new ArrayBlockingQueue<>(100),        // 4. BOUNDED queue
                new NamedThreadFactory("bibliotech-notices", false),      // 5.
                new ThreadPoolExecutor.CallerRunsPolicy()                 // 6.
        );
    }
}

public class PoolSizing {

    private static final int CORES = Runtime.getRuntime().availableProcessors();

    /** Pure computation: more threads than cores only adds context switches. */
    public static int forCompute() {
        return CORES;
    }

    /**
     * I/O: Brian Goetz's formula.
     *   threads = cores * targetUtilisation * (1 + wait / compute)
     * With utilisation 1.0, 300 ms of waiting and 20 ms of computing on 8 cores:
     *   8 * 1.0 * (1 + 15) = 128 threads.
     * In practice you cap it at a sensible maximum: 128 threads are 128 MB
     * of stacks, and the remote resource probably cannot take 128 requests.
     */
    public static int forIo(double waitMs, double computeMs) {
        double ratio = waitMs / computeMs;
        int suggested = (int) Math.ceil(CORES * (1 + ratio));
        return Math.min(suggested, 64);      // sanity cap
    }

    public static void main(String[] args) {
        System.out.println("Cores               : " + CORES);
        System.out.println("Compute pool        : " + forCompute());
        System.out.println("Notices pool        : " + forIo(300, 20));
        System.out.println("Import pool         : " + forIo(50, 30));
    }
}

// GOOD: three pools with different purposes and sizes, isolated
// from each other: saturating the notices pool does not affect the compute one.
private final ExecutorService computePool = Executors.newFixedThreadPool(CORES);
private final ExecutorService ioPool      = Executors.newFixedThreadPool(32);
private final ScheduledExecutorService scheduledPool =
        Executors.newScheduledThreadPool(2);

import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

public final class GracefulShutdown {

    /**
     * Two-phase shutdown:
     *   1. shutdown() + bounded wait: let what is running finish.
     *   2. If they do not finish, shutdownNow() (interruption) + short wait.
     *   3. If they still do not finish, log it: there are tasks that do not
     *      respond to interruption, and that is a BUG in those tasks
     *      (the empty InterruptedException catch of 08-02).
     */
    public static void shutdown(ExecutorService executor, long seconds) {
        executor.shutdown();                     // phase 1: no more accepted
        try {
            if (!executor.awaitTermination(seconds, TimeUnit.SECONDS)) {
                executor.shutdownNow();          // phase 2: interrupt
                if (!executor.awaitTermination(seconds, TimeUnit.SECONDS)) {
                    System.err.println("The executor has not terminated: "
                            + "there are tasks ignoring the interruption");
                }
            }
        } catch (InterruptedException e) {
            // WE have been interrupted while waiting.
            executor.shutdownNow();
            Thread.currentThread().interrupt();  // restore the flag (08-02)
        }
    }
}

@FunctionalInterface
public interface Callable<V> {
    V call() throws Exception;      // returns V and CAN throw
}

public interface Future<V> {
    V get() throws InterruptedException, ExecutionException;
    V get(long time, TimeUnit unit) throws ..., TimeoutException;
    boolean cancel(boolean mayInterruptIfRunning);
    boolean isCancelled();
    boolean isDone();
}

package com.nexussoftware.bibliotech.persistence;

import java.nio.file.Path;
import java.util.concurrent.*;

public class ImportWithFuture {

    public static void main(String[] args) {

        ExecutorService executor = Executors.newFixedThreadPool(2);
        try {
            // Callable<Report>: returns a Report and can throw
            // checked exceptions. Both impossible with Runnable.
            Callable<Report> task = () -> {
                CatalogImporter imp = new CatalogImporter();
                return imp.importFrom(Path.of("data/inventory.csv"));  // throws IOException
            };

            Future<Report> future = executor.submit(task);

            // The main thread is NOT blocked: it can carry on working.
            System.out.println("[main] import launched, serving the menu");
            showMenu();

            // isDone() does not block: it lets you poll without waiting.
            while (!future.isDone()) {
                System.out.println("[main] importing...");
                TimeUnit.MILLISECONDS.sleep(500);
            }

            // get() with a DEADLINE: never use get() without one in production.
            Report report = future.get(30, TimeUnit.SECONDS);
            System.out.println("[main] imported: " + report.imported());

        } catch (TimeoutException e) {
            System.err.println("[main] the import exceeded the deadline");

        } catch (ExecutionException e) {
            // The task threw an exception. get() WRAPS it in
            // ExecutionException; the original is in getCause().
            // It is exactly the exception chaining of 06-03.
            Throwable cause = e.getCause();
            System.err.println("[main] the import failed: " + cause.getMessage());
            if (cause instanceof java.io.IOException) {
                System.err.println("[main] file problem; the previous catalogue is kept");
            }

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();

        } finally {
            GracefulShutdown.shutdown(executor, 10);
        }
    }

    static void showMenu() { /* ... */ }
}

try {
    Report r = future.get(30, TimeUnit.SECONDS);
} catch (ExecutionException e) {
    Throwable cause = e.getCause();
    // Recover the original type to apply the 06-07 policy
    if (cause instanceof InvalidFormatException ife) {
        logAndDegrade(ife);
    } else if (cause instanceof java.io.IOException ioe) {
        abortWithCode(ioe);
    } else {
        throw new BiblioTechException("Unexpected failure in the import", cause);
    }
}

Future<Report> future = executor.submit(task);

// cancel(false): only stops it STARTING if it is still queued.
// If it is already running, it lets it finish.
future.cancel(false);

// cancel(true): in addition, it INTERRUPTS the thread running it.
// It is literally a thread.interrupt() on the pool's thread.
future.cancel(true);

/**
 * A cancellable task inside a pool. Identical to the 08-02 pattern:
 * the Future's cancel(true) translates into interrupt() on this thread.
 */
public class CancellableNoticeTask implements Callable<NoticeResult> {

    private final Notice notice;

    public CancellableNoticeTask(Notice notice) { this.notice = notice; }

    @Override
    public NoticeResult call() throws Exception {
        for (int attempt = 1; attempt <= 3; attempt++) {

            // A cancellation point on every pass.
            if (Thread.currentThread().isInterrupted()) {
                throw new InterruptedException("notice cancelled before attempt " + attempt);
            }
            try {
                send(notice);                        // may throw InterruptedException
                return NoticeResult.success(notice);
            } catch (SendFailedException e) {
                if (attempt == 3) {
                    return NoticeResult.failure(notice, e.getMessage());
                }
                TimeUnit.MILLISECONDS.sleep(100L * attempt);   // backoff
            }
        }
        return NoticeResult.failure(notice, "retries exhausted");
    }

    private void send(Notice n) throws Exception { /* ... */ }
}

Future<Report> f = executor.submit(task);
TimeUnit.SECONDS.sleep(2);

boolean cancelled = f.cancel(true);
System.out.println("cancel() returned : " + cancelled);   // false if it had finished
System.out.println("isCancelled()     : " + f.isCancelled());
System.out.println("isDone()          : " + f.isDone());  // true: cancelled is also "done"

try {
    f.get();
} catch (CancellationException e) {
    // get() on a CANCELLED Future throws CancellationException,
    // which is UNchecked. It is easy to forget to catch it.
    System.out.println("confirmed: the task was cancelled");
}

List<Callable<NoticeResult>> tasks = new ArrayList<>();
for (Notice n : notices) {
    tasks.add(new CancellableNoticeTask(n));
}

// Blocks until they ALL finish. The returned Futures
// are already complete: get() does not block.
List<Future<NoticeResult>> results = executor.invokeAll(tasks);

int successes = 0, failures = 0;
for (Future<NoticeResult> f : results) {
    try {
        if (f.get().isSuccessful()) successes++; else failures++;
    } catch (ExecutionException e) {
        failures++;
        LOG.log(Level.WARNING, "notice failed", e.getCause());
    }
}

// A GLOBAL deadline for the whole set. Those that do not finish in time
// are CANCELLED automatically, and their get() will throw CancellationException.
List<Future<NoticeResult>> results =
        executor.invokeAll(tasks, 30, TimeUnit.SECONDS);

// Three sources for the same catalogue; the first to answer will do.
List<Callable<Catalog>> sources = List.of(
        () -> loadFromLocalCache(),
        () -> loadFromMainFile(),
        () -> loadFromBackup()
);

// Returns the first correct result; cancels the other two.
// If they ALL fail, it throws ExecutionException with the last cause.
Catalog c = executor.invokeAny(sources, 10, TimeUnit.SECONDS);

import java.util.concurrent.*;

public class BiblioTechMaintenance {

    private final ScheduledExecutorService scheduler =
            Executors.newScheduledThreadPool(2, r -> {
                Thread t = new Thread(r, "bibliotech-scheduler");
                t.setDaemon(true);      // maintenance: must not prevent shutdown
                return t;
            });

    public void start() {

        // 1. ONCE, in 5 seconds' time.
        scheduler.schedule(
                () -> System.out.println("[maintenance] start-up complete"),
                5, TimeUnit.SECONDS);

        // 2. PERIODIC at a fixed rate: every hour from the start instant.
        scheduler.scheduleAtFixedRate(
                this::sendDueNotices,
                0, 1, TimeUnit.HOURS);

        // 3. PERIODIC with a fixed delay: 24 h from when the previous one ENDS.
        scheduler.scheduleWithFixedDelay(
                this::backup,
                1, 24, TimeUnit.HOURS);
    }

    /**
     * GOLDEN RULE: a periodic task must catch EVERY exception.
     * If it lets one escape, the JVM CANCELS the schedule and the task
     * IS NEVER RUN AGAIN, silently, with no warning at all.
     * It is the most treacherous failure of this API.
     */
    private void sendDueNotices() {
        try {
            int sent = noticeService.sendDueNotices();
            LOG.log(Level.INFO, "Due-date notices sent: {0}", sent);
        } catch (Exception e) {
            LOG.log(Level.SEVERE, "Failure sending due-date notices", e);
            // Do NOT rethrow: if it escapes, the schedule ends forever.
        }
    }

    private void backup() {
        try {
            backupManager.backup();
        } catch (Exception e) {
            LOG.log(Level.SEVERE, "Backup failure", e);
        }
    }
}

A task lasting 3 s, period 5 s:

scheduleAtFixedRate:   [---3s---]__2s__[---3s---]__2s__[---3s---]
                       0         3     5         8     10
                       starts every 5 s exactly

scheduleWithFixedDelay:[---3s---]____5s____[---3s---]____5s____[--
                       0         3         8         11        16
                       5 s AFTER finishing

CountDownLatch latch = new CountDownLatch(5);   // 5 pending events

latch.countDown();          // subtracts 1 (never below 0)
latch.await();              // blocks until it reaches 0
latch.await(10, TimeUnit.SECONDS);   // with a deadline; returns false if it expires
latch.getCount();           // current count (for showing progress)

import java.util.concurrent.*;

public class AwaitNTasks {

    public static void main(String[] args) throws InterruptedException {

        final int TASKS = 200;
        ExecutorService pool = Executors.newFixedThreadPool(8);
        CountDownLatch finished = new CountDownLatch(TASKS);

        for (int i = 0; i < TASKS; i++) {
            final int n = i;
            pool.execute(() -> {
                try {
                    sendNotice(n);
                } catch (Exception e) {
                    LOG.log(Level.WARNING, "notice " + n + " failed", e);
                } finally {
                    // MANDATORY in finally: if a task fails and does not
                    // count down, main's await() NEVER returns.
                    finished.countDown();
                }
            });
        }

        // Progress while we wait, with no blind sleep polling.
        while (!finished.await(500, TimeUnit.MILLISECONDS)) {
            long pending = finished.getCount();
            System.out.printf("\rProgress: %d/%d (%.0f%%)",
                    TASKS - pending, TASKS,
                    100.0 * (TASKS - pending) / TASKS);
        }
        System.out.println("\nAll the notices processed");
        GracefulShutdown.shutdown(pool, 10);
    }
}

/**
 * Two latches: one to start everybody at once, another to
 * wait for everybody to finish. It is the skeleton of a properly
 * built concurrency stress test: without the gate, the first threads
 * would finish before the last ones started, and there would be no
 * real overlap (the effect you saw in exercise 2 of 08-01).
 */
public static long stressTest(int threads, Runnable task) throws InterruptedException {

    CountDownLatch startGate = new CountDownLatch(1);        // the starting pistol
    CountDownLatch finishLine = new CountDownLatch(threads); // the finish line

    for (int i = 0; i < threads; i++) {
        new Thread(() -> {
            try {
                startGate.await();   // everybody waits here
                task.run();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                finishLine.countDown();
            }
        }, "stress-" + i).start();
    }

    long start = System.nanoTime();
    startGate.countDown();           // go! the N start at once
    finishLine.await();
    return System.nanoTime() - start;
}

import java.util.concurrent.*;

public class PhasedProcessing {

    public static void main(String[] args) {

        final int WORKERS = 4;

        // The barrier action is run by the LAST thread to arrive,
        // before releasing the others. Ideal for consolidating the phase.
        CyclicBarrier barrier = new CyclicBarrier(WORKERS, () ->
                System.out.println("--- phase completed by all 4, consolidating ---"));

        for (int i = 0; i < WORKERS; i++) {
            final int id = i;
            new Thread(() -> {
                try {
                    for (int phase = 1; phase <= 3; phase++) {
                        System.out.printf("  worker %d processes phase %d%n", id, phase);
                        TimeUnit.MILLISECONDS.sleep(100 + id * 50);
                        barrier.await();      // waits for the other three
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } catch (BrokenBarrierException e) {
                    // If a thread is interrupted or times out, the barrier BREAKS
                    // and ALL the others receive this exception. That is correct:
                    // without every participant, the phase makes no sense.
                    System.err.println("broken barrier: " + e);
                }
            }, "worker-" + i).start();
        }
    }
}

package com.nexussoftware.bibliotech.service;

import java.util.concurrent.*;

/**
 * Limits simultaneous exports to 3.
 *
 * Reason: each export opens the whole catalogue file, formats it in
 * memory and writes it. Three at once saturate the disk and multiply
 * memory usage; with twenty, the server crawls.
 * The semaphore puts a hard cap without limiting the rest of the pool.
 */
public class ExportService {

    private static final int MAX_CONCURRENT = 3;

    // 3 permits. The 'true' enables fairness: those who have been
    // waiting longest go first, which prevents starvation (08-04).
    private final Semaphore permits = new Semaphore(MAX_CONCURRENT, true);

    private final ExecutorService pool = Executors.newFixedThreadPool(16);

    public Future<Path> export(String format) {
        return pool.submit(() -> {

            // Waits for a permit. If there are 3 exports in progress, this
            // thread blocks here until one finishes.
            permits.acquire();
            try {
                System.out.printf("[%s] exporting (%d permits free)%n",
                        Thread.currentThread().getName(),
                        permits.availablePermits());
                return generateFile(format);         // heavy work
            } finally {
                // MANDATORY in finally: a permit not returned is
                // lost FOREVER. After three failures, the semaphore
                // sits at 0 and nobody ever exports again.
                permits.release();
            }
        });
    }

    /** A variant that does not wait: it rejects instead of queuing. */
    public Future<Path> exportWithoutWaiting(String format) {
        return pool.submit(() -> {
            if (!permits.tryAcquire(5, TimeUnit.SECONDS)) {
                throw new ServiceOverloadedException(
                        "There are " + MAX_CONCURRENT + " exports in progress; retry");
            }
            try {
                return generateFile(format);
            } finally {
                permits.release();
            }
        });
    }

    private Path generateFile(String format) throws Exception {
        TimeUnit.SECONDS.sleep(2);
        return Path.of("reports/catalog." + format);
    }
}

Exchanger<List<Material>> exchange = new Exchanger<>();

// Reader thread: fills a buffer and swaps it for an empty one.
List<Material> myBuffer = new ArrayList<>();
// ... fill it ...
myBuffer = exchange.exchange(myBuffer);    // receives the other's empty buffer

// Processor thread: processes a full buffer and returns an empty one.
List<Material> toProcess = exchange.exchange(new ArrayList<>());

package com.nexussoftware.bibliotech.service;

import java.util.List;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

/**
 * Calculates the sum of fines for a list of loans by dividing
 * the work recursively.
 *
 * RecursiveTask<Double>: the <Double> is the type of the result
 * this task produces (10-01 for generics).
 */
public class ParallelFineCalculation extends RecursiveTask<Double> {

    /**
     * THRESHOLD: below this size the work is computed directly.
     * Choosing it well is the most important thing in the whole pattern:
     *  - too low: the cost of creating and coordinating subtasks
     *    exceeds the computation, and it runs SLOWER than sequentially;
     *  - too high: there are not enough chunks for all the cores.
     * Rule of thumb: between 100 and 10,000 elements, and MEASURE.
     */
    private static final int THRESHOLD = 1_000;

    private final List<Loan> loans;
    private final int from;
    private final int to;
    private final FineCalculator calculator;

    public ParallelFineCalculation(List<Loan> loans, int from, int to,
                                   FineCalculator calculator) {
        this.loans = loans;
        this.from = from;
        this.to = to;
        this.calculator = calculator;
    }

    @Override
    protected Double compute() {
        int size = to - from;

        // BASE CASE: a small chunk, direct sequential computation.
        if (size <= THRESHOLD) {
            double sum = 0;
            for (int i = from; i < to; i++) {
                sum += calculator.calculate(loans.get(i));
            }
            return sum;
        }

        // RECURSIVE CASE: split in half.
        int middle = from + size / 2;
        ParallelFineCalculation left =
                new ParallelFineCalculation(loans, from, middle, calculator);
        ParallelFineCalculation right =
                new ParallelFineCalculation(loans, middle, to, calculator);

        // CORRECT PATTERN: fork ONE and compute the other on THIS thread.
        // Doing left.fork() + right.fork() + left.join() + right.join()
        // wastes the current thread, which would just sit and wait.
        left.fork();                        // queued: another thread can steal it
        double rightResult = right.compute();    // this thread works
        double leftResult = left.join();         // collect the forked one

        return leftResult + rightResult;
    }

    // --- Usage ---
    public static double calculateAll(List<Loan> loans,
                                      FineCalculator calculator) {
        // The COMMON pool: shared by the whole JVM, with
        // (cores - 1) threads. Parallel streams use it too (10-04).
        ForkJoinPool common = ForkJoinPool.commonPool();
        return common.invoke(new ParallelFineCalculation(
                loans, 0, loans.size(), calculator));
    }
}

// FORBIDDEN in the common pool: blocking operations.
ForkJoinPool.commonPool().submit(() -> {
    return Files.readAllLines(hugePath);     // blocks a shared thread
});

// CORRECT: your own pool for blocking work.
ForkJoinPool ownPool = new ForkJoinPool(4);
try {
    ownPool.invoke(new ParallelFineCalculation(...));
} finally {
    ownPool.shutdown();
}

package com.nexussoftware.bibliotech.service;

import com.nexussoftware.bibliotech.domain.Loan;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Bulk sending of due-date notices.
 *
 * Properties:
 *  - A BOUNDED pool with named threads and a bounded queue.
 *  - A SEMAPHORE limiting simultaneous accesses to the notices file to 5.
 *  - A COUNTDOWNLATCH to show progress and wait for the end.
 *  - Clean cancellation through shutdownNow() -> cooperative interruption.
 *  - Aggregated result with atomic counters (details in 08-06).
 */
public class NoticeService implements AutoCloseable {

    private static final Logger LOG = Logger.getLogger(NoticeService.class.getName());

    private static final int THREADS = 8;
    private static final int MAX_CONCURRENT_WRITES = 5;

    private final ThreadPoolExecutor pool;
    private final Semaphore fileAccess =
            new Semaphore(MAX_CONCURRENT_WRITES, true);

    public NoticeService() {
        AtomicInteger n = new AtomicInteger(1);
        this.pool = new ThreadPoolExecutor(
                THREADS, THREADS,
                0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(500),          // BOUNDED queue
                r -> {
                    Thread t = new Thread(r, "bibliotech-notices-" + n.getAndIncrement());
                    t.setUncaughtExceptionHandler((h, e) ->
                            LOG.log(Level.SEVERE, "Uncaught failure in " + h.getName(), e));
                    return t;
                },
                new ThreadPoolExecutor.CallerRunsPolicy());  // backpressure
    }

    /** Aggregated result of the send. Immutable (08-04). */
    public record SendSummary(int total, int sent, int failed,
                              int cancelled, long milliseconds) {

        public double successRate() {
            return total == 0 ? 0 : 100.0 * sent / total;
        }
    }

    /**
     * Sends all the notices in parallel.
     *
     * @param deadlineSeconds global deadline; on expiry the pending work is cancelled
     */
    public SendSummary sendAll(List<Loan> overdue, long deadlineSeconds)
            throws InterruptedException {

        final int total = overdue.size();
        final CountDownLatch finished = new CountDownLatch(total);
        final AtomicInteger sent = new AtomicInteger();
        final AtomicInteger failed = new AtomicInteger();

        long start = System.nanoTime();
        List<Future<?>> futures = new ArrayList<>(total);

        // 1. Submit every task to the pool.
        for (Loan l : overdue) {
            futures.add(pool.submit(() -> {
                try {
                    // A cancellation point before starting.
                    if (Thread.currentThread().isInterrupted()) {
                        return;
                    }
                    // The semaphore limits concurrent access to the file.
                    fileAccess.acquire();
                    try {
                        writeNotice(l);
                        sent.incrementAndGet();
                    } finally {
                        fileAccess.release();  // ALWAYS
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();     // restore (08-02)
                } catch (Exception e) {
                    failed.incrementAndGet();
                    LOG.log(Level.WARNING, "Notice failed for " + l.id(), e);
                } finally {
                    // ALWAYS, or the await() never returns.
                    finished.countDown();
                }
            }));
        }

        // 2. Progress while waiting, with a global deadline.
        long limitNs = TimeUnit.SECONDS.toNanos(deadlineSeconds);
        boolean completed = false;

        while (!(completed = finished.await(300, TimeUnit.MILLISECONDS))) {
            long done = total - finished.getCount();
            System.out.printf("\r  [%s] %d/%d notices (%.0f%%)",
                    bar(done, total), done, total, 100.0 * done / total);

            if (System.nanoTime() - start > limitNs) {
                System.out.println("\n  Deadline expired: cancelling what is pending");
                for (Future<?> f : futures) {
                    f.cancel(true);        // interrupts or drops from the queue
                }
                break;
            }
        }

        long ms = (System.nanoTime() - start) / 1_000_000;
        int cancelled = total - sent.get() - failed.get();

        System.out.printf("\r  [%s] %d/%d notices (100%%)%n",
                bar(total, total), total, total);

        return new SendSummary(total, sent.get(), failed.get(), cancelled, ms);
    }

    private static String bar(long done, long total) {
        int width = 30;
        int filled = total == 0 ? 0 : (int) (width * done / total);
        return "#".repeat(filled) + "-".repeat(width - filled);
    }

    /** Simulates writing the notice: 250 ms of I/O. */
    private void writeNotice(Loan l) throws InterruptedException {
        TimeUnit.MILLISECONDS.sleep(250);
        if (l.id().hashCode() % 37 == 0) {
            throw new IllegalStateException("employee has no contact address");
        }
    }

    /** AutoCloseable: two-phase shutdown (06-06 + section 7). */
    @Override
    public void close() {
        pool.shutdown();
        try {
            if (!pool.awaitTermination(30, TimeUnit.SECONDS)) {
                pool.shutdownNow();
                if (!pool.awaitTermination(10, TimeUnit.SECONDS)) {
                    LOG.severe("There are notice tasks ignoring the interruption");
                }
            }
        } catch (InterruptedException e) {
            pool.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}

package com.nexussoftware.bibliotech.presentation;

import java.util.List;

public class NoticeDemo {

    public static void main(String[] args) throws InterruptedException {

        List<Loan> overdue = generateOverdue(200);

        System.out.println("=== SENDING DUE-DATE NOTICES ===");
        System.out.println("Pending notices: " + overdue.size());
        System.out.println("Estimated sequential time: "
                + (overdue.size() * 250 / 1000) + " s");
        System.out.println();

        // try-with-resources: close() shuts the pool down in an orderly way.
        try (NoticeService service = new NoticeService()) {

            NoticeService.SendSummary r = service.sendAll(overdue, 60);

            System.out.println();
            System.out.println("=== SUMMARY ===");
            System.out.println("Total      : " + r.total());
            System.out.println("Sent       : " + r.sent());
            System.out.println("Failed     : " + r.failed());
            System.out.println("Cancelled  : " + r.cancelled());
            System.out.println("Time       : " + r.milliseconds() + " ms");
            System.out.printf ("Success    : %.1f%%%n", r.successRate());
            System.out.printf ("Speed-up   : %.1fx over sequential sending%n",
                    (overdue.size() * 250.0) / r.milliseconds());
        }
    }
}

=== SENDING DUE-DATE NOTICES ===
Pending notices: 200
Estimated sequential time: 50 s

  [##############################] 200/200 notices (100%)

=== SUMMARY ===
Total      : 200
Sent       : 194
Failed     : 6
Cancelled  : 0
Time       : 10247 ms
Success    : 97.0%
Speed-up   : 4.9x over sequential sending

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.*;

public class PoolComparison {

    static final int TASKS = 300;
    static final long DURATION_MS = 100;

    /** A simulated I/O task that records which thread ran it. */
    static Runnable task(Set<String> threadsUsed) {
        return () -> {
            threadsUsed.add(Thread.currentThread().getName());
            try {
                TimeUnit.MILLISECONDS.sleep(DURATION_MS);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        };
    }

    record Result(String strategy, long ms, int threads) { }

    /** (a) One thread per task, managed by hand. */
    static Result threadPerTask() throws InterruptedException {
        Set<String> threads = Collections.synchronizedSet(new HashSet<>());
        Thread[] ts = new Thread[TASKS];
        long start = System.nanoTime();
        for (int i = 0; i < TASKS; i++) {
            ts[i] = new Thread(task(threads), "manual-" + i);
            ts[i].start();
        }
        for (Thread t : ts) t.join();
        return new Result("thread per task",
                (System.nanoTime() - start) / 1_000_000, threads.size());
    }

    /** (b)-(e) Any ExecutorService. */
    static Result withExecutor(String name, ExecutorService ex)
            throws InterruptedException {
        Set<String> threads = Collections.synchronizedSet(new HashSet<>());
        long start = System.nanoTime();
        for (int i = 0; i < TASKS; i++) {
            ex.execute(task(threads));
        }
        ex.shutdown();
        ex.awaitTermination(2, TimeUnit.MINUTES);
        return new Result(name,
                (System.nanoTime() - start) / 1_000_000, threads.size());
    }

    public static void main(String[] args) throws InterruptedException {

        System.out.printf("%d tasks of %d ms of simulated I/O%n", TASKS, DURATION_MS);
        System.out.printf("Sequential would be: %d ms%n%n", TASKS * DURATION_MS);

        Result[] results = {
            threadPerTask(),
            withExecutor("fixed(4)",  Executors.newFixedThreadPool(4)),
            withExecutor("fixed(32)", Executors.newFixedThreadPool(32)),
            withExecutor("cached",    Executors.newCachedThreadPool()),
            withExecutor("bespoke (16, queue 50, CallerRuns)",
                new ThreadPoolExecutor(16, 16, 0L, TimeUnit.MILLISECONDS,
                        new ArrayBlockingQueue<>(50),
                        new ThreadPoolExecutor.CallerRunsPolicy()))
        };

        System.out.printf("%-38s | %8s | %8s%n", "Strategy", "ms", "threads");
        System.out.println("---------------------------------------|----------|---------");
        for (Result r : results) {
            System.out.printf("%-38s | %8d | %8d%n", r.strategy(), r.ms(), r.threads());
        }
    }
}

300 tasks of 100 ms of simulated I/O
Sequential would be: 30000 ms

Strategy                               |       ms |  threads
---------------------------------------|----------|---------
thread per task                        |      147 |      300
fixed(4)                               |     7621 |        4
fixed(32)                              |     1024 |       32
cached                                 |      163 |      300
bespoke (16, queue 50, CallerRuns)     |     1938 |       17

package com.nexussoftware.bibliotech.persistence;

import java.nio.file.Path;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

public class ImportWithDeadline {

    private static final Logger LOG = Logger.getLogger(ImportWithDeadline.class.getName());

    /** Immutable report of the result. */
    public record Report(int imported, int discarded, long ms) { }

    /**
     * The import as a Callable<Report>: it returns a value and can
     * throw checked exceptions. It is cancellable: it checks for
     * interruption on every line (08-02).
     */
    static class ImportTask implements Callable<Report> {

        private final Path file;
        private final AtomicInteger progress = new AtomicInteger();

        ImportTask(Path file) { this.file = file; }

        int progress() { return progress.get(); }

        @Override
        public Report call() throws Exception {
            long start = System.nanoTime();
            int imported = 0, discarded = 0;

            for (int line = 0; line < 50_000; line++) {

                // CANCELLATION POINT: the Future's cancel(true) turns
                // into interrupt() on this thread.
                if (Thread.currentThread().isInterrupted()) {
                    throw new InterruptedException(
                            "import cancelled at line " + line);
                }

                if (line % 97 == 0) discarded++; else imported++;
                progress.incrementAndGet();

                if (line % 1000 == 0) {
                    TimeUnit.MILLISECONDS.sleep(30);   // simulates heavy validation
                }
            }
            return new Report(imported, discarded,
                    (System.nanoTime() - start) / 1_000_000);
        }
    }

    public static void main(String[] args) {

        ExecutorService executor = Executors.newSingleThreadExecutor(
                r -> new Thread(r, "bibliotech-importer"));

        ImportTask task = new ImportTask(Path.of("data/inventory.csv"));
        Future<Report> future = executor.submit(task);

        try {
            // 1-2. The main thread stays alive: it serves the menu and paints progress.
            System.out.println("[main] import launched; the menu is still active");
            while (!future.isDone()) {
                System.out.printf("\r[main] progress: %d lines", task.progress());
                TimeUnit.MILLISECONDS.sleep(300);
            }

            // 3. A 5-second deadline.
            Report report = future.get(5, TimeUnit.SECONDS);
            System.out.printf("%n[main] COMPLETED: %d imported, %d discarded, %d ms%n",
                    report.imported(), report.discarded(), report.ms());

        } catch (TimeoutException e) {
            // 4. The deadline does NOT cancel on its own: you have to ask.
            System.out.printf("%n[main] deadline expired after %d lines; cancelling%n",
                    task.progress());
            boolean requested = future.cancel(true);
            System.out.println("[main] cancel() returned  : " + requested);
            System.out.println("[main] isCancelled()      : " + future.isCancelled());
            System.out.println("[main] the catalogue keeps its previous content");

        } catch (ExecutionException e) {
            // 5. The task failed: the real cause is in getCause() (06-03).
            Throwable cause = e.getCause();
            if (cause instanceof InterruptedException) {
                System.out.println("\n[main] the import stopped because of cancellation");
            } else {
                LOG.log(Level.SEVERE, "The import failed", cause);
                System.out.println("\n[main] failure: " + cause.getMessage());
            }

        } catch (InterruptedException e) {
            // WE have been interrupted; the task is still alive.
            System.out.println("\n[main] wait interrupted");
            future.cancel(true);
            Thread.currentThread().interrupt();

        } finally {
            // 6. Two-phase shutdown.
            executor.shutdown();
            try {
                if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
                    executor.shutdownNow();
                    if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
                        LOG.severe("The importer ignores the interruption");
                    }
                }
            } catch (InterruptedException e) {
                executor.shutdownNow();
                Thread.currentThread().interrupt();
            }
            System.out.println("[main] executor shut down; the JVM can terminate");
        }
    }
}

[main] import launched; the menu is still active
[main] progress: 41893 lines
[main] deadline expired after 42017 lines; cancelling
[main] cancel() returned  : true
[main] isCancelled()      : true
[main] the catalogue keeps its previous content
[main] executor shut down; the JVM can terminate

package com.nexussoftware.bibliotech.service;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ReadingRoom {

    private static final int CAPACITY = 4;
    private static final int EMPLOYEES = 20;

    // A FAIR semaphore: those who have waited longest get in first,
    // which prevents starvation (08-04, section 12).
    private final Semaphore seats = new Semaphore(CAPACITY, true);

    // Instantaneous occupancy, to check the invariant.
    private final AtomicInteger inside = new AtomicInteger();
    private final AtomicInteger maxObserved = new AtomicInteger();
    private volatile boolean invariantBroken = false;

    /** An employee enters, stays a while and leaves. */
    public void visit(String employee) throws InterruptedException {

        seats.acquire();          // blocks if 4 are already inside
        try {
            int current = inside.incrementAndGet();

            // Recording the maximum observed, with a CAS loop (08-06).
            maxObserved.updateAndGet(m -> Math.max(m, current));

            if (current > CAPACITY) {
                invariantBroken = true;
                System.err.println("!!! INVARIANT BROKEN: " + current + " inside !!!");
            }

            System.out.printf("  [+] %-12s enters (inside=%d, waiting=%d)%n",
                    employee, current, seats.getQueueLength());

            TimeUnit.MILLISECONDS.sleep(
                    ThreadLocalRandom.current().nextInt(200, 601));

            System.out.printf("  [-] %-12s leaves (inside=%d)%n",
                    employee, inside.get() - 1);

        } finally {
            inside.decrementAndGet();
            // MANDATORY in finally: a lost permit never comes back,
            // and after four losses the room would be closed forever.
            seats.release();
        }
    }

    public static void main(String[] args) throws InterruptedException {

        ReadingRoom room = new ReadingRoom();

        // Two latches: the "starting pistol" and the "finish line".
        CountDownLatch startGate = new CountDownLatch(1);
        CountDownLatch finishLine = new CountDownLatch(EMPLOYEES);

        // Periodic occupancy monitor.
        ScheduledExecutorService monitor = Executors.newSingleThreadScheduledExecutor(
                r -> {
                    Thread t = new Thread(r, "bibliotech-occupancy-monitor");
                    t.setDaemon(true);
                    return t;
                });

        monitor.scheduleAtFixedRate(() -> {
            try {
                System.out.printf("      [monitor] occupancy=%d/%d  queue=%d%n",
                        room.inside.get(), CAPACITY, room.seats.getQueueLength());
            } catch (Exception e) {
                // An uncaught exception WOULD CANCEL the schedule
                // forever and silently.
                e.printStackTrace();
            }
        }, 200, 200, TimeUnit.MILLISECONDS);

        // 20 threads that wait for the gate and start at once.
        for (int i = 1; i <= EMPLOYEES; i++) {
            final String name = "employee-" + String.format("%02d", i);
            new Thread(() -> {
                try {
                    startGate.await();         // everybody waits here
                    room.visit(name);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    finishLine.countDown();    // ALWAYS
                }
            }, name).start();
        }

        System.out.println("=== READING ROOM (capacity " + CAPACITY + ") ===");
        TimeUnit.MILLISECONDS.sleep(200);      // let everybody reach await()

        long start = System.nanoTime();
        startGate.countDown();                 // all at once!
        finishLine.await();
        long ms = (System.nanoTime() - start) / 1_000_000;

        monitor.shutdown();
        monitor.awaitTermination(2, TimeUnit.SECONDS);

        System.out.println();
        System.out.println("=== SUMMARY ===");
        System.out.println("Employees served    : " + EMPLOYEES);
        System.out.println("Maximum occupancy   : " + room.maxObserved.get()
                + " (limit " + CAPACITY + ")");
        System.out.println("Invariant respected : " + !room.invariantBroken);
        System.out.println("Total time          : " + ms + " ms");
        System.out.println("Inside at the end   : " + room.inside.get());
        System.out.println("Permits available   : " + room.seats.availablePermits()
                + " (must be " + CAPACITY + ")");
    }
}

=== READING ROOM (capacity 4) ===
  [+] employee-03  enters (inside=1, waiting=16)
  [+] employee-07  enters (inside=2, waiting=15)
  [+] employee-01  enters (inside=3, waiting=14)
  [+] employee-12  enters (inside=4, waiting=13)
      [monitor] occupancy=4/4  queue=16
  [-] employee-03  leaves (inside=3)
  [+] employee-05  enters (inside=4, waiting=15)
  ...

=== SUMMARY ===
Employees served    : 20
Maximum occupancy   : 4 (limit 4)
Invariant respected : true
Total time          : 2143 ms
Inside at the end   : 0
Permits available   : 4 (must be 4)