// LO QUE HACIAS: un hilo por tarea, gestion manual completa.
Thread[] hilos = new Thread[200];
for (int i = 0; i < 200; i++) {
    hilos[i] = new Thread(new TareaAviso(avisos.get(i)), "bibliotech-avisos-" + i);
    hilos[i].start();
}
for (Thread h : hilos) {
    h.join();
}
// ¿Y el resultado de cada tarea? Campos volatile y leerlos a mano.
// ¿Y si una falla? Un manejador de excepciones no capturadas.
// ¿Y si quiero cancelar todo? interrupt() a los 200, uno por uno.
// ¿Y si son 200.000 avisos? OutOfMemoryError.

// LO QUE HARAS: describir tareas, el ejecutor las coloca.
ExecutorService ejecutor = Executors.newFixedThreadPool(8);
List<Future<ResultadoAviso>> futuros = new ArrayList<>();

for (Aviso a : avisos) {
    futuros.add(ejecutor.submit(new TareaAviso(a)));   // Callable: devuelve valor
}

for (Future<ResultadoAviso> f : futuros) {
    ResultadoAviso r = f.get();   // espera, y RELANZA la excepcion si la hubo
}

ejecutor.shutdown();

// La interfaz completa. Un solo metodo.
public interface Executor {
    void execute(Runnable comando);
}

// Tres implementaciones validas de la MISMA interfaz:
Executor enElMismoHilo  = tarea -> tarea.run();
Executor unHiloNuevo    = tarea -> new Thread(tarea).start();
Executor unPool         = Executors.newFixedThreadPool(8);

// El codigo cliente no cambia:
enElMismoHilo.execute(() -> System.out.println("hola"));
unPool.execute(() -> System.out.println("hola"));

public interface ExecutorService extends Executor {

    // Enviar tareas y obtener un Future
    Future<?> submit(Runnable tarea);
    <T> Future<T> submit(Callable<T> tarea);
    <T> Future<T> submit(Runnable tarea, T resultado);

    // Enviar varias
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tareas);
    <T> T invokeAny(Collection<? extends Callable<T>> tareas);

    // Apagado
    void shutdown();
    List<Runnable> shutdownNow();
    boolean isShutdown();
    boolean isTerminated();
    boolean awaitTermination(long tiempo, TimeUnit unidad);
}

// execute(Runnable): no devuelve nada. Si la tarea lanza una excepcion,
// va al manejador de excepciones no capturadas del hilo (08-02) y
// normalmente se imprime en la consola.
ejecutor.execute(() -> { throw new RuntimeException("fallo visible"); });

// submit(...): devuelve un Future. Si la tarea lanza una excepcion,
// esta se GUARDA en el Future y NO se imprime en ninguna parte.
// Si nadie llama a get(), el fallo desaparece SIN DEJAR RASTRO.
ejecutor.submit(() -> { throw new RuntimeException("fallo INVISIBLE"); });

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

// 1. Pool FIJO: N hilos, siempre los mismos, cola ilimitada.
ExecutorService fijo = Executors.newFixedThreadPool(8);

// 2. Pool ELASTICO: crea hilos segun necesidad, los recicla a los 60 s.
ExecutorService elastico = Executors.newCachedThreadPool();

// 3. UN SOLO HILO: garantiza orden de ejecucion.
ExecutorService unico = Executors.newSingleThreadExecutor();

// 4. PROGRAMADO: tareas retardadas y periodicas.
ScheduledExecutorService programado = Executors.newScheduledThreadPool(2);

// 5. ROBO DE TRABAJO: ForkJoinPool con paralelismo = nucleos.
ExecutorService robo = Executors.newWorkStealingPool();

// Envio de avisos: E/S, carga acotada y conocida -> pool fijo.
ExecutorService avisos = Executors.newFixedThreadPool(8);

// Escritura del log de auditoria: DEBE conservar el orden -> un solo hilo.
// Ademas, al ser un unico hilo, no hace falta sincronizar el fichero.
ExecutorService auditoria = Executors.newSingleThreadExecutor();

// Aviso de vencimientos cada hora y copia de seguridad cada noche.
ScheduledExecutorService mantenimiento = Executors.newScheduledThreadPool(2);

// BOMBA DE RELOJERIA
ExecutorService pool = Executors.newFixedThreadPool(4);

// Cada tarea tarda 100 ms; llegan 1.000 por segundo.
// El pool procesa 40/s. Se acumulan 960 tareas por segundo en memoria.
// En 10 minutos: ~576.000 objetos en cola. OutOfMemoryError.
while (hayPeticiones()) {
    pool.submit(new TareaLenta(siguientePeticion()));
}

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

public class EjecutorBiblioTech {

    /**
     * ThreadFactory que nombra los hilos. Sin ella, los hilos del pool
     * se llaman "pool-1-thread-1", que en un volcado (08-03) no dice nada.
     */
    static class FabricaHilosNombrados implements ThreadFactory {
        private final String prefijo;
        private final boolean demonio;
        private final AtomicInteger contador = new AtomicInteger(1);

        FabricaHilosNombrados(String prefijo, boolean demonio) {
            this.prefijo = prefijo;
            this.demonio = demonio;
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r, prefijo + "-" + contador.getAndIncrement());
            t.setDaemon(demonio);
            // Red de seguridad: cualquier excepcion que escape de una tarea
            // enviada con execute() queda registrada (06-07, 08-02).
            t.setUncaughtExceptionHandler((hilo, error) ->
                    System.err.println("[" + hilo.getName() + "] fallo no capturado: " + error));
            return t;
        }
    }

    public static ThreadPoolExecutor crear() {
        return new ThreadPoolExecutor(
                4,                                    // 1. tamano del nucleo
                8,                                    // 2. tamano maximo
                60L, TimeUnit.SECONDS,                // 3. vida de los hilos extra
                new ArrayBlockingQueue<>(100),        // 4. cola ACOTADA
                new FabricaHilosNombrados("bibliotech-avisos", false),   // 5.
                new ThreadPoolExecutor.CallerRunsPolicy()                 // 6.
        );
    }
}

public class DimensionadoPool {

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

    /** Calculo puro: mas hilos que nucleos solo anade cambios de contexto. */
    public static int paraCalculo() {
        return NUCLEOS;
    }

    /**
     * E/S: la formula de Brian Goetz.
     *   hilos = nucleos * utilizacionObjetivo * (1 + espera / calculo)
     * Con utilizacion 1.0, 300 ms de espera y 20 ms de calculo en 8 nucleos:
     *   8 * 1.0 * (1 + 15) = 128 hilos.
     * En la practica se acota a un maximo razonable: 128 hilos son 128 MB
     * de pilas, y el recurso remoto probablemente no aguante 128 peticiones.
     */
    public static int paraEs(double msEspera, double msCalculo) {
        double ratio = msEspera / msCalculo;
        int sugerido = (int) Math.ceil(NUCLEOS * (1 + ratio));
        return Math.min(sugerido, 64);       // tope de cordura
    }

    public static void main(String[] args) {
        System.out.println("Nucleos             : " + NUCLEOS);
        System.out.println("Pool de calculo     : " + paraCalculo());
        System.out.println("Pool de avisos      : " + paraEs(300, 20));
        System.out.println("Pool de importacion : " + paraEs(50, 30));
    }
}

// BIEN: tres pools con propositos y tamanos distintos, y aislados
// entre si: una saturacion del pool de avisos no afecta al de calculo.
private final ExecutorService poolCalculo = Executors.newFixedThreadPool(NUCLEOS);
private final ExecutorService poolEs      = Executors.newFixedThreadPool(32);
private final ScheduledExecutorService poolProgramado =
        Executors.newScheduledThreadPool(2);

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

public final class ApagadoOrdenado {

    /**
     * Apagado en dos fases:
     *   1. shutdown() + espera cortes: dejar terminar lo que esta en curso.
     *   2. Si no terminan, shutdownNow() (interrupcion) + espera corta.
     *   3. Si aun asi no terminan, registrarlo: hay tareas que no
     *      responden a la interrupcion, y eso es un BUG de esas tareas
     *      (el catch vacio de InterruptedException de 08-02).
     */
    public static void apagar(ExecutorService ejecutor, long segundos) {
        ejecutor.shutdown();                     // fase 1: no acepta mas
        try {
            if (!ejecutor.awaitTermination(segundos, TimeUnit.SECONDS)) {
                ejecutor.shutdownNow();          // fase 2: interrumpir
                if (!ejecutor.awaitTermination(segundos, TimeUnit.SECONDS)) {
                    System.err.println("El ejecutor no ha terminado: "
                            + "hay tareas que ignoran la interrupcion");
                }
            }
        } catch (InterruptedException e) {
            // Nos han interrumpido a NOSOTROS mientras esperabamos.
            ejecutor.shutdownNow();
            Thread.currentThread().interrupt();  // restaurar la bandera (08-02)
        }
    }
}

@FunctionalInterface
public interface Callable<V> {
    V call() throws Exception;      // devuelve V y PUEDE lanzar
}

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

package com.nexussoftware.bibliotech.persistencia;

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

public class ImportacionConFuture {

    public static void main(String[] args) {

        ExecutorService ejecutor = Executors.newFixedThreadPool(2);
        try {
            // Callable<Informe>: devuelve un Informe y puede lanzar
            // excepciones comprobadas. Ambas cosas imposibles con Runnable.
            Callable<Informe> tarea = () -> {
                ImportadorCatalogo imp = new ImportadorCatalogo();
                return imp.importar(Path.of("datos/inventario.csv"));  // throws IOException
            };

            Future<Informe> futuro = ejecutor.submit(tarea);

            // El hilo principal NO esta bloqueado: puede seguir trabajando.
            System.out.println("[main] importacion lanzada, atiendo el menu");
            mostrarMenu();

            // isDone() no bloquea: permite sondear sin esperar.
            while (!futuro.isDone()) {
                System.out.println("[main] importando...");
                TimeUnit.MILLISECONDS.sleep(500);
            }

            // get() con PLAZO: nunca uses get() sin plazo en produccion.
            Informe informe = futuro.get(30, TimeUnit.SECONDS);
            System.out.println("[main] importados: " + informe.importados());

        } catch (TimeoutException e) {
            System.err.println("[main] la importacion excedio el plazo");

        } catch (ExecutionException e) {
            // La tarea lanzo una excepcion. get() la ENVUELVE en
            // ExecutionException; la original esta en getCause().
            // Es exactamente el encadenamiento de causas de 06-03.
            Throwable causa = e.getCause();
            System.err.println("[main] la importacion fallo: " + causa.getMessage());
            if (causa instanceof java.io.IOException) {
                System.err.println("[main] problema de fichero; se conserva el catalogo anterior");
            }

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

        } finally {
            ApagadoOrdenado.apagar(ejecutor, 10);
        }
    }

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

try {
    Informe i = futuro.get(30, TimeUnit.SECONDS);
} catch (ExecutionException e) {
    Throwable causa = e.getCause();
    // Recuperar el tipo original para aplicar la politica de 06-07
    if (causa instanceof FormatoInvalidoException fie) {
        registrarYDegradar(fie);
    } else if (causa instanceof java.io.IOException ioe) {
        abortarConCodigo(ioe);
    } else {
        throw new BiblioTechException("Fallo inesperado en la importacion", causa);
    }
}

Future<Informe> futuro = ejecutor.submit(tarea);

// cancel(false): solo evita que EMPIECE si aun esta en cola.
// Si ya esta corriendo, la deja terminar.
futuro.cancel(false);

// cancel(true): ademas, INTERRUMPE el hilo que la ejecuta.
// Es literalmente un hilo.interrupt() sobre el hilo del pool.
futuro.cancel(true);

/**
 * Tarea cancelable dentro de un pool. Identica al patron de 08-02:
 * cancel(true) del Future se traduce en interrupt() sobre este hilo.
 */
public class TareaAvisoCancelable implements Callable<ResultadoAviso> {

    private final Aviso aviso;

    public TareaAvisoCancelable(Aviso aviso) { this.aviso = aviso; }

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

            // Punto de cancelacion en cada vuelta.
            if (Thread.currentThread().isInterrupted()) {
                throw new InterruptedException("aviso cancelado antes del intento " + intento);
            }
            try {
                enviar(aviso);                       // puede lanzar InterruptedException
                return ResultadoAviso.exito(aviso);
            } catch (EnvioFallidoException e) {
                if (intento == 3) {
                    return ResultadoAviso.fallo(aviso, e.getMessage());
                }
                TimeUnit.MILLISECONDS.sleep(100L * intento);   // retroceso
            }
        }
        return ResultadoAviso.fallo(aviso, "agotados los reintentos");
    }

    private void enviar(Aviso a) throws Exception { /* ... */ }
}

Future<Informe> f = ejecutor.submit(tarea);
TimeUnit.SECONDS.sleep(2);

boolean cancelada = f.cancel(true);
System.out.println("cancel() devolvio : " + cancelada);   // false si ya habia terminado
System.out.println("isCancelled()     : " + f.isCancelled());
System.out.println("isDone()          : " + f.isDone());  // true: cancelada tambien es "hecha"

try {
    f.get();
} catch (CancellationException e) {
    // get() sobre un Future CANCELADO lanza CancellationException,
    // que es NO comprobada. Es facil olvidarse de capturarla.
    System.out.println("confirmado: la tarea fue cancelada");
}

List<Callable<ResultadoAviso>> tareas = new ArrayList<>();
for (Aviso a : avisos) {
    tareas.add(new TareaAvisoCancelable(a));
}

// Bloquea hasta que TODAS terminen. Los Future devueltos
// ya estan completos: get() no bloquea.
List<Future<ResultadoAviso>> resultados = ejecutor.invokeAll(tareas);

int exitos = 0, fallos = 0;
for (Future<ResultadoAviso> f : resultados) {
    try {
        if (f.get().correcto()) exitos++; else fallos++;
    } catch (ExecutionException e) {
        fallos++;
        LOG.log(Level.WARNING, "aviso fallido", e.getCause());
    }
}

// Plazo GLOBAL para el conjunto. Las que no terminen a tiempo
// se CANCELAN automaticamente, y su get() lanzara CancellationException.
List<Future<ResultadoAviso>> resultados =
        ejecutor.invokeAll(tareas, 30, TimeUnit.SECONDS);

// Tres fuentes para el mismo catalogo; nos vale la primera que responda.
List<Callable<Catalogo>> fuentes = List.of(
        () -> cargarDesdeCacheLocal(),
        () -> cargarDesdeFicheroPrincipal(),
        () -> cargarDesdeCopiaSeguridad()
);

// Devuelve el primer resultado correcto; cancela los otros dos.
// Si TODAS fallan, lanza ExecutionException con la ultima causa.
Catalogo c = ejecutor.invokeAny(fuentes, 10, TimeUnit.SECONDS);

import java.util.concurrent.*;

public class MantenimientoBiblioTech {

    private final ScheduledExecutorService programador =
            Executors.newScheduledThreadPool(2, r -> {
                Thread t = new Thread(r, "bibliotech-programador");
                t.setDaemon(true);      // mantenimiento: no debe impedir el cierre
                return t;
            });

    public void arrancar() {

        // 1. UNA vez, dentro de 5 segundos.
        programador.schedule(
                () -> System.out.println("[mantenimiento] arranque completado"),
                5, TimeUnit.SECONDS);

        // 2. PERIODICA a ritmo fijo: cada hora desde el instante de inicio.
        programador.scheduleAtFixedRate(
                this::avisarVencimientos,
                0, 1, TimeUnit.HOURS);

        // 3. PERIODICA con retardo fijo: 24 h desde que TERMINA la anterior.
        programador.scheduleWithFixedDelay(
                this::copiaSeguridad,
                1, 24, TimeUnit.HOURS);
    }

    /**
     * REGLA DE ORO: una tarea periodica debe capturar TODA excepcion.
     * Si deja escapar una, la JVM CANCELA la planificacion y la tarea
     * NO SE VUELVE A EJECUTAR JAMAS, en silencio, sin ningun aviso.
     * Es el fallo mas traicionero de esta API.
     */
    private void avisarVencimientos() {
        try {
            int enviados = servicioAvisos.enviarVencimientos();
            LOG.log(Level.INFO, "Avisos de vencimiento enviados: {0}", enviados);
        } catch (Exception e) {
            LOG.log(Level.SEVERE, "Fallo al avisar vencimientos", e);
            // NO relanzar: si sale, se acaba la planificacion para siempre.
        }
    }

    private void copiaSeguridad() {
        try {
            gestorCopias.copiar();
        } catch (Exception e) {
            LOG.log(Level.SEVERE, "Fallo en la copia de seguridad", e);
        }
    }
}

Tarea que dura 3 s, periodo 5 s:

scheduleAtFixedRate:   [---3s---]__2s__[---3s---]__2s__[---3s---]
                       0         3     5         8     10
                       inicio cada 5 s exactos

scheduleWithFixedDelay:[---3s---]____5s____[---3s---]____5s____[--
                       0         3         8         11        16
                       5 s DESPUES de terminar

CountDownLatch cierre = new CountDownLatch(5);   // 5 eventos pendientes

cierre.countDown();          // resta 1 (nunca por debajo de 0)
cierre.await();              // bloquea hasta llegar a 0
cierre.await(10, TimeUnit.SECONDS);   // con plazo; devuelve false si expira
cierre.getCount();           // cuenta actual (para mostrar progreso)

import java.util.concurrent.*;

public class EsperarNTareas {

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

        final int TAREAS = 200;
        ExecutorService pool = Executors.newFixedThreadPool(8);
        CountDownLatch terminadas = new CountDownLatch(TAREAS);

        for (int i = 0; i < TAREAS; i++) {
            final int n = i;
            pool.execute(() -> {
                try {
                    enviarAviso(n);
                } catch (Exception e) {
                    LOG.log(Level.WARNING, "aviso " + n + " fallido", e);
                } finally {
                    // OBLIGATORIO en finally: si una tarea falla y no
                    // hace countDown, el await() de main no vuelve JAMAS.
                    terminadas.countDown();
                }
            });
        }

        // Progreso mientras esperamos, sin sondear con sleep a ciegas.
        while (!terminadas.await(500, TimeUnit.MILLISECONDS)) {
            long pendientes = terminadas.getCount();
            System.out.printf("\rProgreso: %d/%d (%.0f%%)",
                    TAREAS - pendientes, TAREAS,
                    100.0 * (TAREAS - pendientes) / TAREAS);
        }
        System.out.println("\nTodos los avisos procesados");
        ApagadoOrdenado.apagar(pool, 10);
    }
}

/**
 * Dos cierres: uno para dar la salida a todos a la vez, otro para
 * esperar a que todos acaben. Es el esqueleto de una prueba de estres
 * de concurrencia bien hecha: sin la puerta, los primeros hilos
 * terminarian antes de que arrancaran los ultimos, y no habria
 * solapamiento real (el efecto que viste en el ejercicio 2 de 08-01).
 */
public static long pruebaDeEstres(int hilos, Runnable tarea) throws InterruptedException {

    CountDownLatch salida    = new CountDownLatch(1);       // la pistola
    CountDownLatch terminados = new CountDownLatch(hilos);  // la meta

    for (int i = 0; i < hilos; i++) {
        new Thread(() -> {
            try {
                salida.await();      // todos esperan aqui
                tarea.run();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                terminados.countDown();
            }
        }, "estres-" + i).start();
    }

    long inicio = System.nanoTime();
    salida.countDown();              // ¡ya! los N arrancan a la vez
    terminados.await();
    return System.nanoTime() - inicio;
}

import java.util.concurrent.*;

public class ProcesamientoPorFases {

    public static void main(String[] args) {

        final int TRABAJADORES = 4;

        // La accion de barrera la ejecuta el ULTIMO hilo en llegar,
        // antes de liberar a los demas. Ideal para consolidar la fase.
        CyclicBarrier barrera = new CyclicBarrier(TRABAJADORES, () ->
                System.out.println("--- fase completada por los 4, consolidando ---"));

        for (int i = 0; i < TRABAJADORES; i++) {
            final int id = i;
            new Thread(() -> {
                try {
                    for (int fase = 1; fase <= 3; fase++) {
                        System.out.printf("  trabajador %d procesa la fase %d%n", id, fase);
                        TimeUnit.MILLISECONDS.sleep(100 + id * 50);
                        barrera.await();      // espera a los otros tres
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } catch (BrokenBarrierException e) {
                    // Si un hilo se interrumpe o expira, la barrera se ROMPE
                    // y TODOS los demas reciben esta excepcion. Es correcto:
                    // sin todos los participantes, la fase no tiene sentido.
                    System.err.println("barrera rota: " + e);
                }
            }, "trabajador-" + i).start();
        }
    }
}

package com.nexussoftware.bibliotech.servicio;

import java.util.concurrent.*;

/**
 * Limita a 3 las exportaciones simultaneas.
 *
 * Motivo: cada exportacion abre el fichero completo del catalogo, lo
 * formatea en memoria y lo escribe. Tres a la vez saturan el disco y
 * multiplican el uso de memoria; con veinte, el servidor se arrastra.
 * El semaforo pone un tope duro sin limitar el resto del pool.
 */
public class ServicioExportacion {

    private static final int MAX_SIMULTANEAS = 3;

    // 3 permisos. El 'true' activa la equidad: los que llevan mas tiempo
    // esperando pasan antes, lo que evita la inanicion (08-04).
    private final Semaphore permisos = new Semaphore(MAX_SIMULTANEAS, true);

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

    public Future<Path> exportar(String formato) {
        return pool.submit(() -> {

            // Espera un permiso. Si hay 3 exportaciones en curso, este
            // hilo se bloquea aqui hasta que una termine.
            permisos.acquire();
            try {
                System.out.printf("[%s] exportando (%d permisos libres)%n",
                        Thread.currentThread().getName(),
                        permisos.availablePermits());
                return generarFichero(formato);      // trabajo pesado
            } finally {
                // OBLIGATORIO en finally: un permiso no devuelto se
                // pierde PARA SIEMPRE. Tras tres fallos, el semaforo
                // queda a 0 y nadie vuelve a exportar jamas.
                permisos.release();
            }
        });
    }

    /** Variante que no espera: rechaza en vez de encolar. */
    public Future<Path> exportarSinEsperar(String formato) {
        return pool.submit(() -> {
            if (!permisos.tryAcquire(5, TimeUnit.SECONDS)) {
                throw new ServicioSaturadoException(
                        "Hay " + MAX_SIMULTANEAS + " exportaciones en curso; reintente");
            }
            try {
                return generarFichero(formato);
            } finally {
                permisos.release();
            }
        });
    }

    private Path generarFichero(String formato) throws Exception {
        TimeUnit.SECONDS.sleep(2);
        return Path.of("informes/catalogo." + formato);
    }
}

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

// Hilo lector: llena un bufer y lo intercambia por uno vacio.
List<Material> miBufer = new ArrayList<>();
// ... llenar ...
miBufer = intercambio.exchange(miBufer);   // recibe el bufer vacio del otro

// Hilo procesador: procesa un bufer lleno y devuelve uno vacio.
List<Material> aProcesar = intercambio.exchange(new ArrayList<>());

package com.nexussoftware.bibliotech.servicio;

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

/**
 * Calcula la suma de multas de una lista de prestamos dividiendo
 * el trabajo recursivamente.
 *
 * RecursiveTask<Double>: el <Double> es el tipo del resultado
 * que produce esta tarea (10-01 para los genericos).
 */
public class CalculoMultasParalelo extends RecursiveTask<Double> {

    /**
     * UMBRAL: por debajo de este tamano se calcula directamente.
     * Elegirlo bien es lo mas importante de todo el patron:
     *  - demasiado bajo: el coste de crear y coordinar subtareas
     *    supera al del calculo, y va MAS LENTO que en secuencial;
     *  - demasiado alto: no hay bastantes trozos para todos los nucleos.
     * Regla practica: entre 100 y 10.000 elementos, y MEDIR.
     */
    private static final int UMBRAL = 1_000;

    private final List<Prestamo> prestamos;
    private final int desde;
    private final int hasta;
    private final CalculadoraMultas calculadora;

    public CalculoMultasParalelo(List<Prestamo> prestamos, int desde, int hasta,
                                 CalculadoraMultas calculadora) {
        this.prestamos = prestamos;
        this.desde = desde;
        this.hasta = hasta;
        this.calculadora = calculadora;
    }

    @Override
    protected Double compute() {
        int tamano = hasta - desde;

        // CASO BASE: trozo pequeno, calculo secuencial directo.
        if (tamano <= UMBRAL) {
            double suma = 0;
            for (int i = desde; i < hasta; i++) {
                suma += calculadora.calcular(prestamos.get(i));
            }
            return suma;
        }

        // CASO RECURSIVO: partir por la mitad.
        int medio = desde + tamano / 2;
        CalculoMultasParalelo izq =
                new CalculoMultasParalelo(prestamos, desde, medio, calculadora);
        CalculoMultasParalelo der =
                new CalculoMultasParalelo(prestamos, medio, hasta, calculadora);

        // PATRON CORRECTO: bifurcar UNA y calcular la otra en ESTE hilo.
        // Hacer izq.fork() + der.fork() + izq.join() + der.join()
        // desaprovecha el hilo actual, que se quedaria esperando.
        izq.fork();                       // a la cola: otro hilo puede robarla
        double resultadoDer = der.compute();   // este hilo trabaja
        double resultadoIzq = izq.join();      // recoger la bifurcada

        return resultadoIzq + resultadoDer;
    }

    // --- Uso ---
    public static double calcularTodas(List<Prestamo> prestamos,
                                       CalculadoraMultas calculadora) {
        // El pool COMUN: compartido por toda la JVM, con
        // (nucleos - 1) hilos. Lo usan tambien los streams paralelos (10-04).
        ForkJoinPool comun = ForkJoinPool.commonPool();
        return comun.invoke(new CalculoMultasParalelo(
                prestamos, 0, prestamos.size(), calculadora));
    }
}

// PROHIBIDO en el pool comun: operaciones bloqueantes.
ForkJoinPool.commonPool().submit(() -> {
    return Files.readAllLines(rutaEnorme);   // bloquea un hilo compartido
});

// CORRECTO: pool propio para el trabajo bloqueante.
ForkJoinPool poolPropio = new ForkJoinPool(4);
try {
    poolPropio.invoke(new CalculoMultasParalelo(...));
} finally {
    poolPropio.shutdown();
}

package com.nexussoftware.bibliotech.servicio;

import com.nexussoftware.bibliotech.dominio.Prestamo;

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;

/**
 * Envio masivo de avisos de vencimiento.
 *
 * Propiedades:
 *  - Pool ACOTADO con hilos nombrados y cola acotada.
 *  - SEMAFORO que limita a 5 los accesos simultaneos al fichero de avisos.
 *  - COUNTDOWNLATCH para mostrar progreso y esperar el final.
 *  - Cancelacion limpia mediante shutdownNow() -> interrupcion cooperativa.
 *  - Resultado agregado con contadores atomicos (detalle en 08-06).
 */
public class ServicioAvisos implements AutoCloseable {

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

    private static final int HILOS = 8;
    private static final int MAX_ESCRITURAS_SIMULTANEAS = 5;

    private final ThreadPoolExecutor pool;
    private final Semaphore accesoFichero =
            new Semaphore(MAX_ESCRITURAS_SIMULTANEAS, true);

    public ServicioAvisos() {
        AtomicInteger n = new AtomicInteger(1);
        this.pool = new ThreadPoolExecutor(
                HILOS, HILOS,
                0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(500),          // cola ACOTADA
                r -> {
                    Thread t = new Thread(r, "bibliotech-avisos-" + n.getAndIncrement());
                    t.setUncaughtExceptionHandler((h, e) ->
                            LOG.log(Level.SEVERE, "Fallo no capturado en " + h.getName(), e));
                    return t;
                },
                new ThreadPoolExecutor.CallerRunsPolicy());  // contrapresion
    }

    /** Resultado agregado del envio. Inmutable (08-04). */
    public record ResumenEnvio(int total, int enviados, int fallidos,
                               int cancelados, long milisegundos) {

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

    /**
     * Envia todos los avisos en paralelo.
     *
     * @param plazoSegundos plazo global; al agotarse se cancela lo pendiente
     */
    public ResumenEnvio enviarTodos(List<Prestamo> vencidos, long plazoSegundos)
            throws InterruptedException {

        final int total = vencidos.size();
        final CountDownLatch terminados = new CountDownLatch(total);
        final AtomicInteger enviados = new AtomicInteger();
        final AtomicInteger fallidos = new AtomicInteger();

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

        // 1. Enviar todas las tareas al pool.
        for (Prestamo p : vencidos) {
            futuros.add(pool.submit(() -> {
                try {
                    // Punto de cancelacion antes de empezar.
                    if (Thread.currentThread().isInterrupted()) {
                        return;
                    }
                    // El semaforo limita el acceso concurrente al fichero.
                    accesoFichero.acquire();
                    try {
                        escribirAviso(p);
                        enviados.incrementAndGet();
                    } finally {
                        accesoFichero.release();   // SIEMPRE
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();     // restaurar (08-02)
                } catch (Exception e) {
                    fallidos.incrementAndGet();
                    LOG.log(Level.WARNING, "Aviso fallido para " + p.id(), e);
                } finally {
                    // SIEMPRE, o el await() no vuelve nunca.
                    terminados.countDown();
                }
            }));
        }

        // 2. Progreso mientras se espera, con plazo global.
        long limiteNs = TimeUnit.SECONDS.toNanos(plazoSegundos);
        boolean completado = false;

        while (!(completado = terminados.await(300, TimeUnit.MILLISECONDS))) {
            long hechos = total - terminados.getCount();
            System.out.printf("\r  [%s] %d/%d avisos (%.0f%%)",
                    barra(hechos, total), hechos, total, 100.0 * hechos / total);

            if (System.nanoTime() - inicio > limiteNs) {
                System.out.println("\n  Plazo agotado: cancelando lo pendiente");
                for (Future<?> f : futuros) {
                    f.cancel(true);        // interrumpe o descarta de la cola
                }
                break;
            }
        }

        long ms = (System.nanoTime() - inicio) / 1_000_000;
        int cancelados = total - enviados.get() - fallidos.get();

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

        return new ResumenEnvio(total, enviados.get(), fallidos.get(), cancelados, ms);
    }

    private static String barra(long hechos, long total) {
        int ancho = 30;
        int llenos = total == 0 ? 0 : (int) (ancho * hechos / total);
        return "#".repeat(llenos) + "-".repeat(ancho - llenos);
    }

    /** Simula la escritura del aviso: 250 ms de E/S. */
    private void escribirAviso(Prestamo p) throws InterruptedException {
        TimeUnit.MILLISECONDS.sleep(250);
        if (p.id().hashCode() % 37 == 0) {
            throw new IllegalStateException("empleado sin direccion de contacto");
        }
    }

    /** AutoCloseable: apagado en dos fases (06-06 + apartado 7). */
    @Override
    public void close() {
        pool.shutdown();
        try {
            if (!pool.awaitTermination(30, TimeUnit.SECONDS)) {
                pool.shutdownNow();
                if (!pool.awaitTermination(10, TimeUnit.SECONDS)) {
                    LOG.severe("Hay tareas de aviso que ignoran la interrupcion");
                }
            }
        } catch (InterruptedException e) {
            pool.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}

package com.nexussoftware.bibliotech.presentacion;

import java.util.List;

public class DemostracionAvisos {

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

        List<Prestamo> vencidos = generarVencidos(200);

        System.out.println("=== ENVIO DE AVISOS DE VENCIMIENTO ===");
        System.out.println("Avisos pendientes: " + vencidos.size());
        System.out.println("Tiempo secuencial estimado: "
                + (vencidos.size() * 250 / 1000) + " s");
        System.out.println();

        // try-with-resources: close() apaga el pool ordenadamente.
        try (ServicioAvisos servicio = new ServicioAvisos()) {

            ServicioAvisos.ResumenEnvio r = servicio.enviarTodos(vencidos, 60);

            System.out.println();
            System.out.println("=== RESUMEN ===");
            System.out.println("Total      : " + r.total());
            System.out.println("Enviados   : " + r.enviados());
            System.out.println("Fallidos   : " + r.fallidos());
            System.out.println("Cancelados : " + r.cancelados());
            System.out.println("Tiempo     : " + r.milisegundos() + " ms");
            System.out.printf ("Exito      : %.1f%%%n", r.porcentajeExito());
            System.out.printf ("Aceleracion: %.1fx sobre el envio secuencial%n",
                    (vencidos.size() * 250.0) / r.milisegundos());
        }
    }
}

=== ENVIO DE AVISOS DE VENCIMIENTO ===
Avisos pendientes: 200
Tiempo secuencial estimado: 50 s

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

=== RESUMEN ===
Total      : 200
Enviados   : 194
Fallidos   : 6
Cancelados : 0
Tiempo     : 10247 ms
Exito      : 97.0%
Aceleracion: 4.9x sobre el envio secuencial

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

public class ComparativaPools {

    static final int TAREAS = 300;
    static final long DURACION_MS = 100;

    /** Tarea de E/S simulada que registra qué hilo la ejecutó. */
    static Runnable tarea(Set<String> hilosUsados) {
        return () -> {
            hilosUsados.add(Thread.currentThread().getName());
            try {
                TimeUnit.MILLISECONDS.sleep(DURACION_MS);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        };
    }

    record Resultado(String estrategia, long ms, int hilos) { }

    /** (a) Un hilo por tarea, gestionado a mano. */
    static Resultado unHiloPorTarea() throws InterruptedException {
        Set<String> hilos = Collections.synchronizedSet(new HashSet<>());
        Thread[] ts = new Thread[TAREAS];
        long inicio = System.nanoTime();
        for (int i = 0; i < TAREAS; i++) {
            ts[i] = new Thread(tarea(hilos), "manual-" + i);
            ts[i].start();
        }
        for (Thread t : ts) t.join();
        return new Resultado("un hilo por tarea",
                (System.nanoTime() - inicio) / 1_000_000, hilos.size());
    }

    /** (b)-(e) Cualquier ExecutorService. */
    static Resultado conEjecutor(String nombre, ExecutorService ej)
            throws InterruptedException {
        Set<String> hilos = Collections.synchronizedSet(new HashSet<>());
        long inicio = System.nanoTime();
        for (int i = 0; i < TAREAS; i++) {
            ej.execute(tarea(hilos));
        }
        ej.shutdown();
        ej.awaitTermination(2, TimeUnit.MINUTES);
        return new Resultado(nombre,
                (System.nanoTime() - inicio) / 1_000_000, hilos.size());
    }

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

        System.out.printf("%d tareas de %d ms de E/S simulada%n", TAREAS, DURACION_MS);
        System.out.printf("Secuencial seria: %d ms%n%n", TAREAS * DURACION_MS);

        Resultado[] resultados = {
            unHiloPorTarea(),
            conEjecutor("fixed(4)",  Executors.newFixedThreadPool(4)),
            conEjecutor("fixed(32)", Executors.newFixedThreadPool(32)),
            conEjecutor("cached",    Executors.newCachedThreadPool()),
            conEjecutor("a medida (16, cola 50, CallerRuns)",
                new ThreadPoolExecutor(16, 16, 0L, TimeUnit.MILLISECONDS,
                        new ArrayBlockingQueue<>(50),
                        new ThreadPoolExecutor.CallerRunsPolicy()))
        };

        System.out.printf("%-38s | %8s | %8s%n", "Estrategia", "ms", "hilos");
        System.out.println("---------------------------------------|----------|---------");
        for (Resultado r : resultados) {
            System.out.printf("%-38s | %8d | %8d%n", r.estrategia(), r.ms(), r.hilos());
        }
    }
}

300 tareas de 100 ms de E/S simulada
Secuencial seria: 30000 ms

Estrategia                             |       ms |    hilos
---------------------------------------|----------|---------
un hilo por tarea                      |      147 |      300
fixed(4)                               |     7621 |        4
fixed(32)                              |     1024 |       32
cached                                 |      163 |      300
a medida (16, cola 50, CallerRuns)     |     1938 |       17

package com.nexussoftware.bibliotech.persistencia;

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 ImportacionConPlazo {

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

    /** Informe inmutable del resultado. */
    public record Informe(int importados, int descartados, long ms) { }

    /**
     * Importacion como Callable<Informe>: devuelve valor y puede
     * lanzar excepciones comprobadas. Es cancelable: comprueba la
     * interrupcion en cada linea (08-02).
     */
    static class TareaImportacion implements Callable<Informe> {

        private final Path fichero;
        private final AtomicInteger progreso = new AtomicInteger();

        TareaImportacion(Path fichero) { this.fichero = fichero; }

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

        @Override
        public Informe call() throws Exception {
            long inicio = System.nanoTime();
            int importados = 0, descartados = 0;

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

                // PUNTO DE CANCELACION: cancel(true) del Future se
                // traduce en interrupt() sobre este hilo.
                if (Thread.currentThread().isInterrupted()) {
                    throw new InterruptedException(
                            "importacion cancelada en la linea " + linea);
                }

                if (linea % 97 == 0) descartados++; else importados++;
                progreso.incrementAndGet();

                if (linea % 1000 == 0) {
                    TimeUnit.MILLISECONDS.sleep(30);   // simula validacion pesada
                }
            }
            return new Informe(importados, descartados,
                    (System.nanoTime() - inicio) / 1_000_000);
        }
    }

    public static void main(String[] args) {

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

        TareaImportacion tarea = new TareaImportacion(Path.of("datos/inventario.csv"));
        Future<Informe> futuro = ejecutor.submit(tarea);

        try {
            // 1-2. El hilo principal sigue vivo: atiende el menu y pinta progreso.
            System.out.println("[main] importacion lanzada; el menu sigue activo");
            while (!futuro.isDone()) {
                System.out.printf("\r[main] progreso: %d lineas", tarea.progreso());
                TimeUnit.MILLISECONDS.sleep(300);
            }

            // 3. Plazo de 5 segundos.
            Informe informe = futuro.get(5, TimeUnit.SECONDS);
            System.out.printf("%n[main] COMPLETADA: %d importados, %d descartados, %d ms%n",
                    informe.importados(), informe.descartados(), informe.ms());

        } catch (TimeoutException e) {
            // 4. El plazo NO cancela por si solo: hay que pedirlo.
            System.out.printf("%n[main] plazo agotado tras %d lineas; cancelando%n",
                    tarea.progreso());
            boolean pedida = futuro.cancel(true);
            System.out.println("[main] cancel() devolvio  : " + pedida);
            System.out.println("[main] isCancelled()      : " + futuro.isCancelled());
            System.out.println("[main] el catalogo conserva su contenido anterior");

        } catch (ExecutionException e) {
            // 5. La tarea fallo: la causa real esta en getCause() (06-03).
            Throwable causa = e.getCause();
            if (causa instanceof InterruptedException) {
                System.out.println("\n[main] la importacion se detuvo por cancelacion");
            } else {
                LOG.log(Level.SEVERE, "La importacion fallo", causa);
                System.out.println("\n[main] fallo: " + causa.getMessage());
            }

        } catch (InterruptedException e) {
            // Nos han interrumpido a NOSOTROS; la tarea sigue viva.
            System.out.println("\n[main] espera interrumpida");
            futuro.cancel(true);
            Thread.currentThread().interrupt();

        } finally {
            // 6. Apagado en dos fases.
            ejecutor.shutdown();
            try {
                if (!ejecutor.awaitTermination(5, TimeUnit.SECONDS)) {
                    ejecutor.shutdownNow();
                    if (!ejecutor.awaitTermination(5, TimeUnit.SECONDS)) {
                        LOG.severe("El importador ignora la interrupcion");
                    }
                }
            } catch (InterruptedException e) {
                ejecutor.shutdownNow();
                Thread.currentThread().interrupt();
            }
            System.out.println("[main] ejecutor apagado; la JVM puede terminar");
        }
    }
}

[main] importacion lanzada; el menu sigue activo
[main] progreso: 41893 lineas
[main] plazo agotado tras 42017 lineas; cancelando
[main] cancel() devolvio  : true
[main] isCancelled()      : true
[main] el catalogo conserva su contenido anterior
[main] ejecutor apagado; la JVM puede terminar

package com.nexussoftware.bibliotech.servicio;

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

public class SalaDeLectura {

    private static final int AFORO = 4;
    private static final int EMPLEADOS = 20;

    // Semaforo EQUITATIVO: los que llevan mas esperando entran antes,
    // lo que evita la inanicion (08-04, apartado 12).
    private final Semaphore aforo = new Semaphore(AFORO, true);

    // Aforo instantaneo, para comprobar el invariante.
    private final AtomicInteger dentro = new AtomicInteger();
    private final AtomicInteger maximoObservado = new AtomicInteger();
    private volatile boolean invarianteRoto = false;

    /** Un empleado entra, permanece un rato y sale. */
    public void visitar(String empleado) throws InterruptedException {

        aforo.acquire();          // bloquea si ya hay 4 dentro
        try {
            int actual = dentro.incrementAndGet();

            // Registro del maximo observado, con bucle CAS (08-06).
            maximoObservado.updateAndGet(m -> Math.max(m, actual));

            if (actual > AFORO) {
                invarianteRoto = true;
                System.err.println("!!! INVARIANTE ROTO: " + actual + " dentro !!!");
            }

            System.out.printf("  [+] %-12s entra  (dentro=%d, esperando=%d)%n",
                    empleado, actual, aforo.getQueueLength());

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

            System.out.printf("  [-] %-12s sale   (dentro=%d)%n",
                    empleado, dentro.get() - 1);

        } finally {
            dentro.decrementAndGet();
            // OBLIGATORIO en finally: un permiso perdido no vuelve nunca,
            // y tras cuatro perdidas la sala quedaria cerrada para siempre.
            aforo.release();
        }
    }

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

        SalaDeLectura sala = new SalaDeLectura();

        // Dos cierres: la "pistola de salida" y la "meta".
        CountDownLatch salida = new CountDownLatch(1);
        CountDownLatch meta   = new CountDownLatch(EMPLEADOS);

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

        monitor.scheduleAtFixedRate(() -> {
            try {
                System.out.printf("      [monitor] aforo=%d/%d  cola=%d%n",
                        sala.dentro.get(), AFORO, sala.aforo.getQueueLength());
            } catch (Exception e) {
                // Una excepcion no capturada CANCELARIA la planificacion
                // para siempre y en silencio.
                e.printStackTrace();
            }
        }, 200, 200, TimeUnit.MILLISECONDS);

        // 20 hilos que esperan la salida y arrancan a la vez.
        for (int i = 1; i <= EMPLEADOS; i++) {
            final String nombre = "empleado-" + String.format("%02d", i);
            new Thread(() -> {
                try {
                    salida.await();            // todos esperan aqui
                    sala.visitar(nombre);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                } finally {
                    meta.countDown();          // SIEMPRE
                }
            }, nombre).start();
        }

        System.out.println("=== SALA DE LECTURA (aforo " + AFORO + ") ===");
        TimeUnit.MILLISECONDS.sleep(200);      // que todos lleguen a await()

        long inicio = System.nanoTime();
        salida.countDown();                    // ¡a la vez!
        meta.await();
        long ms = (System.nanoTime() - inicio) / 1_000_000;

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

        System.out.println();
        System.out.println("=== RESUMEN ===");
        System.out.println("Empleados atendidos : " + EMPLEADOS);
        System.out.println("Aforo maximo         : " + sala.maximoObservado.get()
                + " (limite " + AFORO + ")");
        System.out.println("Invariante respetado : " + !sala.invarianteRoto);
        System.out.println("Tiempo total         : " + ms + " ms");
        System.out.println("Dentro al final      : " + sala.dentro.get());
        System.out.println("Permisos disponibles : " + sala.aforo.availablePermits()
                + " (debe ser " + AFORO + ")");
    }
}

=== SALA DE LECTURA (aforo 4) ===
  [+] empleado-03  entra  (dentro=1, esperando=16)
  [+] empleado-07  entra  (dentro=2, esperando=15)
  [+] empleado-01  entra  (dentro=3, esperando=14)
  [+] empleado-12  entra  (dentro=4, esperando=13)
      [monitor] aforo=4/4  cola=16
  [-] empleado-03  sale   (dentro=3)
  [+] empleado-05  entra  (dentro=4, esperando=15)
  ...

=== RESUMEN ===
Empleados atendidos : 20
Aforo maximo         : 4 (limite 4)
Invariante respetado : true
Tiempo total         : 2143 ms
Dentro al final      : 0
Permisos disponibles : 4 (debe ser 4)