Les cues asíncrones en Python són una estructura fonamental de la programació concurrent quan s'utilitza la biblioteca asyncio.
Igual que en la teoria de cues clàssica, serveixen per emmagatzemar elements que han de ser processats per un conjunt de treballadors, però aquí s'integren amb el model d'E/S asíncrona de Python i amb la sintaxi async/await introduïda a Python 3.5 mitjançant el PEP 492.
En informàtica, una cua és una estructura de dades FIFO (First In, First Out), on el primer element que entra és el primer que surt.
Aquesta idea s'utilitza per modelar situacions com ara clients esperant en una caixa de supermercat o peticions entrant a un servidor.
Les cues d'asyncio porten aquest concepte al món de la programació asíncrona: permeten que diverses corrutines productores i consumidores intercanviïn missatges de forma segura sense necessitat de blocs explícits.
Asyncio i l'E/S asíncrona a Python asyncio és la biblioteca estàndard de Python per escriure codi concurrent amb la sintaxi async/await. A partir de Python 3.5, introduïdes via PEP 492, aquestes paraules clau simplifiquen la definició de corrutines i l'ús del bucle d'esdeveniments.
El model d'asyncio es basa en un bucle d'esdeveniments (event loop) que gestiona múltiples tasques que esperen operacions d'E/S: accessos a xarxa, fitxers, bases de dades, etc. En lloc de bloquejar el fil d'execució mentre una operació d'E/S està pendent, la corrutina es «suspèn» amb await i el bucle d'esdeveniments pot continuar executant altres tasques. Dins d'aquest context, les cues asíncrones són una eina per coordinar el flux de treball entre diferents corrutines.
Classes principals de cues en asyncio La biblioteca asyncio proporciona tres classes de cues amb comportaments diferents.
asyncio.Queue (FIFO) asyncio.Queue implementa una cua FIFO clàssica. Els elements es processen en el mateix ordre en què s'han afegit.
await queue.put(item): afegeix un element a la cua, esperant si està plena. item = await queue.get(): extreu el següent element, esperant si la cua és buida. queue.task_done(): indica que el processament d'aquell element ha acabat. await queue.join(): espera fins que tots els elements afegits hagin estat marcats com a completats.
Aquesta és la variant més utilitzada en exemples del patró productor–consumidor i en servidors web que processen tasques en segon pla.
asyncio.PriorityQueue asyncio.PriorityQueue és una cua on cada element té una prioritat, i és aquesta prioritat la que determina l'ordre de sortida.
Normalment s'hi introdueixen tuples del tipus (prioritat, valor) o bé objectes comparables mitjançant dataclass(order=True).
Un valor numèric més petit indica major prioritat, de manera que un element de prioritat 0 es processarà abans que un de prioritat 10, encara que hagi entrat més tard. Això és útil en sistemes on certes tasques (per exemple, alertes crítiques) s'han de processar abans que altres treballs normals.
asyncio.LifoQueue asyncio.LifoQueue implementa una cua LIFO (Last In, First Out), que es comporta com una pila. L'últim element afegit és el primer a ser processat. Aquest comportament és apropiat quan interessa donar preferència a les tasques més recents, o en algorismes com la cerca en profunditat o funcionalitats d'«undo».
Totes tres classes comparteixen la mateixa interfície bàsica (put, get, task_done, join), la qual cosa facilita canviar d'un tipus a un altre segons les necessitats del sistema.
Patró productor i consumidor amb asyncio.Queue El patró productor consumidor és un esquema clàssic de programació concurrent en què alguns components del programa (productors) generen dades o tasques i altres components (consumidors) les processen.
La cua actua com a búfer intermedi que desacobla la velocitat de producció de la velocitat de consum. Amb asyncio.Queue, aquest patró es pot implementar mitjançant corrutines:
El productor fa await queue.put(tasca) per introduir nous elements. Els consumidors fan item = await queue.get() i processen cada tasca. Després del processament, criden queue.task_done() per mantenir el recompte intern de tasques pendents.
Això permet que múltiples consumidors treballin en paral·lel sobre la mateixa cua, aprofitant la concurrència cooperativa que ofereix asyncio.
Ús en servidors web i sistemes de cues de tasques Una aplicació habitual de les cues asíncrones és en servidors web i API implementats amb frameworks basats en asyncio, com aiohttp o FastAPI.
Quan el servidor rep una petició que implica una operació lenta, la petició es converteix en una tasca i s'introdueix a una asyncio.Queue. La resposta HTTP es retorna de seguida al client indicant que la tasca ha estat acceptada, i un conjunt de workers en segon pla extreuen tasques de la cua i les processen a mesura que hi ha recursos disponibles.
Aquest mateix patró inspira sistemes de cues de tasques distribuïdes com Celery, que sovint treballen sobre RabbitMQ o Redis. La diferència principal és que asyncio.Queue viu només dins del procés Python local, mentre que plataformes com RabbitMQ ofereixen cues persistents i distribuïdes entre molts processos i servidors.
Limitacions i comparació amb solucions persistents Les cues d'asyncio presenten diverses limitacions respecte a solucions externes.
Les dades es perden si el programa s'atura Imagina que tens una cua amb cent comandes d'una botiga en línia i el servidor cau per un tall de llum. Quan el servidor torna a arrencar, totes aquelles comandes han desaparegut, perquè asyncio.Queue guarda les dades a la memòria RAM i no al disc.
Per a tasques importants on no et pots permetre perdre dades, cal emprar eines externes com RabbitMQ o Redis, que guarden les tasques de forma persistent.
Només funciona dins d'un sol programa asyncio.Queue és com una pissarra interna del teu programa: només la pot veure i modificar aquell procés Python concret. Si vols que dos servidors diferents comparteixin feina, o que diversos processos col·laborin, hauràs d'usar un sistema extern com una base de dades o un broker de missatges.
Quan usar asyncio.Queue i quan no asyncio.Queue és una opció excel·lent per aprendre, fer prototips o per a projectes petits on perdre alguna tasca ocasionalment no és un problema greu.
En canvi, en un sistema en producció on no et pots permetre perdre cap dada (per exemple, pagaments o comandes), és millor combinar asyncio amb eines com Celery, RabbitMQ o Redis, que ofereixen garanties de persistència i recuperació d'errors.
Exemple de productor-consumidor El codi següent mostra una cua FIFO amb un productor i dos consumidors. Un cop el productor ha acabat i totes les tasques han estat processades (queue.join()), les tasques dels consumidors es cancel·len de forma neta:
import asyncio
async def producer(queue: asyncio.Queue) -> None: for i in range(5): task = f"Tasca {i}" print(f"[PROD] Enviant {task}") await queue.put(task) # Espera si la cua està plena await asyncio.sleep(0.2) # Simula treball del productor
async def consumer(queue: asyncio.Queue, name: str) -> None: while True: task = await queue.get() # Espera fins que hi ha un element print(f"[{name}] Processant {task}") await asyncio.sleep(0.5) # Simula treball real queue.task_done() # Marca l'element com a completat
async def main() -> None: queue = asyncio.Queue(maxsize=3)
prod = asyncio.create_task(producer(queue)) cons = [ asyncio.create_task(consumer(queue, "CONSUMIDOR 1")), asyncio.create_task(consumer(queue, "CONSUMIDOR 2")), ]
await asyncio.gather(prod) # Espera que el productor acabi await queue.join() # Espera que totes les tasques acabin
for c in cons: c.cancel() # Cancel·la els consumidors de forma neta await asyncio.gather(*cons, return_exceptions=True)
asyncio.run(main())
El productor introdueix cinc tasques a la cua.
Cada consumidor extreu tasques amb await queue.get(), les processa i crida queue.task_done().
La crida queue.join() bloqueja fins que totes les tasques han estat marcades com a completades. Finalment, els consumidors es cancel·len amb cancel(), evitant situacions de bloqueig quan hi ha múltiples consumidors.
Vegeu també Programació asíncrona Python Patró de disseny de programari
Referències
Categoria:Python