An asynchronous queue stores work items until workers can process them. Producers add events, while several worker tasks retrieve available items concurrently, allowing long-running work to proceed without forcing the producer to wait synchronously. section 12.0 section 12.1
Use an asynchronous queue to coordinate work and control how much processing happens at once:
- Create a shared asyncio Queue and place events or data into it for later processing. section 12.0
- Run several consumers that await queue items, process each item, and continue with the next one. section 12.1
- Use a bounded queue when producers must wait asynchronously for capacity rather than adding unlimited work. section 12.0 section 12.1
- Choose priority or LIFO variants when processing order should differ from ordinary arrival order. section 12.0
Supermarket checkout
Treat customers as work items and cashiers as consumers. Each customer contains products with different checkout times; multiple cashier tasks pull customers from one queue and asynchronously simulate scanning their products. The same pattern can keep an order-management web API responsive while background workers handle longer tasks. section 12.1 section 12.0
Listing 12.1 A supermarket checkout queue
import asyncio
from asyncio import Queue
from random import randrange
from typing import List
class Product:
def __init__(self, name: str, checkout_time: float):
self.name = name
self.checkout_time = checkout_time
class Customer:
def __init__(self, customer_id: int, products: List[Product]):
self.customer_id = customer_id
self.products = products
async def checkout_customer(queue: Queue, cashier_number: int):
while not queue.empty(): #1
customer: Customer = queue.get_nowait()
print(f'Cashier {cashier_number} '
f'checking out customer '
f'{customer.customer_id}')
for product in customer.products: #2
print(f"Cashier {cashier_number} "
f"checking out customer "
f"{customer.customer_id}'s {product.name}")
await asyncio.sleep(product.checkout_time)
print(f'Cashier {cashier_number} '
f'finished checking out customer '
f'{customer.customer_id}')
queue.task_done()
async def main():
customer_queue = Queue()
all_products = [Product('beer', 2),
Product('bananas', .5),
Product('sausage', .2),
Product('diapers', .2)]
for i in range(10): #3
products = [all_products[randrange(len(all_products))]
for _ in range(randrange(10))]
customer_queue.put_nowait(Customer(i, products))
cashiers = [asyncio.create_task(checkout_customer(customer_queue, i)) #4
for i in range(3)] #4
await asyncio.gather(customer_queue.join(), *cashiers)
asyncio.run(main())
Runnable listing showing product and customer data classes, a shared queue, and cashier tasks that retrieve and process customers concurrently. section 12.1
