Módulo 07: Concurrencia moderna
Objetivos del módulo
- Sustituir la creación manual de hilos por pools gestionados con
ExecutoryExecutorService - Lanzar tareas con resultado (
Callable) y recogerlo conFuture - Ejecutar grupos de tareas con
invokeAlleinvokeAnyy programarlas conScheduledExecutorService - Apagar correctamente un executor
- Componer operaciones asíncronas con
CompletableFuture:thenApply,thenCompose,thenCombine,allOf,exceptionally,handle… - Usar colecciones concurrentes:
ConcurrentHashMap,CopyOnWriteArrayListyBlockingQueue - Coordinar hilos con sincronizadores:
CountDownLatch,CyclicBarrierySemaphore - Paralelizar algoritmos de divide y vencerás con fork/join y
RecursiveTask - Entender los hilos virtuales de Java 21, medir su efecto y conocer el problema del anclaje (pinning)
- Probar la concurrencia estructurada (
StructuredTaskScope, preview en Java 25) y losScopedValue(definitivos en Java 25)
1. Del Thread al Executor
En el módulo 06 creaste hilos con new Thread(...). Funciona, pero en una aplicación real plantea problemas:
- Coste: un hilo de plataforma es un hilo del sistema operativo; crearlo y destruirlo por cada tarea es caro.
- Sin límite: si llegan 10 000 peticiones y creas 10 000 hilos, la máquina se queda sin memoria.
- Sin resultado:
Runnable.run()devuelvevoidy no puede lanzar excepciones comprobadas. - Gestión manual: tienes que guardar los hilos, hacer
join, controlar errores…
Desde Java 5, el paquete java.util.concurrent separa qué hay que hacer (la tarea) de cómo se ejecuta (el executor). Es como pasar de contratar a un cocinero para cada plato a tener una cocina con una plantilla fija y una comanda de pedidos.
Figura 1. Un ExecutorService con un pool de tres hilos que toman tareas de una cola.
| Interfaz | Responsabilidad |
|---|---|
Executor | Un único método: execute(Runnable) |
ExecutorService | Añade submit, invokeAll, invokeAny, shutdown, close… |
ScheduledExecutorService | Añade tareas con retardo o periódicas |
Callable<V> | Tarea con resultado: V call() throws Exception |
Future<V> | Resultado pendiente: get(), cancel(), isDone(), state()… |
Fábricas de Executors
| Método | Hilos | Uso típico |
|---|---|---|
newFixedThreadPool(n) | n fijos, cola ilimitada | Trabajo de CPU (n ≈ número de núcleos) o recursos limitados |
newCachedThreadPool() | Crea según demanda, reutiliza los libres 60 s | Muchas tareas cortas (con cuidado: no tiene límite) |
newSingleThreadExecutor() | 1 | Tareas que deben ejecutarse en orden, de una en una |
newScheduledThreadPool(n) | n | Tareas programadas y periódicas |
newWorkStealingPool() | Un ForkJoinPool | Tareas que generan subtareas |
newVirtualThreadPerTaskExecutor() (21) | Un hilo virtual nuevo por tarea | Muchas tareas que esperan E/S (sección 8) |
💡 Consejo: En aplicaciones exigentes se crea directamente un
ThreadPoolExecutor(nucleo, maximo, tiempoVida, unidad, cola, fabricaHilos, politicaRechazo)para controlar el tamaño de la cola (una cola acotada evita quedarse sin memoria) y qué hacer cuando está llena (CallerRunsPolicy,AbortPolicy…). Las fábricas deExecutorsson atajos con valores por defecto.
2. ExecutorService, Callable y Future
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
public class Pools {
static void dormir(long ms) {
try {
Thread.sleep(ms);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
public static void main(String[] args) throws Exception {
// Un pool de 3 hilos reutilizables para 6 tareas
ExecutorService pool = Executors.newFixedThreadPool(3);
for (int i = 1; i <= 6; i++) {
int n = i;
pool.execute(() -> { // execute(Runnable): sin resultado
System.out.println("Tarea " + n + " en " + Thread.currentThread().getName());
dormir(200);
});
}
// Callable: una tarea que DEVUELVE un resultado (y puede lanzar excepciones)
Callable<Integer> calculo = () -> {
dormir(300);
return 6 * 7;
};
Future<Integer> futuro = pool.submit(calculo); // submit devuelve un Future
System.out.println("¿Terminado ya? " + futuro.isDone());
System.out.println("Resultado (get bloquea hasta tenerlo): " + futuro.get());
// get con tiempo máximo
Future<String> lento = pool.submit(() -> {
dormir(2000);
return "por fin";
});
try {
lento.get(500, TimeUnit.MILLISECONDS);
} catch (TimeoutException e) {
System.out.println("No llegó en 500 ms: cancelo -> " + lento.cancel(true) + ", cancelado=" + lento.isCancelled());
}
// Las excepciones de la tarea llegan envueltas en ExecutionException
Future<Integer> fallo = pool.submit(() -> Integer.parseInt("doce"));
try {
fallo.get();
} catch (ExecutionException e) {
System.out.println("La tarea falló: " + e.getCause());
}
System.out.println("Estado del futuro fallido: " + fallo.state()); // Future.state() es de Java 19
// Apagado correcto en dos fases
pool.shutdown(); // no acepta tareas nuevas, termina las pendientes
if (!pool.awaitTermination(5, TimeUnit.SECONDS)) {
List<Runnable> pendientes = pool.shutdownNow(); // interrumpe a las que siguen
System.out.println("Forzado el cierre; no se ejecutaron " + pendientes.size());
}
System.out.println("Pool terminado: " + pool.isTerminated());
// Desde Java 19, ExecutorService es AutoCloseable: close() espera a que acaben las tareas
List<Future<String>> resultados = new ArrayList<>();
try (ExecutorService otro = Executors.newFixedThreadPool(2)) {
for (String palabra : List.of("hola", "mundo", "java")) {
resultados.add(otro.submit(() -> palabra.toUpperCase()));
}
} // aquí se llama a close()
for (Future<String> f : resultados) {
System.out.print(f.resultNow() + " "); // resultNow() (Java 19): sin bloquear ni checked exceptions
}
System.out.println();
}
}
docker run --rm -v "$PWD:/w" -w /w maven:3.9-eclipse-temurin-25 java Pools.java
Tarea 2 en pool-1-thread-2
Tarea 3 en pool-1-thread-3
Tarea 1 en pool-1-thread-1
¿Terminado ya? false
Tarea 4 en pool-1-thread-3
Tarea 5 en pool-1-thread-1
Tarea 6 en pool-1-thread-2
Resultado (get bloquea hasta tenerlo): 42
No llegó en 500 ms: cancelo -> true, cancelado=true
La tarea falló: java.lang.NumberFormatException: For input string: "doce"
Estado del futuro fallido: FAILED
Pool terminado: true
HOLA MUNDO JAVA
Fíjate en que las seis tareas se ejecutaron con solo tres hilos (pool-1-thread-1 a -3): cuando un hilo termina una tarea, coge la siguiente de la cola. El orden de las líneas varía entre ejecuciones.
Future: el resguardo del resultado
| Método | Qué hace |
|---|---|
get() | Bloquea hasta tener el resultado; lanza ExecutionException si la tarea falló |
get(t, unidad) | Igual, con tiempo máximo: lanza TimeoutException |
isDone() | ¿Ha terminado (bien, mal o cancelada)? |
cancel(interrumpir) | Intenta cancelar; con true interrumpe el hilo si ya se ejecuta |
state() (19) | RUNNING, SUCCESS, FAILED o CANCELLED |
resultNow() / exceptionNow() (19) | Resultado o excepción de una tarea ya terminada, sin bloquear |
⚠️ Importante: Si envías una tarea con
submity nunca llamas aget(), sus excepciones desaparecen en silencio: quedan guardadas en elFutureque nadie consulta. Conexecute, en cambio, la excepción se imprime por el manejador de excepciones no capturadas del hilo. Revisa siempre losFutureo captura las excepciones dentro de la tarea.
Apagar un executor correctamente
Los hilos de un pool no son daemon: si no apagas el executor, la JVM no termina nunca. El patrón recomendado por la documentación de Java es en dos fases:
pool.shutdown(); // 1. no admite tareas nuevas
try {
if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
pool.shutdownNow(); // 2. interrumpe a las rezagadas
if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
System.err.println("El pool no terminó");
}
}
} catch (InterruptedException e) {
pool.shutdownNow();
Thread.currentThread().interrupt();
}
💡 Consejo: Desde Java 19,
ExecutorServiceimplementaAutoCloseable. Untry (ExecutorService pool = ...) { ... }llama aclose()al salir, que haceshutdown()y espera a que terminen todas las tareas. Es la forma más cómoda y segura para executors de vida corta, y la usaremos en el resto del módulo.
3. invokeAll e invokeAny
Cuando tienes un grupo de tareas (una colección de Callable), puedes lanzarlas todas de una vez:
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
public class InvocarTodas {
// Simula consultar el precio de un vuelo en una agencia que tarda 'ms' milisegundos
static Callable<String> agencia(String nombre, int precio, long ms) {
return () -> {
Thread.sleep(ms);
if (precio < 0) throw new IllegalStateException(nombre + " no responde");
return nombre + ": " + precio + " €";
};
}
public static void main(String[] args) throws Exception {
List<Callable<String>> consultas = List.of(
agencia("Viajes Sol", 210, 600),
agencia("Vuela Ya", 185, 300),
agencia("Low Cost", -1, 100), // esta falla
agencia("Air Plus", 240, 900));
try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
// invokeAll: lanza todas y espera a que terminen TODAS
long t0 = System.nanoTime();
List<Future<String>> todas = pool.invokeAll(consultas);
System.out.println("invokeAll tardó " + (System.nanoTime() - t0) / 1_000_000 + " ms");
for (Future<String> f : todas) {
switch (f.state()) {
case SUCCESS -> System.out.println(" OK " + f.resultNow());
case FAILED -> System.out.println(" FALLO " + f.exceptionNow().getMessage());
default -> System.out.println(" " + f.state());
}
}
// invokeAny: devuelve el resultado de la PRIMERA que termine bien y cancela el resto
t0 = System.nanoTime();
String primera = pool.invokeAny(consultas);
System.out.println("invokeAny tardó " + (System.nanoTime() - t0) / 1_000_000 + " ms -> " + primera);
}
}
}
invokeAll tardó 941 ms
OK Viajes Sol: 210 €
OK Vuela Ya: 185 €
FALLO Low Cost no responde
OK Air Plus: 240 €
invokeAny tardó 374 ms -> Vuela Ya: 185 €
invokeAlltarda lo que la tarea más lenta (unos 900 ms), no la suma (1900 ms). Devuelve losFutureen el mismo orden que las tareas, todos ya terminados.invokeAnydevuelve el resultado de la primera que termina con éxito (la de 100 ms falló, así que ganó la de 300 ms) y cancela las demás. Solo lanza excepción si fallan todas.- El
switchsobref.state()(Java 19 + switch expressions) es una forma elegante de distinguir éxitos y fallos.
4. Tareas programadas: ScheduledExecutorService
Para ejecutar algo dentro de un rato o cada cierto tiempo (latidos, limpieza de cachés, copias de seguridad, recordatorios), olvídate de Thread.sleep en un bucle y de la antigua clase java.util.Timer:
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class Programadas {
static final long INICIO = System.nanoTime();
static void log(String msg) {
System.out.printf("%5d ms %s%n", (System.nanoTime() - INICIO) / 1_000_000, msg);
}
public static void main(String[] args) throws Exception {
ScheduledExecutorService prog = Executors.newScheduledThreadPool(2);
// 1. Una sola vez, con retardo
prog.schedule(() -> log("recordatorio a los 500 ms"), 500, TimeUnit.MILLISECONDS);
// 2. Tasa fija: cada 300 ms medidos desde el INICIO de cada ejecución
AtomicInteger latidos = new AtomicInteger();
ScheduledFuture<?> latido = prog.scheduleAtFixedRate(
() -> log("latido " + latidos.incrementAndGet()), 0, 300, TimeUnit.MILLISECONDS);
// 3. Retardo fijo: 300 ms desde el FINAL de cada ejecución (la tarea tarda 200 ms)
AtomicInteger copias = new AtomicInteger();
ScheduledFuture<?> copia = prog.scheduleWithFixedDelay(() -> {
log(" copia de seguridad " + copias.incrementAndGet() + " empieza");
try {
Thread.sleep(200);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}, 100, 300, TimeUnit.MILLISECONDS);
Thread.sleep(1300);
latido.cancel(false); // las periódicas se paran cancelando su ScheduledFuture
copia.cancel(false);
prog.shutdown();
prog.awaitTermination(1, TimeUnit.SECONDS);
log("fin: " + latidos.get() + " latidos y " + copias.get() + " copias");
}
}
405 ms latido 1
645 ms copia de seguridad 1 empieza
895 ms latido 2
898 ms recordatorio a los 500 ms
1042 ms latido 3
1284 ms latido 4
1418 ms copia de seguridad 2 empieza
1583 ms latido 5
1895 ms latido 6
1932 ms fin: 6 latidos y 2 copias
Los tiempos están medidos desde que se cargó la clase e incluyen el arranque de la JVM y una máquina muy cargada, por eso el primer latido no aparece en el milisegundo 0 y hay cierta deriva. Lo importante es el patrón: los latidos van cada ~300 ms (si uno se retrasa, scheduleAtFixedRate intenta recuperar el ritmo), mientras que entre el inicio de dos copias pasan ~500 ms (200 ms de trabajo + 300 ms de espera).
| Método | Cuándo se repite |
|---|---|
schedule(tarea, retardo, unidad) | Una vez, tras el retardo (admite Callable) |
scheduleAtFixedRate(t, inicial, periodo, u) | Cada periodo desde el inicio de la anterior (frecuencia constante) |
scheduleWithFixedDelay(t, inicial, retardo, u) | retardo después del final de la anterior (pausa constante) |
⚠️ Importante: Si una ejecución de una tarea periódica lanza una excepción, las siguientes se cancelan en silencio. Envuelve el cuerpo de las tareas periódicas en un
try/catchque registre el error.
5. CompletableFuture: programación asíncrona
Future tiene una limitación: para usar el resultado hay que bloquearse con get(). CompletableFuture (Java 8) permite describir qué hacer cuando llegue el resultado, encadenando etapas sin bloquear, como una tubería. Es el equivalente Java de las promesas de JavaScript.
Figura 2. Composición de operaciones asíncronas con CompletableFuture.
💻 Ejemplo: la ficha de un producto de una tienda online necesita consultar varios servicios lentos. Algunos dependen de otros y otros son independientes.
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class Futuros {
static final long INICIO = System.nanoTime();
static void log(String msg) {
System.out.printf("%5d ms [%-28s] %s%n", (System.nanoTime() - INICIO) / 1_000_000,
Thread.currentThread().getName(), msg);
}
static void dormir(long ms) {
try {
Thread.sleep(ms);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
// Servicios simulados de una tienda online
record Cliente(int id, String nombre, boolean vip) { }
static Cliente buscarCliente(int id) {
log("buscando cliente " + id);
dormir(300);
return new Cliente(id, "Ana", true);
}
static double precioBase(String producto) {
log("consultando precio de " + producto);
dormir(400);
return 120.0;
}
static int stock(String producto) {
log("consultando stock de " + producto);
dormir(500);
return 7;
}
static double gastosEnvio(String cp) {
log("calculando envío a " + cp);
if (cp.isBlank()) throw new IllegalArgumentException("código postal vacío");
dormir(200);
return 4.95;
}
public static void main(String[] args) {
try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
// supplyAsync + thenApply: transformar el resultado cuando llegue
CompletableFuture<String> saludo = CompletableFuture
.supplyAsync(() -> buscarCliente(1), pool)
.thenApply(c -> "Hola, " + c.nombre());
log(saludo.join());
// thenCompose: encadenar otra operación ASÍNCRONA que depende del resultado
CompletableFuture<Double> descuento = CompletableFuture
.supplyAsync(() -> buscarCliente(2), pool)
.thenCompose(c -> CompletableFuture.supplyAsync(() -> c.vip() ? 0.10 : 0.0, pool));
log("descuento del cliente 2: " + descuento.join());
// thenCombine: dos operaciones INDEPENDIENTES en paralelo y combinar sus resultados
long t0 = System.nanoTime();
CompletableFuture<Double> precio = CompletableFuture.supplyAsync(() -> precioBase("portátil"), pool);
CompletableFuture<Integer> unidades = CompletableFuture.supplyAsync(() -> stock("portátil"), pool);
String ficha = precio.thenCombine(unidades, (p, u) -> "portátil a " + p + " € (" + u + " en stock)").join();
log(ficha + " en " + (System.nanoTime() - t0) / 1_000_000 + " ms (no 900)");
// allOf: esperar a varias; anyOf: a la primera
List<CompletableFuture<Double>> envios = List.of("03001", "28080", "46001").stream()
.map(cp -> CompletableFuture.supplyAsync(() -> gastosEnvio(cp), pool))
.toList();
CompletableFuture.allOf(envios.toArray(new CompletableFuture[0])).join();
double totalEnvios = envios.stream().mapToDouble(CompletableFuture::join).sum();
log("suma de los 3 envíos: " + totalEnvios);
// exceptionally: valor alternativo si algo falla
double envio = CompletableFuture.supplyAsync(() -> gastosEnvio(" "), pool)
.exceptionally(ex -> {
log("fallo: " + ex.getMessage() + " -> uso tarifa plana");
return 9.99;
})
.join();
log("envío aplicado: " + envio);
// handle: recibe resultado O excepción, y siempre se ejecuta
String estado = CompletableFuture.supplyAsync(() -> gastosEnvio("03001"), pool)
.handle((valor, ex) -> ex == null ? "envío OK: " + valor : "envío KO: " + ex.getMessage())
.join();
log(estado);
// Tiempo máximo (Java 9): completeOnTimeout da un valor por defecto, orTimeout falla
String lento = CompletableFuture.supplyAsync(() -> { dormir(2000); return "tarde"; }, pool)
.completeOnTimeout("valor por defecto", 300, TimeUnit.MILLISECONDS)
.join();
log("con completeOnTimeout: " + lento);
// thenAccept / thenRun: consumir sin devolver nada
CompletableFuture.supplyAsync(() -> buscarCliente(3), pool)
.thenAccept(c -> log("enviando email a " + c.nombre()))
.thenRun(() -> log("pedido completado"))
.join();
pool.shutdownNow(); // corta la tarea de 2 s que quedó en marcha
}
}
}
483 ms [pool-1-thread-1 ] buscando cliente 1
1085 ms [main ] Hola, Ana
1142 ms [pool-1-thread-2 ] buscando cliente 2
1527 ms [main ] descuento del cliente 2: 0.1
1639 ms [pool-1-thread-4 ] consultando precio de portátil
1647 ms [pool-1-thread-1 ] consultando stock de portátil
2543 ms [main ] portátil a 120.0 € (7 en stock) en 767 ms (no 900)
2626 ms [pool-1-thread-2 ] calculando envío a 28080
2631 ms [pool-1-thread-3 ] calculando envío a 03001
2633 ms [pool-1-thread-4 ] calculando envío a 46001
3238 ms [main ] suma de los 3 envíos: 14.850000000000001
3309 ms [pool-1-thread-1 ] calculando envío a
3401 ms [main ] fallo: java.lang.IllegalArgumentException: código postal vacío -> uso tarifa plana
3402 ms [main ] envío aplicado: 9.99
3406 ms [pool-1-thread-2 ] calculando envío a 03001
3691 ms [main ] envío OK: 4.95
4318 ms [main ] con completeOnTimeout: valor por defecto
4485 ms [pool-1-thread-3 ] buscando cliente 3
4791 ms [pool-1-thread-3 ] enviando email a Ana
4795 ms [pool-1-thread-3 ] pedido completado
Observa en la salida real:
- Precio y stock se consultaron a la vez (milisegundos 1639 y 1647): la ficha tardó unos 770 ms, cerca de los 500 ms de la consulta más lenta y lejos de los 900 ms de hacerlas una detrás de otra. Lo mismo con los tres envíos.
- Las etapas
thenApply,exceptionallyothenAcceptse ejecutan en el hilo que completó la etapa anterior, o en el que llama si ya estaba completa (fíjate en queexceptionallyaparece enmain). Las variantes...Async(thenApplyAsync(f, pool)) fuerzan su ejecución en un executor. 14.850000000000001: recuerda quedoubleno es adecuado para dinero (usaBigDecimal).- Los tiempos absolutos son de una máquina muy cargada; en la tuya serán menores, pero las proporciones se mantienen.
Las operaciones principales
| Método | Recibe | Equivale a… | Uso |
|---|---|---|---|
supplyAsync(supplier, executor) | Supplier<T> | Lanzar una tarea | Iniciar la cadena con un valor |
runAsync(runnable, executor) | Runnable | Tarea sin resultado | Iniciar sin valor |
thenApply(f) | T -> U | map | Transformar el valor |
thenCompose(f) | T -> CompletableFuture<U> | flatMap | Encadenar otra operación asíncrona dependiente |
thenCombine(otro, f) | (T, U) -> V | — | Unir dos futuros independientes |
thenAccept(c) / thenRun(r) | Consumer<T> / Runnable | — | Consumir el resultado al final |
allOf(cf...) / anyOf(cf...) | Varios futuros | — | Esperar a todos / al primero |
exceptionally(f) | Throwable -> T | catch | Valor alternativo si falla |
handle(f) | (T, Throwable) -> U | finally con resultado | Tratar éxito y fallo a la vez |
whenComplete(c) | (T, Throwable) | — | Efecto secundario (p. ej. log) sin cambiar el resultado |
orTimeout(t, u) / completeOnTimeout(v, t, u) | — | — | Tiempo máximo: fallar o valor por defecto (Java 9) |
join() / get() | — | — | Esperar el resultado (join lanza excepciones no comprobadas) |
⚠️ Importante: Si no pasas un executor,
supplyAsyncusa elForkJoinPool.commonPool(), compartido por toda la JVM y pensado para cálculo (tiene tantos hilos como núcleos menos uno). Si tus tareas bloquean (E/S,sleep, consultas a BD), pásale tu propio executor o uno de hilos virtuales; si no, puedes agotar el pool común y frenar a todo el programa (incluidos los streams paralelos).
💡 Consejo: Dentro de una cadena, las excepciones llegan envueltas en
CompletionException; usaex.getCause()para obtener la original. Y no mezclesjoin()en mitad de la cadena: solo al final, cuando de verdad necesitas el valor.
6. Colecciones concurrentes
Las colecciones de java.util (ArrayList, HashMap…) no son seguras entre hilos. Las versiones “sincronizadas” (Collections.synchronizedMap(...), Vector, Hashtable) bloquean la colección entera en cada operación y escalan mal. El paquete java.util.concurrent ofrece alternativas diseñadas para la concurrencia.
import java.util.ArrayList;
import java.util.ConcurrentModificationException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class Colecciones {
public static void main(String[] args) throws Exception {
// 1. HashMap compartido sin protección frente a ConcurrentHashMap
Map<Integer, Integer> normal = new HashMap<>();
Map<Integer, Integer> concurrente = new ConcurrentHashMap<>();
try (ExecutorService pool = Executors.newFixedThreadPool(8)) {
for (int h = 0; h < 8; h++) {
pool.submit(() -> {
for (int i = 0; i < 100_000; i++) {
int clave = i % 100;
normal.merge(clave, 1, Integer::sum); // carrera
concurrente.merge(clave, 1, Integer::sum); // atómico
}
});
}
}
int sumaNormal = normal.values().stream().mapToInt(Integer::intValue).sum();
int sumaConc = concurrente.values().stream().mapToInt(Integer::intValue).sum();
System.out.printf("HashMap: %,d claves=%d | ConcurrentHashMap: %,d claves=%d (esperado 800.000 y 100)%n",
sumaNormal, normal.size(), sumaConc, concurrente.size());
// Operaciones atómicas útiles de ConcurrentHashMap
ConcurrentHashMap<String, Integer> stock = new ConcurrentHashMap<>(Map.of("ratón", 5, "teclado", 2));
stock.putIfAbsent("monitor", 1);
stock.computeIfPresent("teclado", (k, v) -> v > 0 ? v - 1 : v);
stock.compute("cable", (k, v) -> v == null ? 10 : v + 10);
System.out.println("Stock: " + new java.util.TreeMap<>(stock));
System.out.println("Total de unidades (reduce en paralelo): " + stock.reduceValues(1, Integer::sum));
// 2. Modificar un ArrayList mientras se recorre: ConcurrentModificationException
List<String> oyentes = new ArrayList<>(List.of("A", "B", "C"));
try {
for (String o : oyentes) {
if (o.equals("B")) oyentes.add("D");
}
} catch (ConcurrentModificationException e) {
System.out.println("ArrayList: " + e.getClass().getSimpleName());
}
// CopyOnWriteArrayList: cada escritura copia el array; los iteradores ven una "foto"
List<String> cow = new CopyOnWriteArrayList<>(List.of("A", "B", "C"));
for (String o : cow) {
if (o.equals("B")) cow.add("D");
System.out.print(o + " ");
}
System.out.println("-> tras el bucle: " + cow);
// 3. BlockingQueue: productor-consumidor sin wait/notify
BlockingQueue<String> cola = new ArrayBlockingQueue<>(2);
Thread consumidor = Thread.ofPlatform().name("consumidor").start(() -> {
try {
while (true) {
String pedido = cola.take(); // bloquea si está vacía
if (pedido.equals("FIN")) break;
System.out.println(" procesado " + pedido);
Thread.sleep(100);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
for (int i = 1; i <= 4; i++) {
cola.put("pedido-" + i); // bloquea si está llena
System.out.println("encolado pedido-" + i + " (en cola: " + cola.size() + ")");
}
cola.put("FIN");
consumidor.join();
System.out.println("offer con cola vacía y capacidad: " + cola.offer("x") + ", poll con timeout: "
+ cola.poll(100, TimeUnit.MILLISECONDS) + ", poll sobre vacía: " + cola.poll(100, TimeUnit.MILLISECONDS));
}
}
HashMap: 798.410 claves=100 | ConcurrentHashMap: 800.000 claves=100 (esperado 800.000 y 100)
Stock: {cable=10, monitor=1, ratón=5, teclado=1}
Total de unidades (reduce en paralelo): 17
ArrayList: ConcurrentModificationException
A B C -> tras el bucle: [A, B, C, D]
procesado pedido-1
encolado pedido-1 (en cola: 1)
encolado pedido-2 (en cola: 1)
encolado pedido-3 (en cola: 2)
procesado pedido-2
encolado pedido-4 (en cola: 2)
procesado pedido-3
procesado pedido-4
offer con cola vacía y capacidad: true, poll con timeout: x, poll sobre vacía: null
El HashMap compartido perdió 1590 incrementos (en otras ejecuciones puede incluso corromperse internamente o entrar en un bucle infinito); el ConcurrentHashMap dio exactamente 800 000. Con BlockingQueue, el productor-consumidor del módulo anterior se reduce a put y take: la cola ya contiene los cerrojos y las condiciones.
ConcurrentHashMap
- Permite lecturas sin bloqueo y escrituras concurrentes en distintas partes de la tabla.
- Sus operaciones compuestas son atómicas:
putIfAbsent,computeIfAbsent,computeIfPresent,compute,merge,replace(k, viejo, nuevo). Úsalas en lugar deget+put, que sería un “comprobar y actuar” con carrera. - Ofrece operaciones masivas en paralelo:
forEach,search,reduce(el primer parámetro es el umbral de paralelismo). - No admite claves ni valores
null.
⚠️ Importante:
if (!mapa.containsKey(k)) mapa.put(k, v);no es atómico aunque el mapa sea unConcurrentHashMap: entre las dos llamadas otro hilo puede insertar la clave. Escribemapa.putIfAbsent(k, v)omapa.computeIfAbsent(k, clave -> crear(clave)).
Resumen de colecciones concurrentes
| Colección | Características | Uso típico |
|---|---|---|
ConcurrentHashMap | Mapa concurrente de alto rendimiento | Cachés, contadores por clave |
ConcurrentSkipListMap / Set | Ordenados, concurrentes | Rankings, índices ordenados |
CopyOnWriteArrayList / Set | Copia el array en cada escritura; iteración sin bloqueo | Listas de oyentes (listeners): muchas lecturas, pocas escrituras |
ConcurrentLinkedQueue / Deque | Cola sin bloqueo (no espera) | Colas de trabajo sin límite |
ArrayBlockingQueue | Cola bloqueante acotada (array) | Productor-consumidor con contrapresión |
LinkedBlockingQueue | Bloqueante, opcionalmente acotada | Cola de los executors |
PriorityBlockingQueue | Bloqueante con prioridad | Tareas con prioridades |
DelayQueue | Los elementos salen cuando vence su retardo | Caducidad, reintentos |
SynchronousQueue | Capacidad 0: entrega en mano | Traspaso directo entre hilos |
Operación de BlockingQueue | Si no se puede… |
|---|---|
put(e) / take() | Espera |
offer(e, t, u) / poll(t, u) | Espera como mucho t y devuelve false / null |
offer(e) / poll() | Devuelve false / null inmediatamente |
add(e) / remove() | Lanza excepción |
7. Sincronizadores
Son objetos que coordinan el avance de varios hilos sin que tengas que escribir wait/notify.
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class Sincronizadores {
static void dormir(long ms) {
try {
Thread.sleep(ms);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
public static void main(String[] args) throws Exception {
// 1. CountDownLatch: esperar a que N servicios estén listos antes de abrir la tienda
String[] servicios = {"base de datos", "caché", "pasarela de pago"};
CountDownLatch listos = new CountDownLatch(servicios.length);
try (ExecutorService pool = Executors.newFixedThreadPool(3)) {
for (int i = 0; i < servicios.length; i++) {
String s = servicios[i];
long ms = 100L * (i + 1);
pool.submit(() -> {
dormir(ms);
System.out.println(" " + s + " lista");
listos.countDown(); // resta 1
});
}
listos.await(); // espera a que llegue a 0
System.out.println("Tienda abierta: todos los servicios arrancados");
}
// 2. CyclicBarrier: 3 corredores esperan a los demás al final de cada etapa
CyclicBarrier meta = new CyclicBarrier(3, () -> System.out.println(" --- etapa completada por todos ---"));
try (ExecutorService pool = Executors.newFixedThreadPool(3)) {
for (String corredor : new String[]{"Ana", "Luis", "Eva"}) {
pool.submit(() -> {
for (int etapa = 1; etapa <= 2; etapa++) {
dormir((long) (Math.random() * 300));
System.out.println(corredor + " termina la etapa " + etapa);
try {
meta.await(); // espera a los otros dos; la barrera se reutiliza
} catch (InterruptedException | BrokenBarrierException e) {
return;
}
}
});
}
}
// 3. Semaphore: como mucho 2 descargas simultáneas
Semaphore conexiones = new Semaphore(2);
AtomicInteger activas = new AtomicInteger();
AtomicInteger maximo = new AtomicInteger();
try (ExecutorService pool = Executors.newFixedThreadPool(6)) {
for (int i = 1; i <= 6; i++) {
int n = i;
pool.submit(() -> {
try {
conexiones.acquire(); // pide un permiso (espera si no hay)
try {
int a = activas.incrementAndGet();
maximo.accumulateAndGet(a, Math::max);
System.out.println("descarga " + n + " empieza (activas: " + a + ")");
dormir(200);
activas.decrementAndGet();
} finally {
conexiones.release(); // devuelve el permiso
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
}
}
System.out.println("Máximo de descargas simultáneas: " + maximo.get());
System.out.println("tryAcquire sin esperar: " + conexiones.tryAcquire(0, TimeUnit.SECONDS)
+ ", permisos libres ahora: " + conexiones.availablePermits());
}
}
base de datos lista
caché lista
pasarela de pago lista
Tienda abierta: todos los servicios arrancados
Ana termina la etapa 1
Luis termina la etapa 1
Eva termina la etapa 1
--- etapa completada por todos ---
Ana termina la etapa 2
Luis termina la etapa 2
Eva termina la etapa 2
--- etapa completada por todos ---
descarga 2 empieza (activas: 2)
descarga 1 empieza (activas: 1)
descarga 3 empieza (activas: 2)
descarga 4 empieza (activas: 2)
descarga 5 empieza (activas: 2)
descarga 6 empieza (activas: 2)
Máximo de descargas simultáneas: 2
tryAcquire sin esperar: true, permisos libres ahora: 1
| Sincronizador | Analogía | Funcionamiento | ¿Reutilizable? |
|---|---|---|---|
CountDownLatch(n) | Cuenta atrás de un lanzamiento | countDown() resta; await() espera a que llegue a 0 | No |
CyclicBarrier(n, accion) | Grupo de excursión que espera a todos en cada parada | Cada hilo llama a await(); cuando llegan los n, se ejecuta la acción y siguen todos | Sí |
Semaphore(n) | Aparcamiento con n plazas | acquire() toma un permiso (espera si no hay); release() lo devuelve | Sí |
Phaser | Barrera con participantes variables | Fases numeradas; los hilos se registran y se dan de baja | Sí |
Exchanger<V> | Intercambio de cromos | Dos hilos intercambian un objeto en un punto de encuentro | Sí |
💡 Consejo: El
Semaphorees la herramienta ideal para limitar el acceso a un recurso escaso: conexiones a una base de datos, llamadas a una API con límite de peticiones, licencias de un programa… Lo usarás con hilos virtuales en el ejercicio 6. Y recuerda liberar el permiso en unfinally.
8. Fork/join: divide y vencerás en paralelo
El framework fork/join (Java 7) está pensado para problemas que se pueden dividir recursivamente en subproblemas independientes: sumar un array, ordenar, recorrer un árbol de directorios, procesar una imagen… Es lo que usan por debajo los streams paralelos del módulo 04.
Figura 3. Una RecursiveTask se divide hasta un umbral; los hilos ociosos roban tareas de otros.
import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
public class ForkJoin {
// Suma de un trozo del array: si es pequeño lo suma; si no, lo divide en dos
static class Suma extends RecursiveTask<Long> {
static final int UMBRAL = 10_000;
private final long[] datos;
private final int desde, hasta;
Suma(long[] datos, int desde, int hasta) {
this.datos = datos;
this.desde = desde;
this.hasta = hasta;
}
@Override
protected Long compute() {
if (hasta - desde <= UMBRAL) { // caso base: secuencial
long s = 0;
for (int i = desde; i < hasta; i++) s += raizEntera(datos[i]);
return s;
}
int mitad = (desde + hasta) >>> 1;
Suma izquierda = new Suma(datos, desde, mitad);
Suma derecha = new Suma(datos, mitad, hasta);
izquierda.fork(); // la izquierda, a la cola (otro hilo la puede robar)
long r = derecha.compute(); // la derecha, la hago yo
return r + izquierda.join(); // espero a la izquierda
}
}
// Trabajo "pesado" por elemento para que el paralelismo compense
static long raizEntera(long x) {
return (long) Math.sqrt(Math.sqrt(x) * Math.log(x + 1));
}
static long secuencial(long[] datos) {
long s = 0;
for (long d : datos) s += raizEntera(d);
return s;
}
public static void main(String[] args) {
long[] datos = new Random(1).longs(20_000_000, 1, 1_000_000).toArray();
ForkJoinPool pool = ForkJoinPool.commonPool();
System.out.println("Paralelismo del commonPool: " + pool.getParallelism());
for (int vuelta = 1; vuelta <= 3; vuelta++) { // 3 vueltas para que actúe el JIT
long t0 = System.nanoTime();
long a = secuencial(datos);
long t1 = System.nanoTime();
long b = pool.invoke(new Suma(datos, 0, datos.length));
long t2 = System.nanoTime();
long c = Arrays.stream(datos).parallel().map(ForkJoin::raizEntera).sum(); // usa también fork/join
long t3 = System.nanoTime();
System.out.printf("Vuelta %d: secuencial %d ms | RecursiveTask %d ms | parallel stream %d ms | iguales=%b%n",
vuelta, (t1 - t0) / 1_000_000, (t2 - t1) / 1_000_000, (t3 - t2) / 1_000_000, a == b && b == c);
}
System.out.println("Robos de tareas entre hilos: " + pool.getStealCount());
}
}
Salida real en un equipo de 8 núcleos:
Paralelismo del commonPool: 7
Vuelta 1: secuencial 802 ms | RecursiveTask 246 ms | parallel stream 247 ms | iguales=true
Vuelta 2: secuencial 531 ms | RecursiveTask 156 ms | parallel stream 246 ms | iguales=true
Vuelta 3: secuencial 606 ms | RecursiveTask 211 ms | parallel stream 488 ms | iguales=true
Robos de tareas entre hilos: 139
Los tres métodos dan el mismo resultado, y las versiones paralelas son entre 2,5 y 3,5 veces más rápidas. No llegan a 7 veces porque ese equipo compartía sus núcleos con otros programas, y por eso las cifras oscilan de una vuelta a otra (en una ejecución anterior, con la máquina muy cargada, la versión secuencial tardó más de 5 segundos). La primera vuelta es más lenta porque incluye el calentamiento del JIT. Es una buena lección: mide siempre en condiciones realistas y repite las mediciones (o usa una herramienta como JMH).
Claves de fork/join
| Elemento | Significado |
|---|---|
RecursiveTask<V> | Tarea con resultado (compute() devuelve V) |
RecursiveAction | Tarea sin resultado (por ejemplo, ordenar un array in situ) |
fork() | Encola la subtarea para que la ejecute este u otro hilo |
join() | Espera el resultado de una subtarea |
invokeAll(t1, t2) | Atajo: hace fork de ambas y las espera |
| Umbral | Tamaño por debajo del cual se resuelve secuencialmente; si es muy pequeño, el coste de crear tareas supera la ganancia |
| Work stealing | Cada hilo tiene su propia cola; los ociosos roban tareas del final de las colas de otros |
💡 Consejo: El patrón
izquierda.fork(); derecha.compute(); izquierda.join();aprovecha el hilo actual para la mitad derecha en lugar de dejarlo esperando. Y no bloquees nunca (E/S,sleep) dentro de tareas fork/join: el pool tiene pocos hilos y están pensados para cálculo.
9. Hilos virtuales (Java 21)
Hasta Java 20, cada Thread era un hilo del sistema operativo (hilo de plataforma). Son caros: cada uno reserva una pila de aproximadamente 1 MB y el sistema operativo tarda en crearlos y en cambiar de uno a otro. Un servidor que dedica un hilo a cada petición (el modelo thread-per-request de Tomcat, por ejemplo) se queda sin hilos con unos pocos miles de usuarios simultáneos, aunque casi todos estén esperando a la base de datos o a la red.
La alternativa era la programación asíncrona (CompletableFuture, programación reactiva), eficiente pero difícil de escribir, leer y depurar. Los hilos virtuales (JEP 444, Java 21, fruto del Proyecto Loom) resuelven el problema de raíz.
📘 Concepto: Un hilo virtual es un
Threadgestionado por la JVM, no por el sistema operativo. Miles o millones de hilos virtuales se ejecutan sobre unos pocos hilos portadores (carrier threads, unForkJoinPoolcon tantos hilos como núcleos). Cuando un hilo virtual se bloquea (E/S,sleep,BlockingQueue.take(), un cerrojo…), la JVM lo desmonta: guarda su pila en el heap y el portador queda libre para ejecutar otro hilo virtual. Cuando la operación termina, se vuelve a montar en cualquier portador libre.
Figura 4. Hilos de plataforma frente a hilos virtuales.
Crear hilos virtuales
Thread v1 = Thread.ofVirtual().name("virtual-1").start(tarea); // con el constructor fluido
Thread v2 = Thread.startVirtualThread(tarea); // atajo
ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor(); // un hilo virtual nuevo por tarea
Demostración: 100 000 tareas que esperan
Cada tarea simula una llamada a un servicio externo que tarda 1 segundo:
import java.time.Duration;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;
public class HilosVirtuales {
// Simula una tarea de E/S: una petición a una API que tarda 1 segundo
static int tareaBloqueante(int i) {
try {
Thread.sleep(Duration.ofSeconds(1));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return i;
}
static void medir(String nombre, ExecutorService executor, int tareas) {
AtomicInteger completadas = new AtomicInteger();
long t0 = System.nanoTime();
try (executor) { // close() espera a todas las tareas
IntStream.range(0, tareas).forEach(i -> executor.submit(() -> {
tareaBloqueante(i);
completadas.incrementAndGet();
}));
}
long ms = (System.nanoTime() - t0) / 1_000_000;
System.out.printf("%-34s %,7d tareas en %,6d ms%n", nombre, completadas.get(), ms);
}
public static void main(String[] args) throws Exception {
// Crear hilos virtuales sueltos
Thread v1 = Thread.ofVirtual().name("virtual-1").start(() ->
System.out.println("Soy " + Thread.currentThread() + " ¿virtual? " + Thread.currentThread().isVirtual()));
Thread v2 = Thread.startVirtualThread(() -> System.out.println("Otro hilo virtual sin nombre: " + Thread.currentThread()));
v1.join();
v2.join();
int tareas = 100_000;
medir("Hilos virtuales (uno por tarea)", Executors.newVirtualThreadPerTaskExecutor(), tareas);
medir("Pool de 200 hilos de plataforma", Executors.newFixedThreadPool(200), 2_000);
medir("Pool de 1000 hilos de plataforma", Executors.newFixedThreadPool(1000), 10_000);
}
}
Salida real en una máquina de 8 núcleos:
Otro hilo virtual sin nombre: VirtualThread[#28]/runnable@ForkJoinPool-1-worker-2
Soy VirtualThread[#26,virtual-1]/runnable@ForkJoinPool-1-worker-1 ¿virtual? true
Hilos virtuales (uno por tarea) 100.000 tareas en 2.882 ms
Pool de 200 hilos de plataforma 2.000 tareas en 10.262 ms
Pool de 1000 hilos de plataforma 10.000 tareas en 10.937 ms
Para interpretar estas cifras, haz las cuentas:
| Configuración | Tareas | Tiempo teórico mínimo | Tiempo para 100 000 tareas |
|---|---|---|---|
| 200 hilos de plataforma | 2 000 | 2 000 / 200 = 10 tandas de 1 s = 10 s | 500 tandas = más de 8 minutos |
| 1000 hilos de plataforma | 10 000 | 10 s | 100 tandas = más de 1,5 minutos |
| Hilos virtuales | 100 000 | 1 s (todas esperan a la vez) | Medido: 2,9 s |
Con un pool de plataforma, el tiempo crece linealmente con el número de tareas porque solo hay N trabajadores; no puedes poner 100 000 hilos de plataforma (cada uno es un hilo del sistema operativo con su pila). Con hilos virtuales, las 100 000 tareas duermen a la vez y el tiempo solo depende de la CPU necesaria para crearlas y planificarlas (los 1,9 s que pasan del segundo de espera). Ese coste de CPU se nota mucho si la máquina está ocupada: en otra ejecución de este mismo programa, con el equipo saturado por otros procesos, las 100 000 tareas virtuales tardaron 23 s (y aun así, frente a más de 8 minutos con el pool de 200). Ejecútalo en tu ordenador y compara.
Fíjate también en el formato VirtualThread[#26,virtual-1]/runnable@ForkJoinPool-1-worker-1: el hilo virtual 26 está montado en ese momento sobre el portador worker-1. Los hilos virtuales no tienen nombre por defecto.
Cuándo usar (y cuándo no) hilos virtuales
| Úsalos para… | No aportan nada en… |
|---|---|
| Servidores que atienden muchas peticiones concurrentes | Cálculo intensivo (CPU): el límite son los núcleos, usa pools fijos o fork/join |
| Clientes que hacen muchas llamadas HTTP o a BD | Tareas que no bloquean |
| Código bloqueante “de toda la vida”, fácil de leer | — |
⚠️ Importante: Los hilos virtuales no se reutilizan ni se agrupan en pools: son baratos, así que se crea uno por tarea y se tira. Si necesitas limitar la concurrencia (por ejemplo, 10 conexiones a una base de datos), usa un
Semaphore, no un pool pequeño de hilos virtuales. Y cuidado conThreadLocal: con un millón de hilos, un millón de copias de cada variable (ver sección 11).
El anclaje (pinning)
Un hilo virtual está anclado a su portador cuando no puede desmontarse al bloquearse: el portador queda bloqueado con él, y si todos los portadores se anclan, el programa se para. En Java 21, 22 y 23, el caso más frecuente era bloquearse dentro de un bloque synchronized. Por eso se recomendaba sustituir synchronized por ReentrantLock en código que fuese a ejecutarse en hilos virtuales.
En Java 24 el JEP 491 rediseñó la implementación de los monitores y synchronized ya no ancla los hilos virtuales. Este programa lo demuestra:
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
// ¿Se "ancla" (pinning) un hilo virtual que duerme dentro de un bloque synchronized?
public class Anclaje {
public static void main(String[] args) {
int tareas = 200;
System.out.println("Java " + Runtime.version().feature() + ", núcleos: " + Runtime.getRuntime().availableProcessors());
long t0 = System.nanoTime();
try (ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor()) {
for (int i = 0; i < tareas; i++) {
ex.submit(() -> {
Object cerrojo = new Object(); // cada tarea su propio cerrojo: no compiten entre sí
synchronized (cerrojo) {
try {
Thread.sleep(1000); // bloqueo de 1 s DENTRO de synchronized
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
});
}
}
System.out.printf("%d tareas de 1 s con sleep dentro de synchronized: %,d ms%n",
tareas, (System.nanoTime() - t0) / 1_000_000);
}
}
Ejecutado con el JDK 25 en Docker y, para comparar, con el JDK 21 instalado en el equipo anfitrión (12 núcleos):
Java 25, núcleos: 8
200 tareas de 1 s con sleep dentro de synchronized: 1.790 ms
Java 21, núcleos: 12
200 tareas de 1 s con sleep dentro de synchronized: 17.298 ms
En Java 21, cada hilo virtual se ancló a su portador durante el segundo de espera: con 12 portadores, 200 tareas necesitaron unas 17 tandas (200 / 12 ≈ 17 s). En Java 25, los hilos se desmontan con normalidad y las 200 tareas esperan a la vez.
En Java 24 y posteriores todavía se produce anclaje en casos poco habituales: al bloquearse dentro de código nativo (JNI o la API de funciones externas) y durante la carga e inicialización de clases. Para detectarlo se usa el evento jdk.VirtualThreadPinned de JDK Flight Recorder; la antigua opción -Djdk.tracePinnedThreads ya no tiene efecto.
💡 Consejo: Si tu aplicación debe funcionar también en Java 21 (la LTS anterior), sigue siendo buena idea usar
ReentrantLocken las secciones críticas que hagan E/S.
10. Concurrencia estructurada (preview en Java 25)
Con ExecutorService y CompletableFuture, las subtareas no tienen relación con quien las lanzó. Si una falla, las otras siguen ejecutándose (y gastando recursos) aunque su resultado ya no sirva; si el hilo principal se interrumpe, las subtareas no se enteran; un olvido en el manejo de errores deja hilos “huérfanos”.
📘 Concepto: La concurrencia estructurada aplica a los hilos la misma idea que la programación estructurada aplicó al
goto: si una tarea se divide en subtareas concurrentes, todas deben empezar y terminar dentro del mismo bloque de código. El bloque es el dueño de sus subtareas: espera a que acaben, cancela las demás si una falla y propaga los errores.
Figura 5. Concurrencia estructurada: las subtareas no sobreviven al bloque try.
⚠️ Importante:
StructuredTaskScopees una API en preview en Java 25 (JEP 505, quinta preview) y sigue en preview en las versiones 26 y 27. Su API ha cambiado entre versiones: en Java 25 los ámbitos se crean conStructuredTaskScope.open(...)y unJoiner, mientras que en Java 21-24 se usaban subclases comoShutdownOnFailureque ya no existen. Para usarla hay que activar las características preview:
- Lanzamiento directo:
java --enable-preview Estructurada.java- Compilación:
javac --release 25 --enable-preview Estructurada.javay ejecución conjava --enable-preview Estructurada- En Maven, añade
<compilerArgs><arg>--enable-preview</arg></compilerArgs>almaven-compiler-pluginy--enable-previewalargLinede Surefire.
import java.time.Duration;
import java.util.concurrent.StructuredTaskScope;
import java.util.concurrent.StructuredTaskScope.Joiner;
import java.util.concurrent.StructuredTaskScope.Subtask;
// Concurrencia estructurada (JEP 505, preview en Java 25):
// java --enable-preview Estructurada.java
public class Estructurada {
record Usuario(String nombre) { }
record Pedido(int id, double total) { }
record Pagina(Usuario usuario, Pedido ultimoPedido) { }
static final long INICIO = System.nanoTime();
static void log(String msg) {
System.out.printf("%5d ms %s%n", (System.nanoTime() - INICIO) / 1_000_000, msg);
}
static Usuario buscarUsuario(int id) throws InterruptedException {
Thread.sleep(300);
log("usuario encontrado");
return new Usuario("Ana");
}
static Pedido ultimoPedido(int id, boolean fallar) throws InterruptedException {
Thread.sleep(fallar ? 100 : 500);
if (fallar) throw new IllegalStateException("servicio de pedidos caído");
log("pedido encontrado");
return new Pedido(1042, 89.90);
}
// Las dos subtareas en paralelo; si una falla, se cancela la otra y join() lanza la excepción
static Pagina cargarPagina(int id, boolean fallar) throws InterruptedException {
try (var scope = StructuredTaskScope.open()) { // Joiner por defecto: todas deben ir bien
Subtask<Usuario> usuario = scope.fork(() -> buscarUsuario(id));
Subtask<Pedido> pedido = scope.fork(() -> ultimoPedido(id, fallar));
scope.join(); // espera a ambas (o al primer fallo)
return new Pagina(usuario.get(), pedido.get());
}
}
// Carrera: el primer servidor que responda bien gana; el resto se cancela
static String consultarEspejos() throws InterruptedException {
try (var scope = StructuredTaskScope.open(Joiner.<String>anySuccessfulResultOrThrow())) {
scope.fork(() -> { Thread.sleep(400); return "espejo de Madrid"; });
scope.fork(() -> { Thread.sleep(150); return "espejo de Frankfurt"; });
scope.fork(() -> { Thread.sleep(50); throw new RuntimeException("espejo de París caído"); });
return scope.join();
}
}
// Tiempo máximo para todo el ámbito
static void conTiempoMaximo() throws InterruptedException {
try (var scope = StructuredTaskScope.open(Joiner.<String>awaitAllSuccessfulOrThrow(),
cfg -> cfg.withTimeout(Duration.ofMillis(200)))) {
scope.fork(() -> { Thread.sleep(1000); return "lento"; });
scope.join();
} catch (StructuredTaskScope.TimeoutException e) {
log("tiempo agotado: las subtareas se han cancelado");
}
}
public static void main(String[] args) throws InterruptedException {
log("Página: " + cargarPagina(7, false));
try {
cargarPagina(7, true);
} catch (StructuredTaskScope.FailedException e) {
log("Fallo al cargar la página: " + e.getCause());
}
log("Respuesta más rápida: " + consultarEspejos());
conTiempoMaximo();
}
}
docker run --rm -v "$PWD:/w" -w /w maven:3.9-eclipse-temurin-25 java --enable-preview Estructurada.java
1482 ms usuario encontrado
1919 ms pedido encontrado
3332 ms Página: Pagina[usuario=Usuario[nombre=Ana], ultimoPedido=Pedido[id=1042, total=89.9]]
3523 ms Fallo al cargar la página: java.lang.IllegalStateException: servicio de pedidos caído
3882 ms Respuesta más rápida: espejo de Frankfurt
4264 ms tiempo agotado: las subtareas se han cancelado
Si compilas con javac, verás un aviso de que el programa usa características en preview; es normal. En el segundo caso, el servicio de pedidos falló a los 100 ms y la búsqueda del usuario se canceló sin llegar a escribir “usuario encontrado”: no hay trabajo inútil ni hilos huérfanos. Las subtareas se ejecutan por defecto en hilos virtuales.
| Elemento (Java 25) | Significado |
|---|---|
StructuredTaskScope.open() | Ámbito cuyo join() exige que todas las subtareas terminen bien; si una falla, cancela el resto y lanza FailedException |
open(joiner) | Ámbito con otra política de espera |
open(joiner, cfg -> cfg.withTimeout(d)) | Con tiempo máximo (lanza TimeoutException) o con otra fábrica de hilos |
scope.fork(callable) | Lanza una subtarea; devuelve un Subtask |
scope.join() | Espera según el Joiner y devuelve su resultado |
subtask.get() / subtask.state() | Resultado (tras join) / SUCCESS, FAILED o UNAVAILABLE |
Joiner.anySuccessfulResultOrThrow() | El primer resultado correcto; cancela el resto |
Joiner.allSuccessfulOrThrow() | Un stream con todas las subtareas, si todas van bien |
Joiner.awaitAll() | Espera a todas, vayan bien o mal |
11. ThreadLocal y ScopedValue
A veces hay datos que deben acompañar a una petición a lo largo de muchas capas (el usuario autenticado, el identificador de transacción, el idioma) sin pasarlos como parámetro en cada método. La solución clásica es ThreadLocal; la moderna, ScopedValue.
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ValoresAcotados {
// ThreadLocal: una variable "global" con un valor distinto por hilo
static final ThreadLocal<String> USUARIO_TL = new ThreadLocal<>();
// ScopedValue (final en Java 25, JEP 506): valor inmutable visible solo durante una llamada
static final ScopedValue<String> USUARIO = ScopedValue.newInstance();
// Capa de "repositorio" profunda: no recibe el usuario como parámetro
static void guardarPedido(String producto) {
String quien = USUARIO.isBound() ? USUARIO.get() : "anónimo";
System.out.println(" [" + Thread.currentThread().getName() + "] " + quien + " guarda pedido de " + producto);
}
static void servicio(String producto) {
guardarPedido(producto); // pasa por varias capas sin parámetro extra
}
public static void main(String[] args) throws Exception {
// --- ThreadLocal ---
USUARIO_TL.set("ana");
Thread otro = new Thread(() -> System.out.println("ThreadLocal en otro hilo: " + USUARIO_TL.get()));
otro.start();
otro.join();
System.out.println("ThreadLocal en main: " + USUARIO_TL.get());
USUARIO_TL.remove(); // imprescindible en pools: si no, el valor "se queda" en el hilo
// --- ScopedValue ---
ScopedValue.where(USUARIO, "luis").run(() -> servicio("teclado"));
System.out.println("Fuera del ámbito, ¿ligado? " + USUARIO.isBound() + ", orElse: " + USUARIO.orElse("nadie"));
// Ámbitos anidados: se puede "re-ligar" temporalmente
ScopedValue.where(USUARIO, "admin").run(() -> {
servicio("monitor");
ScopedValue.where(USUARIO, "auditor").run(() -> servicio("ratón"));
servicio("cable");
});
// call(): devuelve un valor
int longitud = ScopedValue.where(USUARIO, "eva").call(() -> USUARIO.get().length());
System.out.println("call() devolvió " + longitud);
// Cada petición de un servidor, en su hilo virtual con su propio usuario
try (ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor()) {
for (String u : new String[]{"marta", "jorge", "nuria"}) {
ex.submit(() -> ScopedValue.where(USUARIO, u).run(() -> servicio("libro")));
}
}
servicio("sin ámbito");
}
}
ThreadLocal en otro hilo: null
ThreadLocal en main: ana
[main] luis guarda pedido de teclado
Fuera del ámbito, ¿ligado? false, orElse: nadie
[main] admin guarda pedido de monitor
[main] auditor guarda pedido de ratón
[main] admin guarda pedido de cable
call() devolvió 3
[] nuria guarda pedido de libro
[] marta guarda pedido de libro
[] jorge guarda pedido de libro
[main] anónimo guarda pedido de sin ámbito
(Los hilos virtuales aparecen con nombre vacío [] porque no se les ha dado nombre.)
ThreadLocal | ScopedValue (Java 25) | |
|---|---|---|
| Mutabilidad | set() en cualquier momento | Inmutable dentro del ámbito |
| Duración | Hasta remove() o hasta que muere el hilo | Exactamente la duración de run/call |
| Riesgo de fugas | Alto: en un pool, el valor pasa a la siguiente tarea si olvidas remove() | Ninguno: se desliga automáticamente |
| Coste con millones de hilos virtuales | Una copia por hilo | Muy bajo |
| Herencia a subtareas | InheritableThreadLocal (copia) | Automática a las subtareas de StructuredTaskScope |
| Lectura | get() (puede ser null) | get() (lanza excepción si no está ligado), isBound(), orElse(v) |
💡 Consejo: En código nuevo, especialmente con hilos virtuales, prefiere
ScopedValuepara pasar contexto “implícito”. Frameworks como Spring siguen usandoThreadLocalinternamente (contexto de seguridad, transacciones), y por eso es importante entender ambos. Recuerda que desde Java 25orElse(null)ya no está permitido.
12. Buenas prácticas
| Práctica | Motivo |
|---|---|
| No crees hilos a mano: usa executors | Reutilización, límites y gestión de resultados |
| Elige el executor según el tipo de tarea | CPU: pool fijo (≈ núcleos) o fork/join; E/S: hilos virtuales |
Apaga siempre los executors (try-con-recursos o shutdown + awaitTermination) | Si no, la JVM no termina y se pierden tareas |
Consulta siempre los Future o trata las excepciones en la tarea | Las excepciones de submit se pierden en silencio |
Pasa tu propio executor a supplyAsync si la tarea bloquea | No agotes el commonPool |
Pon tiempos máximos (get(t), orTimeout, withTimeout) | Un servicio colgado no debe colgar tu aplicación |
| Usa colecciones concurrentes y sus operaciones atómicas | merge, computeIfAbsent, putIfAbsent en lugar de get + put |
| Prefiere colas acotadas | Contrapresión: evitan quedarte sin memoria si el consumidor es lento |
Limita recursos escasos con Semaphore | Especialmente con hilos virtuales, que no tienen límite natural |
Prefiere ScopedValue a ThreadLocal en código nuevo | Sin fugas y eficiente con hilos virtuales |
| Mide antes de optimizar, y en condiciones realistas | El paralelismo no siempre acelera (como has visto en este módulo) |
13. Ejercicios
Ejercicio 1: Informes regionales en paralelo
Una empresa genera un informe de ventas por cada región (cada consulta tarda entre 400 y 600 ms y la de Ceuta falla por falta de conexión). Lanza las 10 consultas con un pool de 4 hilos usando invokeAll, muestra los informes correctos, suma las ventas, lista los errores y compara el tiempo con el que tardaría de forma secuencial.
Ver solución
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
public class Ej1Informes {
record Informe(String region, int ventas, long ms) { }
// Simula una consulta lenta a la base de datos de cada región
static Informe generar(String region) throws InterruptedException {
long t0 = System.nanoTime();
Thread.sleep(200 + region.length() * 40L);
if (region.equals("Ceuta")) throw new IllegalStateException("sin conexión con " + region);
int ventas = region.chars().sum() * 3;
return new Informe(region, ventas, (System.nanoTime() - t0) / 1_000_000);
}
public static void main(String[] args) throws InterruptedException {
List<String> regiones = List.of("Andalucía", "Aragón", "Asturias", "Canarias", "Cataluña", "Ceuta",
"Galicia", "Madrid", "Murcia", "Navarra");
List<Callable<Informe>> tareas = new ArrayList<>();
for (String r : regiones) tareas.add(() -> generar(r));
long t0 = System.nanoTime();
int total = 0;
List<String> errores = new ArrayList<>();
try (ExecutorService pool = Executors.newFixedThreadPool(4)) {
for (Future<Informe> f : pool.invokeAll(tareas)) {
switch (f.state()) {
case SUCCESS -> {
Informe i = f.resultNow();
total += i.ventas();
System.out.printf("%-10s %6d € (%d ms)%n", i.region(), i.ventas(), i.ms());
}
case FAILED -> errores.add(f.exceptionNow().getMessage());
default -> errores.add("tarea cancelada");
}
}
}
long ms = (System.nanoTime() - t0) / 1_000_000;
long secuencial = regiones.stream().mapToLong(r -> 200 + r.length() * 40L).sum();
System.out.println("Total: " + total + " € | errores: " + errores);
System.out.println("Tiempo con 4 hilos: " + ms + " ms (en secuencial serían unos " + secuencial + " ms)");
}
}
Andalucía 3090 € (580 ms)
Aragón 2196 € (535 ms)
Asturias 2532 € (534 ms)
Canarias 2406 € (540 ms)
Cataluña 2820 € (520 ms)
Galicia 2046 € (486 ms)
Madrid 1779 € (440 ms)
Murcia 1827 € (442 ms)
Navarra 2145 € (496 ms)
Total: 20841 € | errores: [sin conexión con Ceuta]
Tiempo con 4 hilos: 2013 ms (en secuencial serían unos 4800 ms)
Con 4 hilos y 10 tareas de unos 500 ms se necesitan 3 tandas (1500 ms teóricos; 2013 medidos por la carga de la máquina), frente a 4800 ms en secuencial. Como invokeAll devuelve los Future en el orden de las tareas, el informe sale ordenado por región aunque las consultas terminen en otro orden. Un fallo no detiene a las demás.
Ejercicio 2: Estadísticas de visitas
Ocho servidores web procesan cada uno su fichero de log (simulado con 250 000 visitas aleatorias a 5 páginas). Cuenta las visitas por página en un ConcurrentHashMap<String, LongAdder> y el total en un LongAdder, muestra las páginas ordenadas de más a menos visitadas y comprueba que la suma por páginas coincide con el total.
Ver solución
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.LongAdder;
public class Ej2Visitas {
public static void main(String[] args) {
List<String> paginas = List.of("/inicio", "/productos", "/carrito", "/ofertas", "/contacto");
ConcurrentHashMap<String, LongAdder> visitas = new ConcurrentHashMap<>();
LongAdder total = new LongAdder();
// 8 "servidores" procesan cada uno su fichero de log (simulado con 250 000 líneas)
try (ExecutorService pool = Executors.newFixedThreadPool(8)) {
for (int s = 0; s < 8; s++) {
long semilla = s;
pool.submit(() -> {
Random r = new Random(semilla);
for (int i = 0; i < 250_000; i++) {
String pagina = paginas.get(r.nextInt(paginas.size()));
visitas.computeIfAbsent(pagina, k -> new LongAdder()).increment();
total.increment();
}
});
}
}
System.out.printf("Total de visitas: %,d%n", total.sum());
visitas.entrySet().stream()
.sorted(Map.Entry.<String, LongAdder>comparingByValue((a, b) -> Long.compare(b.sum(), a.sum())))
.forEach(e -> System.out.printf(" %-10s %,8d%n", e.getKey(), e.getValue().sum()));
long suma = visitas.reduceValuesToLong(1, LongAdder::sum, 0L, Long::sum);
System.out.println("Comprobación (suma de páginas == total): " + (suma == total.sum()));
}
}
Total de visitas: 2.000.000
/inicio 400.647
/carrito 400.469
/ofertas 400.030
/productos 399.433
/contacto 399.421
Comprobación (suma de páginas == total): true
computeIfAbsent garantiza que solo se crea un LongAdder por página aunque varios hilos vean la clave por primera vez a la vez; después, increment() es concurrente y muy rápido. Como cada servidor usa una semilla fija, los resultados son reproducibles.
Ejercicio 3: Tubería de procesamiento
Implementa una tubería de tres etapas conectadas por dos BlockingQueue acotadas: un lector que envía líneas de un “fichero” (nombre;ciudad, con espacios sobrantes, mayúsculas mezcladas y líneas vacías), un transformador que descarta las vacías y normaliza nombre y ciudad, y un escritor que numera y muestra los registros. Usa una píldora venenosa para terminar.
Ver solución
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
public class Ej3Tuberia {
static final String FIN = "<<FIN>>"; // píldora venenosa
public static void main(String[] args) throws InterruptedException {
BlockingQueue<String> lineas = new LinkedBlockingQueue<>(5);
BlockingQueue<String> limpias = new LinkedBlockingQueue<>(5);
List<String> fichero = List.of(" ana;Madrid ", "LUIS;valencia", "", "eva; Sevilla", " ", "jorge;BILBAO");
try (ExecutorService pool = Executors.newFixedThreadPool(3)) {
// Etapa 1: lector
pool.submit(() -> {
for (String l : fichero) lineas.put(l);
lineas.put(FIN);
return null;
});
// Etapa 2: transformador (descarta vacías y normaliza)
pool.submit(() -> {
while (true) {
String l = lineas.take();
if (l == FIN) { limpias.put(FIN); break; }
if (l.isBlank()) continue;
String[] p = l.strip().split(";");
String nombre = p[0].strip();
String ciudad = p[1].strip().toLowerCase();
limpias.put(Character.toUpperCase(nombre.charAt(0)) + nombre.substring(1).toLowerCase()
+ " (" + Character.toUpperCase(ciudad.charAt(0)) + ciudad.substring(1) + ")");
}
return null;
});
// Etapa 3: escritor
pool.submit(() -> {
int n = 0;
while (true) {
String l = limpias.take();
if (l == FIN) break;
System.out.println(++n + ". " + l + " [" + Thread.currentThread().getName() + "]");
}
System.out.println("Escritor: " + n + " registros guardados");
return null;
});
}
}
}
1. Ana (Madrid) [pool-1-thread-3]
2. Luis (Valencia) [pool-1-thread-3]
3. Eva (Sevilla) [pool-1-thread-3]
4. Jorge (Bilbao) [pool-1-thread-3]
Escritor: 4 registros guardados
Cada etapa se envía con submit y un Callable (por eso return null), lo que permite lanzar InterruptedException sin try/catch. La píldora se compara con == a propósito: es el mismo objeto constante, de modo que una línea del fichero que contuviera el texto <<FIN>> no detendría la tubería. Cada etapa reenvía la píldora a la siguiente antes de terminar.
Ejercicio 4: Comparador de precios con tiempo máximo
Consulta en paralelo el precio de un producto en cuatro tiendas con CompletableFuture y un executor de hilos virtuales. Una tienda tarda 3 segundos y otra falla con un error 503. Aplica a cada consulta un tiempo máximo de 1 segundo, descarta las que fallen o tarden demasiado y muestra la mejor oferta.
Ver solución
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class Ej4Comparador {
record Oferta(String tienda, double precio) { }
static Oferta consultar(String tienda, double precio, long ms, boolean falla) {
try {
Thread.sleep(ms);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
if (falla) throw new IllegalStateException(tienda + ": error 503");
return new Oferta(tienda, precio);
}
public static void main(String[] args) {
try (ExecutorService pool = Executors.newVirtualThreadPerTaskExecutor()) {
List<CompletableFuture<Oferta>> consultas = List.of(
CompletableFuture.supplyAsync(() -> consultar("TiendaA", 899.0, 300, false), pool),
CompletableFuture.supplyAsync(() -> consultar("TiendaB", 849.5, 500, false), pool),
CompletableFuture.supplyAsync(() -> consultar("TiendaC", 799.0, 3000, false), pool), // demasiado lenta
CompletableFuture.supplyAsync(() -> consultar("TiendaD", 750.0, 200, true), pool)) // falla
.stream()
.map(cf -> cf.orTimeout(1, TimeUnit.SECONDS)
.exceptionally(ex -> {
System.out.println(" descartada: " + ex);
return null;
}))
.toList();
long t0 = System.nanoTime();
CompletableFuture.allOf(consultas.toArray(CompletableFuture[]::new)).join();
List<Oferta> validas = consultas.stream().map(CompletableFuture::join)
.filter(o -> o != null)
.sorted(Comparator.comparingDouble(Oferta::precio))
.toList();
System.out.println("Ofertas válidas: " + validas);
System.out.println("Mejor precio: " + validas.getFirst() + " en " + (System.nanoTime() - t0) / 1_000_000 + " ms");
}
}
}
descartada: java.util.concurrent.CompletionException: java.lang.IllegalStateException: TiendaD: error 503
descartada: java.util.concurrent.TimeoutException
Ofertas válidas: [Oferta[tienda=TiendaB, precio=849.5], Oferta[tienda=TiendaA, precio=899.0]]
Mejor precio: Oferta[tienda=TiendaB, precio=849.5] en 1394 ms
La TiendaC (la más barata, pero lenta) se descarta al cumplirse el segundo; sin el orTimeout el usuario esperaría 3 segundos. getFirst() es uno de los métodos de las colecciones secuenciadas de Java 21. Fíjate en que el try con el executor espera al salir a que termine la tarea de 3 s (los CompletableFuture no interrumpen la tarea al expirar el tiempo): en una aplicación real, la tarea lenta debería tener su propio tiempo máximo (por ejemplo, el timeout del cliente HTTP del módulo 09).
Ejercicio 5: Carrera con salida simultánea
Simula una carrera de cinco corredores (hilos de un pool). Todos deben esperar al pistoletazo de salida (CountDownLatch de 1) para empezar a la vez, tardan un tiempo aleatorio entre 100 y 400 ms y el programa principal espera a que lleguen todos (otro CountDownLatch) para mostrar la clasificación.
Ver solución
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadLocalRandom;
public class Ej5Carrera {
record Llegada(String corredor, long ms) { }
public static void main(String[] args) throws InterruptedException {
List<String> corredores = List.of("Rayo", "Trueno", "Centella", "Ciclón", "Brisa");
CountDownLatch salida = new CountDownLatch(1); // pistoletazo
CountDownLatch llegada = new CountDownLatch(corredores.size()); // todos en meta
ConcurrentLinkedQueue<Llegada> llegadas = new ConcurrentLinkedQueue<>();
long[] inicio = new long[1];
try (ExecutorService pool = Executors.newFixedThreadPool(corredores.size())) {
for (String c : corredores) {
pool.submit(() -> {
try {
salida.await(); // todos esperan la señal
Thread.sleep(ThreadLocalRandom.current().nextInt(100, 400));
long ms = (System.nanoTime() - inicio[0]) / 1_000_000;
llegadas.add(new Llegada(c, ms));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
llegada.countDown();
}
});
}
Thread.sleep(100);
System.out.println("Preparados... ¡YA!");
inicio[0] = System.nanoTime();
salida.countDown(); // suelta a todos a la vez
llegada.await(); // espera a que lleguen todos
}
// Ordenamos por tiempo: el orden de inserción en la cola puede no coincidir con el tiempo medido
int puesto = 1;
for (Llegada l : llegadas.stream().sorted(Comparator.comparingLong(Llegada::ms)).toList()) {
System.out.println(puesto++ + ".º " + l.corredor() + " (" + l.ms() + " ms)");
}
}
}
Preparados... ¡YA!
1.º Brisa (186 ms)
2.º Trueno (187 ms)
3.º Ciclón (322 ms)
4.º Rayo (346 ms)
5.º Centella (359 ms)
ThreadLocalRandom.current() da un generador por hilo, sin contención. ¿Por qué ordenar al final en lugar de usar el orden de la cola? En una primera versión de este ejercicio se mostraba la cola tal cual, y en una ejecución salió Brisa (215 ms) detrás de Rayo (349 ms): el hilo de Brisa midió su tiempo y, antes de añadirlo a la cola, el planificador le quitó la CPU. Medir y añadir son dos pasos, y entre ellos puede colarse otro hilo. La escritura de inicio[0] antes de countDown() es visible para los corredores porque countDown()/await() establecen una relación happens-before.
Ejercicio 6: Servidor con hilos virtuales y conexiones limitadas
Simula un servidor que recibe 1000 peticiones; cada una se atiende en un hilo virtual, tarda 50 ms en preparar la petición y luego necesita una conexión a la base de datos durante 100 ms. La base de datos solo admite 10 conexiones simultáneas. Limita el acceso con un Semaphore, verifica que nunca se usan más de 10 conexiones y compara el tiempo total con el mínimo teórico.
Ver solución
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
public class Ej6Servidor {
static final Semaphore POOL_BD = new Semaphore(10); // la base de datos admite 10 conexiones
static final AtomicInteger enUso = new AtomicInteger();
static final AtomicInteger maxEnUso = new AtomicInteger();
static String atenderPeticion(int id) throws InterruptedException {
Thread.sleep(50); // leer la petición, validar...
POOL_BD.acquire();
try {
maxEnUso.accumulateAndGet(enUso.incrementAndGet(), Math::max);
Thread.sleep(100); // consulta SQL
enUso.decrementAndGet();
} finally {
POOL_BD.release();
}
return "respuesta " + id;
}
public static void main(String[] args) {
int peticiones = 1_000;
AtomicInteger atendidas = new AtomicInteger();
long t0 = System.nanoTime();
try (ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor()) {
for (int i = 0; i < peticiones; i++) {
int id = i;
ex.submit(() -> {
atenderPeticion(id);
atendidas.incrementAndGet();
return null;
});
}
}
long ms = (System.nanoTime() - t0) / 1_000_000;
System.out.printf("%,d peticiones atendidas en %,d ms%n", atendidas.get(), ms);
System.out.println("Máximo de conexiones a la BD usadas a la vez: " + maxEnUso.get());
System.out.println("Mínimo teórico: " + (peticiones / 10 * 100) + " ms (1000 consultas de 100 ms de 10 en 10)");
}
}
1.000 peticiones atendidas en 12.988 ms
Máximo de conexiones a la BD usadas a la vez: 10
Mínimo teórico: 10000 ms (1000 consultas de 100 ms de 10 en 10)
Las 1000 peticiones se aceptan a la vez (mil hilos virtuales cuestan muy poco), pero el semáforo garantiza que nunca hay más de 10 en la base de datos: el cuello de botella es la BD, y el tiempo total se acerca al mínimo teórico de 10 s. Es exactamente lo que hace un pool de conexiones como HikariCP (módulo 13). Con hilos virtuales, acquire() desmonta el hilo mientras espera, así que esperar un permiso no gasta ningún hilo del sistema operativo.
Ejercicio 7 (propuesto): Ordenación con fork/join
Implementa el algoritmo merge sort como una RecursiveAction que ordena un int[] de 10 millones de elementos (por debajo de 8192 elementos usa Arrays.sort). Compara su tiempo con Arrays.sort (secuencial) y con Arrays.parallelSort, repitiendo cada medición 5 veces. Prueba distintos umbrales y explica qué ocurre con umbrales muy pequeños.
Ejercicio 8 (propuesto): Monitor de servicios
Con un ScheduledExecutorService, comprueba cada 2 segundos el “estado” de cinco servicios simulados (cada comprobación tarda entre 100 y 800 ms y falla aleatoriamente un 10 % de las veces). Las comprobaciones de cada ronda deben hacerse en paralelo con CompletableFuture y un tiempo máximo de 500 ms. Muestra una tabla con el estado de cada servicio en cada ronda y para el programa tras 5 rondas.
Ejercicio 9 (propuesto): Caché con carga perezosa
Crea una clase Cache<K, V> que reciba una Function<K, V> lenta (por ejemplo, 500 ms) y use ConcurrentHashMap.computeIfAbsent para que, aunque 100 hilos pidan la misma clave a la vez, la función solo se ejecute una vez. Añade caducidad: cada entrada guarda el Instant de carga y se recalcula si tiene más de 5 segundos (usa un Clock inyectable para probarlo, como en el módulo 05).
Ejercicio 10 (propuesto): Simulación por fases con CyclicBarrier
Simula el “juego de la vida” de Conway en un tablero de 40 × 20 dividido en 4 franjas horizontales, una por hilo. En cada generación, cada hilo calcula su franja y espera en una CyclicBarrier cuya acción de barrera intercambia el tablero actual y el siguiente e imprime el resultado. Ejecuta 10 generaciones.
Ejercicio 11 (propuesto): Agregador con concurrencia estructurada
Con StructuredTaskScope (recuerda --enable-preview), implementa un método que obtenga el perfil de un usuario a partir de tres servicios simulados (datos personales, pedidos y recomendaciones). Los dos primeros son obligatorios: si fallan, la operación falla. Las recomendaciones son opcionales: si fallan o tardan más de 300 ms, se devuelve una lista vacía. Pista: usa un ámbito anidado o un Joiner.awaitAll() y consulta subtask.state().
Ejercicio 12 (propuesto): Hilos virtuales frente a plataforma, con números propios
Repite en tu ordenador la medición de la sección 9 con 10 000, 100 000 y 1 000 000 de tareas virtuales, y con pools de 100, 1000 y 5000 hilos de plataforma (con menos tareas). Mide también la memoria usada (Runtime.totalMemory() - freeMemory()) y el número de hilos del proceso (ManagementFactory.getThreadMXBean().getThreadCount()). Presenta los resultados en una tabla y explica las diferencias.
Resumen
| Concepto | API | Idea clave |
|---|---|---|
| Pool de hilos | Executors.newFixedThreadPool(n) | Reutiliza hilos; cola de tareas |
| Tarea con resultado | Callable<V>, Future<V> | submit, get, state(), resultNow() |
| Grupos de tareas | invokeAll, invokeAny | Todas / la primera que va bien |
| Programadas | ScheduledExecutorService | schedule, scheduleAtFixedRate, scheduleWithFixedDelay |
| Apagado | shutdown, awaitTermination, shutdownNow, close | try (ExecutorService ...) desde Java 19 |
| Asincronía componible | CompletableFuture | thenApply, thenCompose, thenCombine, allOf, exceptionally, handle |
| Mapa concurrente | ConcurrentHashMap | merge, compute, computeIfAbsent atómicos |
| Lista con muchas lecturas | CopyOnWriteArrayList | Iteración sin bloqueo sobre una “foto” |
| Productor-consumidor | BlockingQueue | put / take; colas acotadas |
| Esperar a N eventos | CountDownLatch | countDown / await; de un solo uso |
| Punto de encuentro | CyclicBarrier | Reutilizable, con acción de barrera |
| Limitar accesos | Semaphore | acquire / release en finally |
| Divide y vencerás | ForkJoinPool, RecursiveTask | fork, compute, join; robo de tareas |
| Hilos virtuales (21) | Executors.newVirtualThreadPerTaskExecutor() | Millones de tareas que esperan E/S; sin pools |
| Anclaje | JEP 491 (Java 24) | synchronized ya no ancla; solo código nativo e inicialización de clases |
| Concurrencia estructurada | StructuredTaskScope.open(...) | Preview en Java 25: --enable-preview |
| Contexto implícito | ScopedValue (25), ThreadLocal | where(K, v).run(...); inmutable y sin fugas |