Código asíncrono en Java: CompletableFuture y carreras

Problema

La última unidad, vendida dos veces

Las rebajas van a todo gas. Queda un par de auriculares. checkout comprueba las existencias, espera a que el banco confirme el pago y solo entonces reduce las existencias. Dos clientes pulsan «Comprar» en el mismo momento:

// Un hilo hace de bucle de eventos: las continuaciones se ejecutan en él una por una.
static final ExecutorService loop = Executors.newSingleThreadExecutor();

// after(ms): ejecuta la tarea en loop después de ms milisegundos.
static Executor after(long ms) {
    return CompletableFuture.delayedExecutor(ms, TimeUnit.MILLISECONDS, loop);
}

// El banco confirma el pago en 100 ms.
static CompletableFuture<Void> pay(String buyer) {
    return CompletableFuture.runAsync(() -> {}, after(100));
}

static int stock = 1;
static int sold = 0;

static CompletableFuture<Void> checkout(String buyer) {
    int left = stock;
    if (left == 0) {
        return CompletableFuture.completedFuture(null);
    }
    return pay(buyer).thenRunAsync(() -> {
        stock = left - 1;
        sold++;
    }, loop);
}

var first = checkout("Ana");
var second = checkout("Beto");
CompletableFuture.allOf(first, second).join();
System.out.println("Vendidos: " + sold + ", existencias: " + stock);

Vendidos: 2, existencias: 0

Se venden dos, y hay una unidad. Un cliente recibirá un correo «Tu pedido ha sido enviado» por unos auriculares que no existen.

Todo aquí se ejecuta en un hilo: loop es Executors.newSingleThreadExecutor(). En la lección Condición de carrera dos hilos chocaron en count++. Aquí no hay un segundo hilo.

La salida es la misma en cada ejecución: no es mala suerte, es el orden en que se ejecuta el código. Para verlo, hay que saber qué hace await.

La idea

Un hilo y muchas esperas

La mayor parte del trabajo en una tienda en línea es trabajo que espera (mira trabajo que espera y trabajo que calcula): el pago espera al banco, el correo espera al servidor de correo, las existencias esperan al almacén. Un hilo que se queda sentado esperando es un hilo desperdiciado.

Un bucle de eventos es un hilo y una cola de trozos de código que están listos. El hilo toma el siguiente trozo y lo ejecuta hasta el final. Cuando el código llega a una espera, no retiene el hilo: dice «despiértame cuando responda el banco» y lo suelta. El hilo toma el siguiente trozo listo. Cuando el banco responde, el resto de la función (la continuación) entra en la cola y espera su turno.

Java no tiene async ni await. Lo más parecido es CompletableFuture: la promesa de un resultado, a la que enganchas una continuación con thenRunAsync o thenApplyAsync. Tú dices en qué executor se ejecuta la continuación. Aquí todas se ejecutan en loop, un executor con un hilo que hace de bucle de eventos. Así funcionan Netty y Vert.x, y muchos servidores Java se construyen sobre ellos.

Tres peticiones a tres servicios, lanzadas a la vez:

static int waiting = 0;
static int most = 0;

static CompletableFuture<Void> call(String name, int ms) {
    waiting++;
    most = Math.max(most, waiting);
    return CompletableFuture.runAsync(() -> {
        waiting--;
        System.out.println(name + ": listo tras " + ms + " ms");
    }, after(ms));
}

CompletableFuture.allOf(
    call("pago", 300),
    call("correo", 100),
    call("almacén", 200)
).join();
System.out.println("Esperaron a la vez: " + most);

correo: listo tras 100 ms
almacén: listo tras 200 ms
pago: listo tras 300 ms
Esperaron a la vez: 3

after(ms) da un executor que ejecuta la tarea en loop después de ms milisegundos: es el temporizador. call aumenta waiting y vuelve de inmediato. allOf(...).join() espera a las tres.

«Esperaron a la vez: 3»: las tres peticiones esperan al mismo tiempo, y solo hay un hilo. Todo tarda unos 300 ms (la más larga), no 600 (la suma). El hilo no esperó a ninguna de ellas.

Pruébalo

Cuántas esperan a la vez

El mismo call, pero la primera petición se espera sola:

¿Qué imprimirá el programa?

call("pago", 300).join();
CompletableFuture.allOf(call("correo", 100), call("almacén", 200)).join();
System.out.println("Esperaron a la vez: " + most);

Respuesta

correo: listo tras 100 ms
almacén: listo tras 200 ms
pago: listo tras 300 ms
Esperaron a la vez: 3

pago: listo tras 300 ms
correo: listo tras 100 ms
almacén: listo tras 200 ms
Esperaron a la vez: 3

pago: listo tras 300 ms
correo: listo tras 100 ms
almacén: listo tras 200 ms
Esperaron a la vez: 2

El primer call(...).join() espera el pago hasta el final. Solo después allOf lanza las otras dos a la vez, así que el máximo es 2, y el orden de la salida es el orden en que terminan. Todo tarda unos 500 ms.

await justo después de la llamada significa «espera esto antes de seguir». Para que esperen juntas, lanza primero todas, luego espera.

Míralo

Una carrera entre dos awaits

Volvamos a la última unidad. Así se entrelazan los dos pedidos:

static int stock = 1;
static int sold = 0;

static CompletableFuture<Void> checkout(String buyer) {
    int left = stock;
    if (left == 0) {
        return CompletableFuture.completedFuture(null);
    }
    return pay(buyer).thenRunAsync(() -> {
        stock = left - 1;
        sold++;
    }, loop);
}

Paso a paso, en palabras

Queda una unidad. Dos pedidos llegan en el mismo momento: el de Ana y el de Beto.
El checkout de Ana lee las existencias: left es 1. El resto está en thenRunAsync: se ejecutará cuando responda el banco, y hasta entonces el hilo está libre.
El hilo está libre, así que se ejecuta el checkout de Beto. También lee 1: Ana todavía no ha escrito nada.
El banco responde a Ana. Su código continúa después de thenRunAsync: stock = left - 1 = 0, sold = 1. Hasta aquí es correcto.
Beto continúa con el left = 1 antiguo: otra vez stock = 0, sold = 2. Una unidad, dos ventas, y con un solo hilo.

Entre dos await (aquí: entre el inicio del método y thenRunAsync) el código se ejecuta sin interrupción. Pero en un await la función suelta el hilo, y mientras tanto puede ejecutarse cualquier otra cosa. Cuando continúa, el mundo puede ser distinto: left es un valor leído antes de la espera.

Es el mismo «leer, cambiar, escribir» que con hilos, solo que la interrupción no es en cualquier parte, sino exactamente en el await. Por eso la salida es la misma en cada ejecución: la interrupción no es aleatoria.

La idea

La solución: reservar la unidad antes del await

La comprobación y la reducción deben ser un solo trozo de código sin ningún await en medio. El pedido reserva la unidad de inmediato y solo después espera al banco:

static int stock = 1;
static int sold = 0;

static CompletableFuture<Void> checkout(String buyer) {
    if (stock == 0) {
        return CompletableFuture.completedFuture(null);
    }
    stock--;   // reserva la unidad antes del await
    return pay(buyer).thenRunAsync(() -> sold++, loop);
}

Vendidos: 1, existencias: 0

La comprobación y stock-- van antes de pay(...) y se ejecutan de una vez. Aquí checkout se llama desde un hilo, así que basta. Si lo llamaran muchos hilos (las peticiones de un servidor web), necesitarías además synchronized alrededor de la comprobación y la reducción, como en la lección Bloqueo.

Si el pago falla, hay que devolver la unidad. Ese es el ejercicio del final.

Más precisamente, para curiosos

A veces hay que bloquear todo el trozo junto con la espera: por ejemplo, solo un pedido a la vez puede escribir en un sistema de almacén antiguo que no soporta dos. Entonces necesitas un cerrojo que entienda await:

En un bucle con un solo hilo, cada trozo entre dos continuaciones se ejecuta solo de todos modos. Con muchos hilos usas un Semaphore o una cola con un hilo para el almacén.

Cómo es en los otros lenguajes: en JavaScript y en asyncio hay un hilo, y el trozo entre dos await se ejecuta solo. En C# las continuaciones están en el grupo de hilos, así que allí también hace falta un lock. Java no tiene await: allí se hace lo mismo con CompletableFuture, y desde Java 21 también con hilos virtuales, donde escribes código bloqueante normal y la JVM desmonta el hilo por sí sola mientras espera.

Pruébalo

Encuentra el error

El carrito suma los precios de tres pedidos. El precio viene del almacén con await. Los tres pedidos cuestan 25 € cada uno, y el total sale como 25:

static int total = 0;

static CompletableFuture<Void> addOrder(int id) {
    int before = total;
    return priceOf(id).thenAcceptAsync(price -> {
        total = before + price;
    }, loop);
}

CompletableFuture.allOf(addOrder(1), addOrder(2), addOrder(3)).join();
System.out.println("Total: " + total + " €");

Total: 25 €

Elige la línea con el error.

int before = total lee el total antes de que llegue el precio. Los tres pedidos leen 0, y las continuaciones escriben 0 + 25. La suma debe estar en la continuación: priceOf(id).thenAcceptAsync(price -> total += price, loop).

El error no está en los hilos sino en el orden: un valor leído antes de un await está obsoleto después.

La idea

No bloquees el bucle de eventos

El bucle de eventos solo funciona si ningún trozo de código retiene el hilo durante mucho tiempo. El informe de ventas se hace con Thread.sleep(300) en loop. A la vez, un cliente abre la página de un producto, que espera 50 ms al almacén:

static void buildReport() throws InterruptedException {
    Thread.sleep(300);   // bloquea: espera 300 ms sin await
}

long start = System.nanoTime();
CompletableFuture<Void> report = CompletableFuture.runAsync(() -> {
    try {
        buildReport();
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
    System.out.println("El informe está listo");
}, loop);
CompletableFuture<Void> page = CompletableFuture.runAsync(() -> {
    long ms = (System.nanoTime() - start) / 1_000_000;
    System.out.println("Ana espera más de 0,25 s: " + (ms > 250));
}, after(50));
CompletableFuture.allOf(report, page).join();

El informe está listo
Ana espera más de 0,25 s: true

La página esperó más de 0,25 s en lugar de 50 ms. Mientras el informe retiene el hilo, ninguna otra continuación puede ejecutarse: ni esta página, ni los pagos, ni los temporizadores. En un servidor con miles de clientes, todos se congelan a la vez.

La solución es sacar el trabajo pesado del hilo del bucle de eventos. Va a un grupo de hilos, y el bucle de eventos solo espera el resultado:

static final ExecutorService workers = Executors.newFixedThreadPool(2);

static CompletableFuture<Void> report() {
    return CompletableFuture.runAsync(() -> {
        try {
            buildReport();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }, workers).thenRunAsync(() -> System.out.println("El informe está listo"), loop);
}

CompletableFuture<Void> page = CompletableFuture.runAsync(
    () -> System.out.println("Ana: la página está lista"), after(50));
CompletableFuture.allOf(report(), page).join();
workers.shutdown();

Ana: la página está lista
El informe está listo

runAsync(..., workers) hace el informe en el grupo workers, y thenRunAsync(..., loop) devuelve la continuación a loop. loop queda libre, y la página está lista a los 50 ms.

Pruébalo

Completa

Lanza las tres peticiones a la vez y espéralas juntas:

(
    call("pago", 300),
    call("correo", 100),
    call("almacén", 200)
).join();
System.out.println("Esperaron a la vez: " + most);

Ejercicio

Reservar, cobrar, devolver

Resuélvelo en tu propio editor.

Quedan dos unidades. Cuatro clientes piden a la vez: Ana, Beto, Vera y Gala. El pago tarda 100 ms. La tarjeta de Beto es rechazada: el pago lanza una excepción con el texto «tarjeta rechazada».

checkout reserva la unidad antes del await, luego cobra. Si el pago falla, la unidad se devuelve. Imprime el resultado de cada cliente en el orden de los pedidos, y al final las existencias y el número de vendidos:

Ana: pagado
Beto: tarjeta rechazada, la unidad se devuelve
Vera: sin existencias
Gala: sin existencias
Existencias: 1, vendidos: 1

Vera y Gala ven «sin existencias», porque la unidad de Beto sigue reservada hasta que responde el banco. Es justo: la unidad devuelta va al siguiente cliente.

Más tareas

Intentar de nuevo. Si un cliente no encuentra existencias pero otro podría devolver una unidad, déjalo esperar hasta 200 ms y volver a intentarlo.
Timeout. El banco a veces no responde. Si el pago tarda más de 1 s, rechaza el pedido y devuelve la unidad. (orTimeout)
Informe. Calcula el total de las ventas (25 € cada una) e imprímelo al final.

Test

Ponte a prueba

1. ¿En cuántos hilos se ejecuta tu código en JavaScript (Node) y en Python asyncio?En tantos como núcleos hayaEn uno: el bucle de eventos ejecuta los trozos de código uno por unoEn un hilo por cada await

2. Tres peticiones esperan 100, 200 y 300 ms. ¿Cuánto tarda todo si las esperas una por una con await, y cuánto con Promise.all, gather, Task.WhenAll o allOf?Unos 300 ms de las dos formasUnos 600 ms y unos 300 msUnos 600 ms de las dos formas

3. En JavaScript y Python una función async lee datos compartidos, espera con await y luego escribe. ¿Cuándo puede otro código cambiar los datos?Entre dos líneas cualesquieraSolo mientras espera en el awaitNunca: hay un solo hilo

4. Atender una petición incluye un cálculo que dura 2 segundos, sin un solo await. ¿Qué pasa con las demás peticiones en Node o asyncio?Esperan 2 segundos: el bucle de eventos está ocupadoPasan a otro hiloNada: la función async se ejecuta aparte

5. ¿Cómo evitas vender dos veces la última unidad?Ejecutar los pedidos en dos hilosAñadir un await más antes de escribirComprobar y reducir las existencias sin ningún await en medio, luego cobrar; si el pago falla, devolver la unidad

async no es un hilo

Cookies