+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
Home / Tutorials / 🐍 python
Part 326 of 365

📘 Asyncio Queues: Producer-Consumer

Master asyncio queues: producer-consumer in Python with practical examples, best practices, and real-world applications 🚀

💎Advanced
25 min read

Prerequisites

  • Basic understanding of programming concepts 📝
  • Python installation (3.8+) 🐍
  • VS Code or preferred IDE 💻

What you'll learn

  • Understand the concept fundamentals 🎯
  • Apply the concept in real projects 🏗️
  • Debug common issues 🐛
  • Write clean, Pythonic code ✨

🎯 Introduction

Welcome to this exciting tutorial on asyncio queues and the producer-consumer pattern! 🎉 In this guide, we’ll explore how to build powerful concurrent applications using Python’s asyncio queues.

You’ll discover how asyncio queues can transform your concurrent Python development experience. Whether you’re building web scrapers 🕷️, data pipelines 🚰, or real-time systems 📡, understanding asyncio queues is essential for writing efficient, scalable code.

By the end of this tutorial, you’ll feel confident using producer-consumer patterns in your own projects! Let’s dive in! 🏊‍♂️

📚 Understanding Asyncio Queues

🤔 What are Asyncio Queues?

Asyncio queues are like conveyor belts in a factory 🏭. Think of them as buffered channels that allow different parts of your program to communicate asynchronously without blocking each other.

In Python terms, asyncio queues provide thread-safe, async-friendly communication between coroutines. This means you can:

  • ✨ Decouple producers from consumers
  • 🚀 Handle different processing speeds gracefully
  • 🛡️ Prevent data loss with buffering
  • 🎯 Scale your application efficiently

💡 Why Use Producer-Consumer Pattern?

Here’s why developers love the producer-consumer pattern:

  1. Decoupling 🔓: Producers and consumers work independently
  2. Buffering 📦: Handle speed mismatches between components
  3. Scalability 📈: Easy to add more producers or consumers
  4. Resilience 🛡️: System continues even if parts slow down

Real-world example: Imagine a restaurant kitchen 🍳. Waiters (producers) place orders on a queue, and chefs (consumers) process them at their own pace!

🔧 Basic Syntax and Usage

📝 Simple Queue Example

Let’s start with a friendly example:

import asyncio

# 👋 Hello, Asyncio Queues!
async def producer(queue):
    # 🎨 Creating items to process
    for i in range(5):
        item = f"Task {i} 🎯"
        await queue.put(item)
        print(f"📥 Produced: {item}")
        await asyncio.sleep(0.5)  # Simulate work
    
    # 🛑 Signal completion
    await queue.put(None)

async def consumer(queue, consumer_id):
    # 🔄 Process items from queue
    while True:
        item = await queue.get()
        
        # 🛑 Check for completion signal
        if item is None:
            queue.task_done()
            break
            
        print(f"  📤 Consumer {consumer_id} processing: {item}")
        await asyncio.sleep(1)  # Simulate processing
        queue.task_done()

# 🚀 Run the example
async def main():
    # 📦 Create a queue with max size 3
    queue = asyncio.Queue(maxsize=3)
    
    # 🎮 Start producer and consumer
    await asyncio.gather(
        producer(queue),
        consumer(queue, "👨‍🍳")
    )

asyncio.run(main())

💡 Explanation: Notice how the producer creates items faster than the consumer processes them. The queue handles this speed mismatch perfectly!

🎯 Common Queue Patterns

Here are patterns you’ll use daily:

# 🏗️ Pattern 1: Multiple Consumers
async def multi_consumer_example():
    queue = asyncio.Queue()
    
    # 🎨 Create multiple consumers
    consumers = [
        consumer(queue, f"Worker-{i} 🤖")
        for i in range(3)
    ]
    
    # 🚀 Run producer and consumers together
    await asyncio.gather(
        producer(queue),
        *consumers
    )

# 🎯 Pattern 2: Priority Queue
async def priority_example():
    # 📊 Priority queue sorts items automatically
    priority_queue = asyncio.PriorityQueue()
    
    # Items are (priority, data) tuples
    await priority_queue.put((3, "Low priority 🐌"))
    await priority_queue.put((1, "High priority 🚀"))
    await priority_queue.put((2, "Medium priority 🚶"))
    
    # 🎯 Items come out in priority order
    while not priority_queue.empty():
        priority, item = await priority_queue.get()
        print(f"Processing: {item}")

# 🔄 Pattern 3: LIFO Queue (Stack)
async def lifo_example():
    # 📚 Last In, First Out
    lifo_queue = asyncio.LifoQueue()
    
    # 📥 Add items
    for emoji in ["🍎", "🍊", "🍌"]:
        await lifo_queue.put(emoji)
    
    # 📤 Items come out in reverse order
    while not lifo_queue.empty():
        item = await lifo_queue.get()
        print(f"Got: {item}")  # 🍌, 🍊, 🍎

💡 Practical Examples

🕷️ Example 1: Web Scraper

Let’s build a concurrent web scraper:

import asyncio
import aiohttp
from asyncio import Queue

# 🌐 URL producer
async def url_producer(queue, urls):
    print("🎯 Starting URL producer...")
    for url in urls:
        await queue.put(url)
        print(f"  📥 Queued: {url}")
    
    # 🛑 Signal completion to all workers
    for _ in range(3):  # Number of workers
        await queue.put(None)

# 🕷️ Web scraper consumer
async def scraper_worker(queue, session, worker_id):
    print(f"🤖 Worker {worker_id} started!")
    
    while True:
        url = await queue.get()
        
        if url is None:
            queue.task_done()
            print(f"  🛑 Worker {worker_id} finished!")
            break
        
        try:
            # 🌐 Fetch the page
            async with session.get(url) as response:
                content = await response.text()
                print(f"  ✅ Worker {worker_id} scraped {url} ({len(content)} bytes)")
        except Exception as e:
            print(f"  ❌ Worker {worker_id} error on {url}: {e}")
        finally:
            queue.task_done()

# 🚀 Main scraper
async def web_scraper():
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/1",
        "https://httpbin.org/uuid",
        "https://httpbin.org/user-agent"
    ]
    
    # 📦 Create work queue
    work_queue = Queue(maxsize=10)
    
    async with aiohttp.ClientSession() as session:
        # 🎮 Start producer and workers
        tasks = [
            url_producer(work_queue, urls),
            scraper_worker(work_queue, session, "🕷️-1"),
            scraper_worker(work_queue, session, "🕷️-2"),
            scraper_worker(work_queue, session, "🕷️-3")
        ]
        
        await asyncio.gather(*tasks)
    
    print("🎉 Scraping complete!")

# asyncio.run(web_scraper())

🎯 Try it yourself: Add a result queue to collect scraped data and process it separately!

🚰 Example 2: Data Processing Pipeline

Let’s make a multi-stage data pipeline:

import asyncio
import random
from dataclasses import dataclass
from typing import Optional

# 📊 Data model
@dataclass
class DataItem:
    id: int
    value: float
    emoji: str = "📊"

# 🎯 Stage 1: Data Generator
async def data_generator(raw_queue: Queue, count: int):
    print("🏭 Data generator started...")
    
    for i in range(count):
        # 🎲 Generate random data
        item = DataItem(
            id=i,
            value=random.uniform(0, 100),
            emoji=random.choice(["📊", "📈", "📉", "💹"])
        )
        
        await raw_queue.put(item)
        print(f"  🎲 Generated: {item.emoji} ID={item.id}, Value={item.value:.2f}")
        await asyncio.sleep(0.1)
    
    # 🛑 Signal completion
    await raw_queue.put(None)

# 🔧 Stage 2: Data Processor
async def data_processor(raw_queue: Queue, processed_queue: Queue):
    print("⚙️ Data processor started...")
    
    while True:
        item = await raw_queue.get()
        
        if item is None:
            raw_queue.task_done()
            await processed_queue.put(None)
            break
        
        # 🎨 Process the data
        item.value = item.value ** 2  # Square the value
        await processed_queue.put(item)
        print(f"  ⚙️ Processed: {item.emoji} ID={item.id}, New Value={item.value:.2f}")
        raw_queue.task_done()
        await asyncio.sleep(0.2)

# 💾 Stage 3: Data Saver
async def data_saver(processed_queue: Queue):
    print("💾 Data saver started...")
    results = []
    
    while True:
        item = await processed_queue.get()
        
        if item is None:
            processed_queue.task_done()
            break
        
        # 💾 Save the data
        results.append(item)
        print(f"  💾 Saved: {item.emoji} ID={item.id}")
        processed_queue.task_done()
    
    # 📊 Final stats
    total = sum(item.value for item in results)
    print(f"\n🎉 Pipeline complete! Processed {len(results)} items, Total: {total:.2f}")

# 🚀 Run the pipeline
async def data_pipeline():
    # 📦 Create queues for each stage
    raw_queue = Queue(maxsize=5)
    processed_queue = Queue(maxsize=5)
    
    # 🎮 Run all stages concurrently
    await asyncio.gather(
        data_generator(raw_queue, 10),
        data_processor(raw_queue, processed_queue),
        data_saver(processed_queue)
    )

# asyncio.run(data_pipeline())

🚀 Advanced Concepts

🧙‍♂️ Advanced Pattern: Rate Limiting

When you’re ready to level up, try this rate-limiting pattern:

import asyncio
from asyncio import Queue
import time

# 🎯 Rate-limited producer
class RateLimitedProducer:
    def __init__(self, rate_limit: int):
        self.rate_limit = rate_limit  # Items per second
        self.semaphore = asyncio.Semaphore(rate_limit)
        self.reset_task = None
    
    async def start(self):
        # 🔄 Reset semaphore every second
        self.reset_task = asyncio.create_task(self._reset_loop())
    
    async def _reset_loop(self):
        while True:
            await asyncio.sleep(1.0)
            # 🔓 Release all semaphore slots
            for _ in range(self.rate_limit):
                try:
                    self.semaphore.release()
                except ValueError:
                    pass  # Already at max
    
    async def produce(self, queue: Queue, items: list):
        await self.start()
        
        for item in items:
            await self.semaphore.acquire()
            await queue.put(item)
            print(f"  🚦 Rate-limited produce: {item}")
        
        await queue.put(None)
        self.reset_task.cancel()

# 🪄 Using the rate limiter
async def rate_limited_example():
    queue = Queue()
    producer = RateLimitedProducer(rate_limit=3)  # 3 items per second
    
    items = [f"Item-{i} 🎯" for i in range(10)]
    
    # 🎮 Run with rate limiting
    await asyncio.gather(
        producer.produce(queue, items),
        consumer(queue, "🤖")
    )

🏗️ Advanced Pattern: Backpressure Handling

For the brave developers:

# 🚀 Backpressure-aware system
class BackpressureQueue:
    def __init__(self, maxsize: int, high_water: int, low_water: int):
        self.queue = Queue(maxsize=maxsize)
        self.high_water = high_water
        self.low_water = low_water
        self.is_paused = False
        self.pause_event = asyncio.Event()
        self.pause_event.set()  # Start unpaused
    
    async def put(self, item):
        await self.queue.put(item)
        
        # 🌊 Check water levels
        if self.queue.qsize() >= self.high_water and not self.is_paused:
            print("  ⚠️ High water mark reached! Pausing producers...")
            self.is_paused = True
            self.pause_event.clear()
    
    async def get(self):
        item = await self.queue.get()
        
        # 🌊 Check if we can resume
        if self.queue.qsize() <= self.low_water and self.is_paused:
            print("  ✅ Low water mark reached! Resuming producers...")
            self.is_paused = False
            self.pause_event.set()
        
        return item
    
    async def wait_if_paused(self):
        await self.pause_event.wait()

# 🎮 Example usage
async def backpressure_producer(bp_queue: BackpressureQueue):
    for i in range(20):
        await bp_queue.wait_if_paused()
        await bp_queue.put(f"Item-{i} 📦")
        print(f"Produced: Item-{i}")
        await asyncio.sleep(0.1)
    
    await bp_queue.put(None)

async def slow_consumer(bp_queue: BackpressureQueue):
    while True:
        item = await bp_queue.get()
        if item is None:
            break
        
        print(f"  🐌 Slowly consuming: {item}")
        await asyncio.sleep(0.5)  # Slow consumer!

⚠️ Common Pitfalls and Solutions

😱 Pitfall 1: Forgetting task_done()

# ❌ Wrong way - Queue.join() will hang forever!
async def bad_consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        process(item)
        # Missing queue.task_done()! 😰

# ✅ Correct way - Always call task_done()!
async def good_consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            queue.task_done()  # Even for None! 🛡️
            break
        
        try:
            await process(item)
        finally:
            queue.task_done()  # Always mark done! ✅

🤯 Pitfall 2: Queue Size Deadlock

# ❌ Dangerous - Can deadlock with small queue!
async def deadlock_prone():
    queue = Queue(maxsize=1)  # Very small queue
    
    # Producer fills queue and blocks
    await queue.put("Item 1")
    await queue.put("Item 2")  # 💥 Blocks here if no consumer!

# ✅ Safe - Use appropriate queue size or handle blocking!
async def deadlock_free():
    queue = Queue(maxsize=10)  # Reasonable buffer
    
    # Or use put_nowait with error handling
    try:
        queue.put_nowait("Item")
    except asyncio.QueueFull:
        print("⚠️ Queue full! Handle appropriately")

🛠️ Best Practices

  1. 🎯 Use Type Hints: Be explicit about queue contents
  2. 📏 Size Queues Appropriately: Not too small (deadlock), not too large (memory)
  3. 🛡️ Always Handle Exceptions: Don’t let one bad item crash everything
  4. 📊 Monitor Queue Sizes: Log when queues get full or empty
  5. ✨ Signal Completion Clearly: Use None or a sentinel value consistently

🧪 Hands-On Exercise

🎯 Challenge: Build a Multi-Stage Image Processing Pipeline

Create an async image processing pipeline:

📋 Requirements:

  • ✅ Stage 1: Load image URLs from a list
  • 🎨 Stage 2: Download images concurrently
  • 🔧 Stage 3: Process images (resize, filter)
  • 💾 Stage 4: Save processed images
  • 📊 Track progress and statistics
  • 🚦 Implement rate limiting for downloads

🚀 Bonus Points:

  • Add retry logic for failed downloads
  • Implement progress reporting
  • Create a dashboard showing pipeline status

💡 Solution

🔍 Click to see solution
import asyncio
from asyncio import Queue
import aiohttp
from dataclasses import dataclass
from typing import Optional
import time

# 🎯 Our image processing pipeline!
@dataclass
class ImageJob:
    url: str
    image_data: Optional[bytes] = None
    processed_data: Optional[bytes] = None
    status: str = "pending"
    emoji: str = "📷"

class ImagePipeline:
    def __init__(self, download_limit: int = 3):
        self.url_queue = Queue(maxsize=20)
        self.download_queue = Queue(maxsize=10)
        self.process_queue = Queue(maxsize=10)
        self.stats = {
            "downloaded": 0,
            "processed": 0,
            "failed": 0,
            "start_time": time.time()
        }
        self.download_semaphore = asyncio.Semaphore(download_limit)
    
    # 📥 Stage 1: URL Producer
    async def url_producer(self, urls: list):
        print("🎯 URL Producer started!")
        for url in urls:
            job = ImageJob(url=url)
            await self.url_queue.put(job)
            print(f"  📥 Queued: {url}")
        
        # Signal completion
        for _ in range(3):  # Number of downloaders
            await self.url_queue.put(None)
    
    # 🌐 Stage 2: Image Downloader
    async def image_downloader(self, session: aiohttp.ClientSession, worker_id: int):
        print(f"🌐 Downloader {worker_id} started!")
        
        while True:
            job = await self.url_queue.get()
            
            if job is None:
                self.url_queue.task_done()
                await self.download_queue.put(None)
                break
            
            async with self.download_semaphore:  # Rate limiting
                try:
                    async with session.get(job.url) as response:
                        job.image_data = await response.read()
                        job.status = "downloaded"
                        job.emoji = "✅"
                        self.stats["downloaded"] += 1
                        print(f"  ✅ Worker {worker_id} downloaded: {job.url}")
                except Exception as e:
                    job.status = "failed"
                    job.emoji = "❌"
                    self.stats["failed"] += 1
                    print(f"  ❌ Worker {worker_id} failed: {e}")
                finally:
                    await self.download_queue.put(job)
                    self.url_queue.task_done()
    
    # 🎨 Stage 3: Image Processor
    async def image_processor(self):
        print("🎨 Image Processor started!")
        
        while True:
            job = await self.download_queue.get()
            
            if job is None:
                self.download_queue.task_done()
                await self.process_queue.put(None)
                break
            
            if job.status == "downloaded":
                # Simulate image processing
                await asyncio.sleep(0.1)
                job.processed_data = job.image_data  # In real app, process here
                job.status = "processed"
                job.emoji = "🎨"
                self.stats["processed"] += 1
                print(f"  🎨 Processed image from: {job.url}")
            
            await self.process_queue.put(job)
            self.download_queue.task_done()
    
    # 💾 Stage 4: Image Saver
    async def image_saver(self):
        print("💾 Image Saver started!")
        results = []
        
        while True:
            job = await self.process_queue.get()
            
            if job is None:
                self.process_queue.task_done()
                break
            
            if job.status == "processed":
                # Simulate saving
                results.append(job)
                print(f"  💾 Saved image from: {job.url}")
            
            self.process_queue.task_done()
        
        # Final report
        elapsed = time.time() - self.stats["start_time"]
        print(f"\n🎉 Pipeline Complete!")
        print(f"  📊 Downloaded: {self.stats['downloaded']}")
        print(f"  🎨 Processed: {self.stats['processed']}")
        print(f"  ❌ Failed: {self.stats['failed']}")
        print(f"  ⏱️ Time: {elapsed:.2f}s")
        print(f"  🚀 Rate: {self.stats['processed']/elapsed:.2f} images/sec")
    
    # 🚀 Run the pipeline
    async def run(self, urls: list):
        async with aiohttp.ClientSession() as session:
            tasks = [
                self.url_producer(urls),
                self.image_downloader(session, 1),
                self.image_downloader(session, 2),
                self.image_downloader(session, 3),
                self.image_processor(),
                self.image_saver()
            ]
            
            await asyncio.gather(*tasks)

# 🎮 Test it out!
async def main():
    urls = [
        f"https://picsum.photos/200/300?random={i}"
        for i in range(10)
    ]
    
    pipeline = ImagePipeline(download_limit=3)
    await pipeline.run(urls)

# asyncio.run(main())

🎓 Key Takeaways

You’ve learned so much! Here’s what you can now do:

  • ✅ Create asyncio queues with confidence 💪
  • ✅ Build producer-consumer systems that scale 🚀
  • ✅ Handle backpressure and rate limiting 🛡️
  • ✅ Debug common queue issues like a pro 🐛
  • ✅ Design multi-stage pipelines with asyncio! 🏗️

Remember: Asyncio queues are your friends for building concurrent systems. They help you write clean, scalable, and efficient code! 🤝

🤝 Next Steps

Congratulations! 🎉 You’ve mastered asyncio queues and the producer-consumer pattern!

Here’s what to do next:

  1. 💻 Practice with the exercises above
  2. 🏗️ Build a real-time data processing system
  3. 📚 Explore asyncio’s other coordination primitives
  4. 🌟 Share your concurrent creations with others!

Remember: Every concurrent programming expert was once a beginner. Keep coding, keep learning, and most importantly, have fun! 🚀

Happy concurrent coding! 🎉🚀✨

Previous Tutorial
📘 Asyncio Patterns: Gather and Wait
Next Tutorial
🚀 Asyncio Streams: Network Programming

More python Tutorials

Part 325

📘 Asyncio Patterns: Gather and Wait

💎Advanced • 25 min read
Part 326 - Currently Reading

📘 Asyncio Queues: Producer-Consumer

💎Advanced • 25 min read
Part 327

🚀 Asyncio Streams: Network Programming

💎Advanced • 25 min read
View all 365 python tutorials →

Tutorial Info

Series Progress
Part 326 of 365 89%
Category
🐍 python
Difficulty
💎Advanced
Duration
25 min read
Published
July 7, 2025