Grupo de hilos en Java

Problema

Un hilo por cada imagen

La tienda empieza unas rebajas. Los vendedores suben imágenes de sus productos, y cada imagen necesita una miniatura para la lista de productos. Hacer una miniatura es trabajo que calcula: recorre cada píxel (mira trabajo que espera y trabajo que calcula).

La primera versión inicia un hilo por cada imagen subida:

List<Thread> threads = new ArrayList<>();
for (String image : uploads) {
    Thread thread = new Thread(() -> saveThumbnail(image));
    thread.start();
    threads.add(thread);
}

Con 20 imágenes todo funciona en tu computadora: «Listas: 20 miniaturas». En la primera hora de las rebajas llegan 10 000 a la vez.

Cada hilo necesita su propia pila (normalmente hasta 1 MB) y un lugar en el sistema operativo. Con unos pocos miles de hilos la JVM se detiene con OutOfMemoryError: unable to create native thread.
Aunque haya memoria, 10 000 hilos en 8 núcleos solo se turnan. El procesador pierde tiempo cambiando de uno a otro, y cada imagen se vuelve más lenta.

El servidor no se vuelve más lento; se cae. Y un servidor caído tampoco atiende a los compradores.

La idea

Unos pocos trabajadores y una cola

Un grupo de hilos son unos pocos trabajadores, creados una vez, y delante de ellos una cola de tareas. Una tarea es «reduce esta imagen». Un trabajador libre toma la siguiente tarea de la cola, la termina y vuelve por una nueva.

El número de trabajadores es constante. 10 000 imágenes no hacen 10 000 hilos, sino 10 000 tareas en la cola. Una tarea es un objeto pequeño; un hilo es caro.
Los trabajadores se reutilizan. No se crea ni se destruye un hilo por cada imagen.
La cola absorbe el pico. Cuando las subidas llegan más rápido de lo que los trabajadores pueden atender, las tareas esperan en lugar de tumbar el servidor.

Esto es productor y consumidor: el código que acepta las subidas pone tareas en la cola, y los trabajadores son los consumidores.

Ya conociste el grupo en la lección sobre hilos: ExecutorService, submit y Future.get(). Aquí veremos qué pasa por dentro, cómo ordenar los resultados, qué pasa con los errores y cuántos trabajadores necesitas.

Míralo

Tres trabajadores

Hacer una miniatura aquí es una pausa, 2 ms por cada KB, y el tamaño sale del nombre de la imagen. Así la salida es la misma en cada ejecución:

// Crear una miniatura: aquí es una pausa, y el tamaño sale del nombre.
static int thumbnail(String image) throws InterruptedException {
    if (image.startsWith("dañada")) {
        throw new IllegalArgumentException("archivo dañado");
    }
    int kb = 20 + 3 * image.length();
    Thread.sleep(2 * kb);
    return kb;
}

Ocho imágenes, un grupo de tres trabajadores:

try (var pool = Executors.newFixedThreadPool(3)) {
    List<Future<Integer>> sizes = new ArrayList<>();
    for (String image : images) {
        sizes.add(pool.submit(() -> thumbnail(image)));
    }
    int total = 0;
    for (int i = 0; i < images.size(); i++) {
        int kb = sizes.get(i).get();
        System.out.printf("%s: %d KB%n", images.get(i), kb);
        total += kb;
    }
    System.out.println("Total: " + total + " KB");
}

Paso a paso, en palabras

8 imágenes esperan. El grupo son 3 hilos: Executors.newFixedThreadPool(3).
submit convierte cada imagen en una tarea de la cola. Los tres hilos toman las tres primeras.
Hilo 3 termina primero: Top es la tercera imagen, y su resultado va al tercer lugar. El trabajador no se detiene: toma la siguiente tarea.
La cola está vacía. Hilo 1 terminó y no tiene nada que tomar, así que espera. Los demás terminan sus últimas imágenes.
get() lee los tamaños en el orden de las imágenes, no en el orden en que terminaron. Total 379 KB; el final de try cierra el grupo.

List<String> images = List.of("Bolso.jpg", "Zapato.jpg", "Top.jpg", "Falda.jpg", "Pantalón.jpg", "Camisa.jpg", "Botín.jpg", "Boa.jpg");
try (var pool = Executors.newFixedThreadPool(3)) {
    List<Future<Integer>> sizes = new ArrayList<>();
    for (String image : images) {
        sizes.add(pool.submit(() -> thumbnail(image)));
    }
    int total = 0;
    for (int i = 0; i < images.size(); i++) {
        int kb = sizes.get(i).get();
        System.out.printf("%s: %d KB%n", images.get(i), kb);
        total += kb;
    }
    System.out.println("Total: " + total + " KB");
}

Bolso.jpg: 47 KB
Zapato.jpg: 50 KB
Top.jpg: 41 KB
Falda.jpg: 47 KB
Pantalón.jpg: 56 KB
Camisa.jpg: 50 KB
Botín.jpg: 47 KB
Boa.jpg: 41 KB
Total: 379 KB

Executors.newFixedThreadPool(3) crea tres hilos y delante de ellos una cola sin fin.
submit pone la tarea en la cola y devuelve un Future de inmediato. El bucle no espera nada.
get() espera exactamente su propia tarea. El bucle las espera en el orden de las imágenes, así que la salida también sale en ese orden.
try (var pool = …) cierra el grupo al final: close() no acepta tareas nuevas, espera a las iniciadas y detiene los hilos. Esto es de Java 19. En código más antiguo verás shutdown() y awaitTermination en un finally.

Más precisamente, para curiosos

La cola de newFixedThreadPool no tiene fin. Si las subidas llegan siempre más rápido de lo que el grupo puede atender, la cola crece hasta agotar la memoria. Un grupo con cola limitada es new ThreadPoolExecutor(3, 3, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000), new ThreadPoolExecutor.CallerRunsPolicy()): cuando la cola está llena, la tarea se ejecuta en el hilo que la envía, así que solo se ralentiza la aceptación de subidas.

Cómo es en los otros lenguajes: en C# hay un grupo compartido (ThreadPool, Task.Run, Parallel), en Python es ThreadPoolExecutor, y Node no tiene un grupo de Workers ya hecho. En un proyecto allí usas el paquete Piscina.

Pruébalo

Encuentra el error

Un compañero reescribió las miniaturas con un grupo, y las rebajas no fueron más rápidas. Seis imágenes tardan lo mismo que sin grupo:

long start = System.nanoTime();
List<Integer> sizes = new ArrayList<>();
try (var pool = Executors.newFixedThreadPool(3)) {
    for (String image : images) {
        Future<Integer> size = pool.submit(() -> thumbnail(image));
        sizes.add(size.get());
    }
}
long ms = (System.nanoTime() - start) / 1_000_000;
System.out.println("Tardó más de 0,5 s: " + (ms > 500));

Tardó más de 0,5 s: true

Elige la línea con el error.

size.get() espera la tarea justo después de submit, todavía dentro del bucle. La siguiente imagen entra al grupo solo cuando la anterior termina: tres hilos, y solo uno trabaja. Primero envía todas (sizes.add(pool.submit(…)) a una lista de Future), luego espéralas en un segundo bucle.

Las seis pausas suman 582 ms una tras otra. Con la corrección los tres trabajadores las reparten, y en mi máquina todo tarda unos 0,2 s. El tiempo depende de la máquina, por eso el programa imprime solo «más de 0,5 s»: seis pausas una tras otra tardan al menos eso.

La idea

Resultados en el orden de entrada

Las tareas terminan en un orden que no conoces: depende de las imágenes, de la carga, de qué trabajador estaba libre. Así que nunca imprimas «en el orden en que terminaron» si quieres la misma salida cada vez.

Dos formas dan el mismo resultado en cada ejecución:

En el orden de entrada. Guardas las promesas en una lista en el orden de las tareas y las esperas en ese orden. Así funciona el código de arriba.
Una suma o una cuenta. Si solo necesitas el total, el orden no importa: 47 + 50 es lo mismo que 50 + 47.

A veces quieres los resultados en el orden en que terminaron: para una barra de «3 de 8 listas» o para mostrar las primeras imágenes listas. ExecutorCompletionService los da así. Entonces recoges solo cosas que no dependen del orden:

try (var pool = Executors.newFixedThreadPool(3)) {
    var done = new ExecutorCompletionService<Integer>(pool);
    for (String image : images) {
        done.submit(() -> thumbnail(image));
    }
    int count = 0;
    int total = 0;
    for (int i = 0; i < images.size(); i++) {
        total += done.take().get();
        count++;
    }
    System.out.println("Listas: " + count + " de " + images.size() + ", en total: " + total + " KB");
}

Listas: 8 de 8, en total: 379 KB

ExecutorCompletionService mantiene su propia cola de tareas terminadas. take() espera la siguiente que termine, sea cual sea.

La idea

Un error dentro de una tarea

Una de las imágenes está dañada, y thumbnail lanza una excepción. No sale en el hilo que inició la tarea: ese hilo siguió su camino hace rato. El grupo la captura y la guarda en la promesa del resultado. Sale cuando esperas ese resultado:

List<String> images = List.of("Bolso.jpg", "Zapato.jpg", "dañada.jpg", "Top.jpg");
try (var pool = Executors.newFixedThreadPool(3)) {
    List<Future<Integer>> sizes = new ArrayList<>();
    for (String image : images) {
        sizes.add(pool.submit(() -> thumbnail(image)));
    }
    for (int i = 0; i < images.size(); i++) {
        try {
            int kb = sizes.get(i).get();
            System.out.println(images.get(i) + ": " + kb + " KB");
        } catch (ExecutionException e) {
            String why = e.getCause().getMessage();
            System.out.println(images.get(i) + ": error: " + why);
        }
    }
}

Bolso.jpg: 47 KB
Zapato.jpg: 50 KB
dañada.jpg: error: archivo dañado
Top.jpg: 41 KB

get() lanza una ExecutionException, y la excepción real está en getCause(). Las otras imágenes no sufren: el grupo sigue con las siguientes tareas.

Pruébalo

Adónde fue el error

La segunda imagen está dañada. ¿Qué imprime el programa?

List<String> images = List.of("Bolso.jpg", "dañada.jpg", "Top.jpg");
try (var pool = Executors.newFixedThreadPool(3)) {
    for (String image : images) {
        pool.submit(() -> thumbnail(image));
    }
}
System.out.println("Listo: " + images.size() + " imágenes");

Respuesta

Listo: 3 imágenes

Exception in thread "pool-1-thread-2" java.lang.IllegalArgumentException: archivo dañado
Listo: 3 imágenes

Listo: 2 imágenes

Nadie llama a get(), así que la excepción se queda en el Future y nadie la ve. Nada en la pantalla, nada en el registro: la imagen simplemente no tiene miniatura. Con execute en lugar de submit el hilo imprimiría el error, pero igualmente nadie lo capturaría. Así que guarda las promesas y espéralas.

La idea

Cuántos trabajadores

Depende de lo que hagan:

Trabajo que calcula (miniaturas, PDF, compresión): tantos como núcleos. Un noveno trabajador en 8 núcleos no tiene dónde ejecutarse; solo se turna con los demás.
Trabajo que espera (correo, pago, base de datos): más. Mientras un trabajador espera a la red, el núcleo queda libre para otro. Si una tarea espera W ms y calcula C ms, caben unos 1 + W / C trabajadores en un núcleo.

static int poolSize(int cores, int waitMs, int computeMs) {
    return cores * (1 + waitMs / computeMs);
}

System.out.println("Miniaturas (solo cálculo): " + poolSize(8, 0, 50));
System.out.println("Correos (90 ms de espera, 10 ms de cálculo): " + poolSize(8, 90, 10));

Miniaturas (solo cálculo): 8
Correos (90 ms de espera, 10 ms de cálculo): 80

El número de núcleos viene de Runtime.getRuntime().availableProcessors(). La fórmula es un punto de partida, no una ley: mide. Si los correos pasan por un servidor con un límite de 20 conexiones, 80 trabajadores solo esperarán en una cola delante de él.

Pruébalo

Completa

¿Cuántos trabajadores para una tarea que espera waitMs y calcula computeMs?

static int poolSize(int cores, int waitMs, int computeMs) {
    return cores * (1 + );
}

Más precisamente, para curiosos

Los hilos virtuales (Java 21) cambian la cuenta para el trabajo que espera. Un hilo virtual cuesta unos cientos de bytes, y cuando espera (sleep, red, base de datos), la JVM lo saca del hilo real y ejecuta otro. Executors.newVirtualThreadPerTaskExecutor() da un hilo virtual nuevo para cada tarea, y 10 000 correos son de verdad 10 000 hilos, sin grupo. Para el trabajo que calcula no ayudan: siguen siendo 8 núcleos. Y los recursos detrás (conexiones a la base de datos) siguen limitados, por ejemplo con un Semaphore.

Ejercicio

Las facturas de las rebajas

Resuelve la tarea en tu editor.

Después de las rebajas cada pedido necesita una factura. invoice(amount) crea el PDF (aquí una pausa de 50 ms) y devuelve el importe con 20 % de IVA. Un pedido con un importe de 0 o menos no es válido, y la función lanza una excepción con el texto «importe no válido».

Los pedidos son del 1001 al 1008 con los importes 50 €, 125 €, 80 €, 0 €, 240 €, 35 €, 90 € y 15 €. Haz las facturas en un grupo de 4 trabajadores. Imprímelas en el orden de los pedidos, y la no válida con el texto del error. Al final: la suma de las facturas buenas y cuántas fallaron.

Pedido 1001: 60 €
Pedido 1002: 150 €
Pedido 1003: 96 €
Pedido 1004: error: importe no válido
Pedido 1005: 288 €
Pedido 1006: 42 €
Pedido 1007: 108 €
Pedido 1008: 18 €
Total: 762 €, con error: 1

Más tareas

Una cola limitada. Haz que el grupo mantenga como máximo 10 tareas esperando en la cola. ¿Qué pasa con la undécima? (ArrayBlockingQueue y CallerRunsPolicy)
Progreso. Imprime «3 de 8 listas» después de cada factura terminada, y al final otra vez la lista ordenada.
Un timeout. Una factura que tarda más de 1 segundo cuenta como fallida, sin detener a las demás.

Test

Ponte a prueba

1. ¿Por qué un grupo de pocos hilos y no un hilo por cada imagen subida?Un hilo por imagen da tamaños incorrectosEl grupo hace más rápida cada imagenHay pocos hilos y se reutilizan, y la cola absorbe el pico

2. Un grupo de 3 trabajadores reduce 8 imágenes. La tercera termina primera. ¿Dónde está su tamaño cuando las lees en el orden de la lista con get, result, Promise.all o sizes[i]?Depende de la ejecuciónEn el primer lugar: terminó primeraEn el tercer lugar: el orden es el de las imágenes

3. Una tarea del grupo lanza una excepción. ¿Cuándo la ves?Cuando esperas su resultado: get(), result(), awaitDe inmediato, el programa se detieneNunca

4. El servidor tiene 8 núcleos. Cada correo espera 90 ms a la red y calcula 10 ms. ¿Cuántos trabajadores mantienen ocupados los núcleos, más o menos?Unos 10 000Exactamente 8Unos 80

5. ¿Cuántos trabajadores para hacer miniaturas, si solo calcula y el servidor tiene 8 núcleos?Unos 8Tantos como imágenes haya en la colaUnos 80

Grupo de hilos y tareas

Cookies