Ir al contenido
wannacode

Productor y consumidor

En resumen: Productor y consumidor es un patrón en el que unos hilos ponen tareas en una cola limitada y otros las sacan; una cola llena hace esperar al productor (backpressure).

Problema

Black Friday

A las 10:00 empieza la venta y el sitio recibe 50 pedidos por segundo. El almacén empaca 10. Hay tres malas maneras de manejarlo.

El sitio empaca por sí mismo. El cliente hace clic en «Pedir» y espera hasta que alguien empaca la caja. El sitio es tan rápido como el almacén.

El sitio pone los pedidos en una lista simple y el almacén los toma de ahí. Dos empacadores pueden tomar el mismo pedido y otro puede perderse: eso es una condición de carrera. Si añades un cerrojo, un empacador con la lista vacía tiene que comprobar una y otra vez si ha llegado algo.

La lista no tiene límite. El sitio está contento: nunca espera. Pero cada segundo la lista crece 40 pedidos. Después de una hora tiene 144 000, el último ha esperado una hora y se acaba la memoria.

La idea

Una cola con espacio limitado

Entre los dos lados va una cola. El productor pone tareas al final, los consumidores las toman del frente. Este es el patrón productor y consumidor. Los dos lados no se conocen: el sitio no sabe cuántos empacadores hay, y los empacadores no saben de dónde vienen los pedidos.

La cola es bloqueante y tiene espacio limitado:

  • si está vacía, el consumidor espera, sin hacer girar el procesador, hasta que llegue algo;
  • si está llena, el productor espera hasta que un consumidor libere un lugar;
  • el cerrojo está dentro de la cola: cada pedido va a exactamente un consumidor.
En Java esto es una BlockingQueue. ArrayBlockingQueue<>(3) tiene 3 lugares. put espera si está llena, y take espera si está vacía. Vive en java.util.concurrent, como ConcurrentLinkedQueue, donde juntamos los pedidos empacados.

Míralo

El sitio y los empacadores

static final Order DONE = new Order(0, 0, false);


static void front(List<Order> orders,
        BlockingQueue<Order> queue, int packers)
        throws InterruptedException {
    for (Order order : orders) {
        queue.put(order); // espera si está llena
    }
    for (int i = 0; i < packers; i++) {
        queue.put(DONE); // una píldora por empacador
    }
}


static void pack(BlockingQueue<Order> queue,
        Queue<Order> packed)
        throws InterruptedException {
    while (true) {
        Order order = queue.take(); // espera si está vacía
        if (order == DONE) return; // la píldora: para
        packed.add(order);
    }
}
Paso a paso, en palabras
  1. El sitio (productor) pone pedidos en una cola con 3 lugares; dos empacadores (consumidores) los toman del frente. Está vacía, así que esperan en take.
  2. El sitio pone 1001 y 1002 al final de la cola.
  3. Cada empacador toma el pedido del frente y lo empaca. Empacar es más lento que recibir pedidos.
  4. El sitio pone 1003, 1004 y 1005. Los 3 lugares están ocupados.
  5. 1006 no cabe: put espera y el sitio deja de recibir pedidos. Esto es backpressure: el lado lento frena al rápido.
  6. El empacador 1 termina con 1001 y toma 1003. Se libera un lugar y el put que esperaba deja entrar 1006.
  7. Y así sigue: los empacadores toman del frente, el sitio añade al final y en la cola nunca hay más de 3.
  8. Se acabaron los pedidos. El sitio pone una píldora por empacador después de los pedidos, así cada pedido se empaca antes que ellas.
  9. Cada empacador termina su pedido, toma una píldora y se detiene. Sin la píldora esperaría en take para siempre.
  10. 8 de 8, cada uno una vez: 3 exprés por 90 €, 5 normales por 180 €. Quién empaca cuál cambia de una ejecución a otra; los totales no.

El pedido:

record Order(int id, int euro, boolean express) {}

El sitio y los empacadores:

static final Order DONE = new Order(0, 0, false);


static void front(List<Order> orders,
        BlockingQueue<Order> queue, int packers)
        throws InterruptedException {
    for (Order order : orders) {
        queue.put(order); // espera si está llena
    }
    for (int i = 0; i < packers; i++) {
        queue.put(DONE); // una píldora por empacador
    }
}


static void pack(BlockingQueue<Order> queue,
        Queue<Order> packed)
        throws InterruptedException {
    while (true) {
        Order order = queue.take(); // espera si está vacía
        if (order == DONE) return; // la píldora: para
        packed.add(order);
    }
}
front y pack corren en tres hilos: un sitio y dos empacadores. El sitio pone los ocho pedidos y termina, y los empacadores toman hasta que reciben una señal para parar.
List<Order> orders = List.of(
        new Order(1001, 25, false),
        new Order(1002, 40, true),
        new Order(1003, 15, false),
        new Order(1004, 60, false),
        new Order(1005, 30, true),
        new Order(1006, 45, false),
        new Order(1007, 20, true),
        new Order(1008, 35, false));
BlockingQueue<Order> queue = new ArrayBlockingQueue<>(3);
Queue<Order> packed = new ConcurrentLinkedQueue<>();
List<Thread> threads = List.of(
        new Thread(() -> run(() -> front(orders, queue, 2))),
        new Thread(() -> run(() -> pack(queue, packed))),
        new Thread(() -> run(() -> pack(queue, packed))));
for (Thread thread : threads) thread.start();
for (Thread thread : threads) thread.join();
report(orders, packed);

El informe comprueba cosas que no dependen de qué empacador tomó qué pedido:

static void report(List<Order> orders, Queue<Order> packed) {
    long distinct = packed.stream().map(Order::id).distinct().count();
    System.out.println("Empacados: " + packed.size() + " de " + orders.size());
    System.out.println("Cada pedido una vez: " + (distinct == orders.size() ? "sí" : "no"));
    for (boolean express : new boolean[] {true, false}) {
        List<Order> group = packed.stream().filter(o -> o.express() == express).toList();
        int euro = group.stream().mapToInt(Order::euro).sum();
        String kind = express ? "Exprés" : "Normales";
        System.out.println(kind + ": " + group.size() + " pedidos, " + euro + " €");
    }
}
Salida
Empacados: 8 de 8
Cada pedido una vez: sí
Exprés: 3 pedidos, 90 €
Normales: 5 pedidos, 180 €

La salida es la misma en cada ejecución. Cuántos pedidos tomó cada empacador cambia, por eso el programa no lo imprime.

La idea

Cómo paran los empacadores

El empacador espera en take() hasta que llega un pedido. Si el sitio simplemente termina, los empacadores esperan para siempre y el programa nunca acaba.

Por eso, al final el sitio pone una píldora venenosa (poison pill): un valor especial que no es un pedido. Un empacador que toma la píldora se detiene. Hay una píldora por empacador: cada uno toma solo una y no la devuelve.

Las píldoras entran en la cola después de los pedidos, así cada pedido se empaca antes que ellas.

Aquí la píldora es DONE: un objeto Order que se compara con ==, es decir, por referencia. Ningún pedido real es el mismo objeto.

La idea

Backpressure

Una cola llena hace esperar al sitio. No es un error, es el objetivo: el lado lento empuja hacia atrás al rápido. Esto se llama backpressure. El sitio no puede adelantarse al almacén más que el tamaño de la cola, la memoria no crece y el último pedido espera como máximo lo que tarda en empacarse los 3 que tiene delante.

Esperar no es la única respuesta a una cola llena. También hay un método que no espera, sino que dice si funcionó:

offer(order) devuelve false si no hay lugar. offer(order, 1, SECONDS) espera como máximo un segundo.

Con él, el sitio puede mostrar «Mucha demanda, vuelve a intentarlo en un minuto» en vez de colgarse. Lo que importa es que la decisión es tuya: el pedido no desaparece en silencio.

Pruébalo

Predice la salida

Una cola con 2 lugares y sin empacadores. ¿Qué imprime?

BlockingQueue<String> queue = new ArrayBlockingQueue<>(2);
System.out.println(queue.offer("1001"));
System.out.println(queue.offer("1002"));
System.out.println(queue.offer("1003"));
System.out.println(queue.poll());
System.out.println(queue.offer("1003"));
System.out.println(queue.size());
Respuesta

Pruébalo

Completa

¿Cuántas píldoras pone el sitio?
for (int i = 0; i < ; i++) {
    queue.put(DONE);
}

Pruébalo

Encuentra el error

Un colega no quiere que el sitio espere y cambia una línea. Ahora el programa nunca se cuelga, pero el informe a veces dice «Empacados: 5 de 8». ¿En qué línea está el error?

Elige la línea con el error.

Míralo

Hilos, Workers y mensajes

Una cola también puede ser sin límite: new LinkedBlockingQueue<>(). Entonces put nunca espera y vuelve el problema del principio de la lección. Un grupo de hilos fijo (Executors.newFixedThreadPool) es exactamente una cola así, sin límite, por dentro: mira la lección siguiente.

Ejercicio

Dos entradas, tres empacadores

Resuelve la tarea en tu editor.

Los pedidos llegan de dos entradas: el sitio (2001–2006) y la aplicación móvil (3001–3006). Cada entrada es un productor aparte, y hay tres empacadores. La cola tiene 4 lugares. Las píldoras (una por empacador) llegan solo cuando las dos entradas han terminado. Imprime:

Salida
Empacados: 12 de 12
Cada pedido una vez: sí
Total: 325 €

Más tareas

  • Rechazo. Haz la cola de 1 lugar y que el sitio use offer(order, 10, MILLISECONDS). Cuenta los pedidos rechazados e imprime «aceptados + rechazados = 12».
  • Por tipo. Los pedidos exprés deben empacarse primero. ¿Ayuda PriorityBlockingQueue? ¿Qué pasa con las píldoras?
  • Una cadena. Añade una segunda cola después de los empacadores: ellos ponen en ella las cajas empacadas, y un mensajero las toma. ¿Cuántas píldoras hacen falta ahora y quién las pone?

Test

Ponte a prueba

1. La cola entre el sitio y los empacadores está llena. ¿Qué hace put?
2. ¿Por qué tiene la cola un límite? Sin él, el sitio nunca espera.
3. Dos sitios ponen pedidos, tres empacadores los toman. ¿Cuántas píldoras hacen falta y cuándo se ponen?
4. ¿Cómo sabes que el programa con la cola es correcto, si el orden de empaque es distinto en cada ejecución?

Creado por Ivo Gergov como parte de la campaña de Evo Tech por una educación accesible en programación ·