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 Python esto es queue.Queue(maxsize=3). put espera si está llena, y get espera si está vacía. (El módulo queue es para hilos. También hay asyncio.Queue, con los mismos métodos, para el código async.)

Míralo

El sitio y los empacadores

DONE = None




def front(orders, queue, packers):
    for order in orders:
        queue.put(order)  # espera si está llena
    for _ in range(packers):
        queue.put(DONE)  # una píldora por empacador




def pack(queue, packed):
    while True:
        order = queue.get()  # espera si está vacía
        if order is DONE:
            return  # la píldora: para
        packed.append(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 get.
  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 get 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:

@dataclass(frozen=True)
class Order:
    id: int
    euro: int
    express: bool

El sitio y los empacadores:

DONE = None




def front(orders, queue, packers):
    for order in orders:
        queue.put(order)  # espera si está llena
    for _ in range(packers):
        queue.put(DONE)  # una píldora por empacador




def pack(queue, packed):
    while True:
        order = queue.get()  # espera si está vacía
        if order is DONE:
            return  # la píldora: para
        packed.append(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.
orders = [
    Order(1001, 25, False),
    Order(1002, 40, True),
    Order(1003, 15, False),
    Order(1004, 60, False),
    Order(1005, 30, True),
    Order(1006, 45, False),
    Order(1007, 20, True),
    Order(1008, 35, False),
]
queue = Queue(maxsize=3)
packed = []  # list.append es seguro desde muchos hilos
threads = [threading.Thread(target=front, args=(orders, queue, 2)),
           threading.Thread(target=pack, args=(queue, packed)),
           threading.Thread(target=pack, args=(queue, packed))]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()
report(orders, packed)

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

def report(orders, packed):
    distinct = len({order.id for order in packed})
    print(f"Empacados: {len(packed)} de {len(orders)}")
    print("Cada pedido una vez:", "sí" if distinct == len(orders) else "no")
    for express in (True, False):
        group = [order for order in packed if order.express == express]
        kind = "Exprés" if express else "Normales"
        print(f"{kind}: {len(group)} pedidos, {sum(o.euro for o in group)} €")
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 get() 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 = None, y is DONE la compara por identidad.

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ó:

put_nowait(order) lanza queue.Full si no hay lugar. put(order, timeout=1) espera como máximo un segundo y luego lanza Full.

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?

def offer(queue, item):
    try:
        queue.put_nowait(item)
        return True
    except Full:
        return False




queue = Queue(maxsize=2)
print(offer(queue, "1001"))
print(offer(queue, "1002"))
print(offer(queue, "1003"))
print(queue.get())
print(offer(queue, "1003"))
print(queue.qsize())
Respuesta

Pruébalo

Completa

¿Cuántas píldoras pone el sitio?
for _ in range():
    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

Queue() sin maxsize no tiene límite: put nunca espera y vuelve el problema del principio de la lección. En código async la cola es asyncio.Queue(maxsize=3) con await queue.put(...) y await queue.get(): lo mismo, pero en un solo hilo.

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 put(order, timeout=0.01). Cuenta los pedidos rechazados e imprime «aceptados + rechazados = 12».
  • Por tipo. Los pedidos exprés deben empacarse primero. ¿Ayuda queue.PriorityQueue? ¿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 ·