Message Queues

Message Queue Patterns ()

Asynchronous communication patterns for distributed systems using RabbitMQ, Redis Streams, Kafka, and FastStream.

Overview

• Decoupling services in microservices architecture
• Implementing pub/sub and work queue patterns
• Building event-driven systems with reliable delivery
• Load leveling and buffering between services
• Task distribution across multiple workers
• High-throughput event streaming (Kafka)

Quick Reference

FastStream: Unified API ( Recommended)

# pip install faststream[kafka,rabbit,redis]
from faststream import FastStream
from faststream.kafka import KafkaBroker
from pydantic import BaseModel

broker = KafkaBroker("localhost:9092")
app = FastStream(broker)

class OrderCreated(BaseModel):
    order_id: str
    customer_id: str
    total: float

@broker.subscriber("orders.created")
async def handle_order(event: OrderCreated):
    """Automatic Pydantic validation and deserialization."""
    print(f"Processing order {event.order_id}")
    await process_order(event)

@broker.publisher("orders.processed")
async def publish_processed(order_id: str) -> dict:
    return {"order_id": order_id, "status": "processed"}

# Run with: faststream run app:app

Kafka Producer (aiokafka)

from aiokafka import AIOKafkaProducer
import json

class KafkaPublisher:
    def __init__(self, bootstrap_servers: str):
        self.bootstrap_servers = bootstrap_servers
        self._producer: AIOKafkaProducer | None = None

    async def start(self):
        self._producer = AIOKafkaProducer(
            bootstrap_servers=self.bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode(),
            acks="all",  # Wait for all replicas
            enable_idempotence=True,  # Exactly-once semantics
        )
        await self._producer.start()

    async def publish(
        self,
        topic: str,
        value: dict,
        key: str | None = None,
    ):
        await self._producer.send_and_wait(
            topic,
            value=value,
            key=key.encode() if key else None,
        )

    async def stop(self):
        await self._producer.stop()

Kafka Consumer with Consumer Group

from aiokafka import AIOKafkaConsumer
from aiokafka.errors import OffsetOutOfRangeError

class KafkaConsumer:
    def __init__(
        self,
        topic: str,
        group_id: str,
        bootstrap_servers: str,
    ):
        self.consumer = AIOKafkaConsumer(
            topic,
            bootstrap_servers=bootstrap_servers,
            group_id=group_id,
            auto_offset_reset="earliest",
            enable_auto_commit=False,  # Manual commit for reliability
            value_deserializer=lambda v: json.loads(v.decode()),
        )

    async def consume(self, handler):
        await self.consumer.start()
        try:
            async for msg in self.consumer:
                try:
                    await handler(msg.value, msg.key, msg.partition)
                    await self.consumer.commit()
                except Exception as e:
                    # Handle or send to DLQ
                    await self.send_to_dlq(msg, e)
        finally:
            await self.consumer.stop()

RabbitMQ Publisher

import aio_pika
from aio_pika import Message, DeliveryMode

class RabbitMQPublisher:
    def __init__(self, url: str):
        self.url = url
        self._connection = None
        self._channel = None

    async def connect(self):
        self._connection = await aio_pika.connect_robust(self.url)
        self._channel = await self._connection.channel()
        await self._channel.set_qos(prefetch_count=10)

    async def publish(self, exchange: str, routing_key: str, message: dict):
        exchange_obj = await self._channel.get_exchange(exchange)
        await exchange_obj.publish(
            Message(
                body=json.dumps(message).encode(),
                delivery_mode=DeliveryMode.PERSISTENT,
                content_type="application/json"
            ),
            routing_key=routing_key
        )

RabbitMQ Consumer with Retry

class RabbitMQConsumer:
    async def consume(self, queue_name: str, handler, max_retries: int = 3):
        queue = await self._channel.get_queue(queue_name)
        async with queue.iterator() as queue_iter:
            async for message in queue_iter:
                async with message.process(requeue=False):
                    try:
                        body = json.loads(message.body.decode())
                        await handler(body)
                    except Exception as e:
                        retry_count = message.headers.get("x-retry-count", 0)
                        if retry_count < max_retries:
                            await self.publish(exchange, routing_key, body,
                                headers={"x-retry-count": retry_count + 1})
                        else:
                            await self.publish("dlx", "failed", body,
                                headers={"x-error": str(e)})

Redis Streams Consumer Group

import redis.asyncio as redis

class RedisStreamConsumer:
    def __init__(self, url: str, stream: str, group: str, consumer: str):
        self.redis = redis.from_url(url)
        self.stream, self.group, self.consumer = stream, group, consumer

    async def setup(self):
        try:
            await self.redis.xgroup_create(self.stream, self.group, "0", mkstream=True)
        except redis.ResponseError as e:
            if "BUSYGROUP" not in str(e): raise

    async def consume(self, handler):
        while True:
            messages = await self.redis.xreadgroup(
                groupname=self.group, consumername=self.consumer,
                streams={self.stream: ">"}, count=10, block=5000
            )
            for stream, stream_messages in messages:
                for message_id, data in stream_messages:
                    try:
                        await handler(message_id, data)
                        await self.redis.xack(self.stream, self.group, message_id)
                    except Exception:
                        pass  # Message redelivered on restart

"Just Use Postgres" Pattern

# For simpler use cases - Postgres LISTEN/NOTIFY + FOR UPDATE SKIP LOCKED
from sqlalchemy import text

class PostgresQueue:
    """Simple queue using Postgres - good for moderate throughput."""

    async def publish(self, db: AsyncSession, channel: str, payload: dict):
        await db.execute(
            text("SELECT pg_notify(:channel, :payload)"),
            {"channel": channel, "payload": json.dumps(payload)}
        )

    async def get_next_job(self, db: AsyncSession) -> dict | None:
        """Get next job with advisory lock."""
        result = await db.execute(text("""
            SELECT id, payload FROM job_queue
            WHERE status = 'pending'
            ORDER BY created_at
            FOR UPDATE SKIP LOCKED
            LIMIT 1
        """))
        return result.first()

Key Decisions

When to Choose Each

┌────────────────────────────────────────────────────────────────────────┐
│                     DECISION FLOWCHART                                  │
├────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│  Need > 50K msg/s?                                                     │
│      YES → Kafka (partitioned, replicated)                             │
│      NO ↓                                                              │
│                                                                         │
│  Need complex routing (topic, headers)?                                │
│      YES → RabbitMQ (exchanges, bindings)                              │
│      NO ↓                                                              │
│                                                                         │
│  Need real-time + simple?                                              │
│      YES → Redis Streams (XREAD, consumer groups)                      │
│      NO ↓                                                              │
│                                                                         │
│  Already using Postgres + < 10K msg/s?                                 │
│      YES → Postgres (LISTEN/NOTIFY + FOR UPDATE SKIP LOCKED)           │
│      NO → Re-evaluate requirements                                     │
│                                                                         │
└────────────────────────────────────────────────────────────────────────┘

Anti-Patterns (FORBIDDEN)

# NEVER process without acknowledgment
async for msg in consumer:
    process(msg)  # Message lost on failure!

# NEVER use sync calls in handlers
def handle(msg):
    requests.post(url, data=msg)  # Blocks event loop!

# NEVER ignore ordering when required
await publish("orders", {"order_id": "123"})  # No partition key!

# NEVER store large payloads
await publish("files", {"content": large_bytes})  # Use URL reference!

# NEVER skip dead letter handling
except Exception:
    pass  # Failed messages vanish!

# NEVER choose Kafka for simple task queue
# RabbitMQ or Redis is simpler for work distribution

# NEVER use Redis Streams when strict delivery matters
# Use RabbitMQ or Kafka for guaranteed delivery

Related Skills

• outbox-pattern - Transactional outbox for reliable publishing
• background-jobs - Celery/ARQ task processing
• streaming-api-patterns - SSE/WebSocket real-time
• observability-monitoring - Queue metrics and alerting
• event-sourcing - Event store and CQRS patterns

Capability Details

kafka-streaming

Keywords: kafka, aiokafka, partition, consumer group, exactly-once, offset Solves:

• How do I set up Kafka producers/consumers?
• Partition key selection for ordering
• Exactly-once semantics with idempotence
• Consumer group rebalancing

rabbitmq-messaging

Keywords: rabbitmq, amqp, aio-pika, exchange, queue, topic, fanout, routing Solves:

• How do I set up RabbitMQ pub/sub?
• Exchange types and queue binding
• Dead letter queue configuration
• Message persistence and acknowledgment

redis-streams

Keywords: redis streams, xadd, xread, xreadgroup, consumer group, xack Solves:

• How do I use Redis Streams?
• Consumer group setup and message claiming
• Stream trimming and retention
• At-least-once delivery patterns

faststream-framework

Keywords: faststream, unified api, pydantic, asyncapi, broker Solves:

• Unified API for Kafka/RabbitMQ/Redis
• Automatic Pydantic serialization
• AsyncAPI documentation generation
• Dependency injection for handlers

postgres-queue

Keywords: postgres queue, listen notify, skip locked, simple queue Solves:

• When to use Postgres instead of dedicated queue
• LISTEN/NOTIFY for pub/sub
• FOR UPDATE SKIP LOCKED for job queue

Message Queue Implementation Checklist

Verification checklist for production-ready message queue deployments.

Message Durability

Queue Configuration

• [ ] Queues declared as durable=True
• [ ] Messages published with delivery_mode=PERSISTENT
• [ ] Queue auto_delete=False for persistent queues
• [ ] Appropriate x-message-ttl set (prevent unbounded growth)
• [ ] x-max-length configured with overflow policy

Persistence

• [ ] RabbitMQ: Disk nodes configured (not RAM-only)
• [ ] RabbitMQ: Mirrored queues for HA (or quorum queues)
• [ ] Redis: AOF persistence enabled (appendonly yes)
• [ ] Redis: Appropriate appendfsync setting (everysec or always)
• [ ] Backup strategy for queue data

---

Consumer Error Handling

Acknowledgment

• [ ] Manual acknowledgment enabled (not auto-ack)
• [ ] requeue=False on permanent failures (prevents infinite loops)
• [ ] Proper exception handling around message processing
• [ ] Context manager used for automatic ack on success

Error Classification

• [ ] Transient errors trigger retry (network, timeout)
• [ ] Permanent errors route to DLQ (validation, business logic)
• [ ] Unknown errors logged with full context
• [ ] Error metrics emitted for monitoring

Graceful Shutdown

• [ ] SIGTERM handler stops accepting new messages
• [ ] In-flight messages complete before shutdown
• [ ] Unacked messages redelivered to other consumers
• [ ] Connection closed cleanly

---

Retry Strategies

Configuration

• [ ] Retry count limit defined (typically 3-5)
• [ ] Exponential backoff implemented
• [ ] Jitter added to prevent thundering herd
• [ ] Max delay capped (e.g., 30 seconds)

Retry Tracking

• [ ] Retry count stored in message headers
• [ ] Original timestamp preserved
• [ ] Error history attached to message
• [ ] Correlation ID maintained across retries

Dead Letter Handling

• [ ] DLX configured for exhausted retries
• [ ] DLQ consumer processes failed messages
• [ ] Alerting on DLQ growth
• [ ] DLQ retention policy defined

---

Monitoring Setup

Metrics

• [ ] Queue depth (messages waiting)
• [ ] Consumer count and utilization
• [ ] Message publish/consume rates
• [ ] Acknowledgment latency
• [ ] Retry rate by queue
• [ ] DLQ message count

Alerts

• [ ] Queue depth threshold (e.g., >1000 messages)
• [ ] Consumer down alert
• [ ] High retry rate (>5% of messages)
• [ ] DLQ growth rate
• [ ] Connection failures

Logging

• [ ] Structured logging with correlation IDs
• [ ] Message lifecycle events (published, consumed, acked, rejected)
• [ ] Error details with stack traces
• [ ] Performance timing (processing duration)

---

Security

Authentication

• [ ] Credentials not hardcoded (use environment variables)
• [ ] Separate credentials per service
• [ ] TLS enabled for connections
• [ ] Virtual hosts for tenant isolation (RabbitMQ)

Authorization

• [ ] Minimal permissions per service (read/write separation)
• [ ] No management access from application code
• [ ] Audit logging for admin operations

---

Performance

Producer

• [ ] Connection pooling configured
• [ ] Batch publishing where appropriate
• [ ] Async publishing for non-blocking operation
• [ ] Publisher confirms enabled for critical messages

Consumer

• [ ] Prefetch count tuned (start with 10, adjust based on load)
• [ ] Concurrent consumers for parallel processing
• [ ] Async handlers (no blocking I/O)
• [ ] Batch processing where applicable

Capacity

• [ ] Load testing completed
• [ ] Auto-scaling configured for consumers
• [ ] Resource limits set (memory, disk)
• [ ] Horizontal scaling plan documented

---

Testing

Unit Tests

• [ ] Message serialization/deserialization
• [ ] Error handling logic
• [ ] Retry logic with mocked failures
• [ ] Idempotency handling

Integration Tests

• [ ] End-to-end message flow
• [ ] Consumer failure and recovery
• [ ] DLQ routing
• [ ] Cluster failover (if applicable)

Chaos Testing

• [ ] Consumer crashes during processing
• [ ] Network partition between producer/broker
• [ ] Broker restart during operation
• [ ] Disk full scenario

---

Pre-Production Checklist

Before going live:

• [ ] All durability checks passed
• [ ] Error handling verified with failure injection
• [ ] Retry strategy tested with realistic failures
• [ ] Monitoring dashboards created
• [ ] Alerts configured and tested
• [ ] Runbook documented for common issues
• [ ] Load test completed at 2x expected volume
• [ ] Rollback procedure documented

Message Queue Examples

Complete, production-ready code examples for message queue implementations.

RabbitMQ Producer/Consumer

Full Producer Implementation

import asyncio
import json
import uuid
from datetime import datetime, timezone
from typing import Any

import aio_pika
from aio_pika import Message, DeliveryMode, ExchangeType


class RabbitMQProducer:
    """Production-ready RabbitMQ producer with connection management."""

    def __init__(self, url: str):
        self.url = url
        self._connection: aio_pika.RobustConnection | None = None
        self._channel: aio_pika.Channel | None = None
        self._exchanges: dict[str, aio_pika.Exchange] = {}

    async def connect(self):
        """Establish robust connection with auto-reconnect."""
        self._connection = await aio_pika.connect_robust(
            self.url,
            reconnect_interval=5,
            fail_fast=False
        )
        self._channel = await self._connection.channel()

        # Enable publisher confirms for reliable delivery
        await self._channel.set_qos(prefetch_count=10)

    async def declare_exchange(
        self,
        name: str,
        exchange_type: ExchangeType = ExchangeType.TOPIC
    ) -> aio_pika.Exchange:
        """Declare and cache exchange."""
        if name not in self._exchanges:
            self._exchanges[name] = await self._channel.declare_exchange(
                name=name,
                type=exchange_type,
                durable=True
            )
        return self._exchanges[name]

    async def publish(
        self,
        exchange: str,
        routing_key: str,
        payload: dict[str, Any],
        correlation_id: str | None = None,
        headers: dict | None = None
    ) -> str:
        """Publish message with full metadata."""
        message_id = str(uuid.uuid4())
        correlation_id = correlation_id or message_id

        message = Message(
            body=json.dumps(payload).encode(),
            delivery_mode=DeliveryMode.PERSISTENT,
            content_type="application/json",
            message_id=message_id,
            correlation_id=correlation_id,
            timestamp=datetime.now(timezone.utc),
            headers=headers or {}
        )

        exchange_obj = await self.declare_exchange(exchange)
        await exchange_obj.publish(message, routing_key=routing_key)

        return message_id

    async def close(self):
        """Clean shutdown."""
        if self._connection:
            await self._connection.close()


# Usage
async def main():
    producer = RabbitMQProducer("amqp://guest:guest@localhost/")
    await producer.connect()

    message_id = await producer.publish(
        exchange="orders",
        routing_key="order.created",
        payload={
            "order_id": "ORD-123",
            "customer_id": "CUST-456",
            "items": [{"sku": "PROD-1", "qty": 2}]
        }
    )
    print(f"Published message: {message_id}")

    await producer.close()

Full Consumer Implementation

import asyncio
import json
import signal
from typing import Callable, Any
from contextlib import asynccontextmanager

import aio_pika
from aio_pika import IncomingMessage


class RabbitMQConsumer:
    """Production-ready consumer with graceful shutdown."""

    def __init__(self, url: str, queue_name: str, prefetch: int = 10):
        self.url = url
        self.queue_name = queue_name
        self.prefetch = prefetch
        self._connection: aio_pika.RobustConnection | None = None
        self._channel: aio_pika.Channel | None = None
        self._should_stop = False

    async def connect(self):
        self._connection = await aio_pika.connect_robust(self.url)
        self._channel = await self._connection.channel()
        await self._channel.set_qos(prefetch_count=self.prefetch)

    async def consume(
        self,
        handler: Callable[[dict], Any],
        max_retries: int = 3
    ):
        """Consume messages with retry logic."""
        queue = await self._channel.get_queue(self.queue_name)

        async with queue.iterator() as queue_iter:
            async for message in queue_iter:
                if self._should_stop:
                    break

                await self._process_message(message, handler, max_retries)

    async def _process_message(
        self,
        message: IncomingMessage,
        handler: Callable,
        max_retries: int
    ):
        """Process single message with error handling."""
        async with message.process(requeue=False):
            try:
                body = json.loads(message.body.decode())
                await handler(body)

            except json.JSONDecodeError as e:
                # Permanent failure - send to DLQ
                await self._reject_to_dlq(message, f"Invalid JSON: {e}")

            except Exception as e:
                retry_count = message.headers.get("x-retry-count", 0)

                if retry_count < max_retries:
                    await self._retry_message(message, retry_count + 1)
                else:
                    await self._reject_to_dlq(message, str(e))

    async def _retry_message(self, message: IncomingMessage, retry_count: int):
        """Republish message for retry with backoff."""
        delay = min(1000 * (2 ** retry_count), 30000)  # Exponential, max 30s

        # Publish to delay exchange (requires retry topology setup)
        exchange = await self._channel.get_exchange(f"retry.{delay}ms")
        await exchange.publish(
            aio_pika.Message(
                body=message.body,
                headers={
                    **dict(message.headers),
                    "x-retry-count": retry_count,
                    "x-original-routing-key": message.routing_key
                }
            ),
            routing_key=self.queue_name
        )

    async def _reject_to_dlq(self, message: IncomingMessage, error: str):
        """Send to dead letter queue with error info."""
        # Message will be routed to DLX configured on queue
        await message.reject(requeue=False)

    def stop(self):
        """Signal consumer to stop."""
        self._should_stop = True

    async def close(self):
        if self._connection:
            await self._connection.close()


# Usage with graceful shutdown
async def main():
    consumer = RabbitMQConsumer(
        url="amqp://guest:guest@localhost/",
        queue_name="orders"
    )
    await consumer.connect()

    # Setup signal handlers
    loop = asyncio.get_event_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, consumer.stop)

    async def handle_order(order: dict):
        print(f"Processing order: {order['order_id']}")
        # Simulate processing
        await asyncio.sleep(0.1)

    try:
        await consumer.consume(handle_order)
    finally:
        await consumer.close()

---

Redis Streams Consumer Groups

Full Implementation

import asyncio
import json
import signal
from datetime import datetime, timezone
from typing import Callable, Any

import redis.asyncio as redis


class RedisStreamConsumer:
    """Production-ready Redis Streams consumer with consumer groups."""

    def __init__(
        self,
        url: str,
        stream: str,
        group: str,
        consumer: str,
        batch_size: int = 10,
        block_ms: int = 5000
    ):
        self.url = url
        self.stream = stream
        self.group = group
        self.consumer = consumer
        self.batch_size = batch_size
        self.block_ms = block_ms
        self._redis: redis.Redis | None = None
        self._should_stop = False

    async def connect(self):
        """Connect and setup consumer group."""
        self._redis = redis.from_url(self.url)

        # Create consumer group if not exists
        try:
            await self._redis.xgroup_create(
                self.stream,
                self.group,
                id="0",  # Start from beginning
                mkstream=True
            )
        except redis.ResponseError as e:
            if "BUSYGROUP" not in str(e):
                raise

    async def consume(self, handler: Callable[[str, dict], Any]):
        """Consume messages with acknowledgment."""
        while not self._should_stop:
            try:
                messages = await self._redis.xreadgroup(
                    groupname=self.group,
                    consumername=self.consumer,
                    streams={self.stream: ">"},
                    count=self.batch_size,
                    block=self.block_ms
                )

                if not messages:
                    continue

                for stream_name, stream_messages in messages:
                    for message_id, data in stream_messages:
                        await self._process_message(
                            message_id.decode(),
                            data,
                            handler
                        )

            except redis.ConnectionError:
                await asyncio.sleep(1)
                await self.connect()

    async def _process_message(
        self,
        message_id: str,
        data: dict,
        handler: Callable
    ):
        """Process and acknowledge message."""
        try:
            # Decode bytes to strings
            decoded = {
                k.decode() if isinstance(k, bytes) else k:
                v.decode() if isinstance(v, bytes) else v
                for k, v in data.items()
            }

            await handler(message_id, decoded)

            # Acknowledge successful processing
            await self._redis.xack(self.stream, self.group, message_id)

        except Exception as e:
            # Log error - message will be redelivered on restart
            print(f"Error processing {message_id}: {e}")

    async def claim_pending(self, min_idle_ms: int = 60000):
        """Claim messages stuck with dead consumers."""
        pending = await self._redis.xpending_range(
            self.stream,
            self.group,
            min="-",
            max="+",
            count=100
        )

        for entry in pending:
            message_id = entry["message_id"]
            idle_time = entry["time_since_delivered"]

            if idle_time > min_idle_ms:
                claimed = await self._redis.xclaim(
                    self.stream,
                    self.group,
                    self.consumer,
                    min_idle_time=min_idle_ms,
                    message_ids=[message_id]
                )
                print(f"Claimed {len(claimed)} messages")

    def stop(self):
        self._should_stop = True

    async def close(self):
        if self._redis:
            await self._redis.close()


class RedisStreamProducer:
    """Simple Redis Streams producer."""

    def __init__(self, url: str, stream: str, maxlen: int = 10000):
        self.url = url
        self.stream = stream
        self.maxlen = maxlen
        self._redis: redis.Redis | None = None

    async def connect(self):
        self._redis = redis.from_url(self.url)

    async def publish(self, data: dict[str, Any]) -> str:
        """Add message to stream."""
        message_id = await self._redis.xadd(
            self.stream,
            data,
            maxlen=self.maxlen,
            approximate=True
        )
        return message_id.decode()

    async def close(self):
        if self._redis:
            await self._redis.close()


# Usage
async def main():
    # Producer
    producer = RedisStreamProducer("redis://localhost", "events")
    await producer.connect()

    msg_id = await producer.publish({
        "event_type": "user.created",
        "user_id": "123",
        "timestamp": datetime.now(timezone.utc).isoformat()
    })
    print(f"Published: {msg_id}")

    # Consumer
    consumer = RedisStreamConsumer(
        url="redis://localhost",
        stream="events",
        group="processors",
        consumer="worker-1"
    )
    await consumer.connect()

    async def handle_event(msg_id: str, data: dict):
        print(f"Processing {msg_id}: {data}")

    # Run consumer (would normally run indefinitely)
    asyncio.create_task(consumer.consume(handle_event))
    await asyncio.sleep(5)
    consumer.stop()

---

Error Handling Patterns

Transactional Outbox Pattern

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text


class OutboxPublisher:
    """Transactional outbox for reliable message publishing."""

    def __init__(self, db: AsyncSession, producer: RabbitMQProducer):
        self.db = db
        self.producer = producer

    async def publish_with_outbox(
        self,
        exchange: str,
        routing_key: str,
        payload: dict,
        business_operation: Callable
    ):
        """Execute business logic and queue message atomically."""
        async with self.db.begin():
            # Execute business operation
            result = await business_operation()

            # Store message in outbox table
            await self.db.execute(
                text("""
                    INSERT INTO outbox (exchange, routing_key, payload, status)
                    VALUES (:exchange, :routing_key, :payload, 'pending')
                """),
                {
                    "exchange": exchange,
                    "routing_key": routing_key,
                    "payload": json.dumps(payload)
                }
            )

            return result

    async def process_outbox(self):
        """Background worker to publish pending messages."""
        while True:
            async with self.db.begin():
                # Get pending messages
                result = await self.db.execute(
                    text("""
                        SELECT id, exchange, routing_key, payload
                        FROM outbox
                        WHERE status = 'pending'
                        ORDER BY created_at
                        LIMIT 100
                        FOR UPDATE SKIP LOCKED
                    """)
                )

                for row in result:
                    try:
                        await self.producer.publish(
                            row.exchange,
                            row.routing_key,
                            json.loads(row.payload)
                        )

                        await self.db.execute(
                            text("UPDATE outbox SET status = 'published' WHERE id = :id"),
                            {"id": row.id}
                        )
                    except Exception as e:
                        await self.db.execute(
                            text("UPDATE outbox SET status = 'failed', error = :error WHERE id = :id"),
                            {"id": row.id, "error": str(e)}
                        )

            await asyncio.sleep(1)

---

Dead Letter Processing

DLQ Processor with Retry Decision

class DeadLetterProcessor:
    """Process dead letters with automated and manual retry options."""

    def __init__(self, consumer: RabbitMQConsumer, producer: RabbitMQProducer):
        self.consumer = consumer
        self.producer = producer

    async def process_dlq(self):
        """Consume DLQ and decide on retry or archive."""

        async def handle_dead_letter(message: dict):
            death_info = message.get("x-death", [{}])[0]
            reason = death_info.get("reason", "unknown")
            original_queue = death_info.get("queue", "unknown")

            if self._is_retriable(reason, message):
                # Re-publish to original queue
                await self.producer.publish(
                    exchange="main",
                    routing_key=original_queue,
                    payload=message.get("payload", {}),
                    headers={"x-dlq-retry": True}
                )
            else:
                # Archive for manual review
                await self._archive_message(message, reason)

        await self.consumer.consume(handle_dead_letter)

    def _is_retriable(self, reason: str, message: dict) -> bool:
        """Determine if message should be retried."""
        # Don't retry validation errors
        if "validation" in str(message.get("error", "")).lower():
            return False

        # Retry timeouts and transient failures
        retriable_reasons = ["expired", "maxlen"]
        return reason in retriable_reasons

    async def _archive_message(self, message: dict, reason: str):
        """Store in database for manual review."""
        # Implementation: store in dead_letters table
        pass

FastStream Patterns

FastStream provides a unified API for Kafka, RabbitMQ, NATS, and Redis Streams.

Installation

pip install faststream[kafka]      # Kafka only
pip install faststream[rabbit]     # RabbitMQ only
pip install faststream[redis]      # Redis Streams only
pip install faststream[nats]       # NATS only
pip install faststream[all]        # All brokers

Basic Kafka Application

from faststream import FastStream
from faststream.kafka import KafkaBroker
from pydantic import BaseModel

broker = KafkaBroker("localhost:9092")
app = FastStream(broker)

class OrderEvent(BaseModel):
    order_id: str
    customer_id: str
    total: float

@broker.subscriber("orders.created")
async def handle_order(event: OrderEvent):
    """Automatic Pydantic validation."""
    print(f"Processing order {event.order_id}")

@broker.publisher("orders.processed")
async def process_order(order_id: str) -> dict:
    return {"order_id": order_id, "status": "processed"}

RabbitMQ with Exchanges

from faststream import FastStream
from faststream.rabbit import RabbitBroker, RabbitExchange, RabbitQueue

broker = RabbitBroker("amqp://guest:guest@localhost/")
app = FastStream(broker)

# Define exchange and queue
orders_exchange = RabbitExchange("orders", type="topic")
orders_queue = RabbitQueue("order-processor", routing_key="orders.#")

@broker.subscriber(orders_queue, orders_exchange)
async def handle_order(data: dict):
    print(f"Received: {data}")

Dependency Injection

from faststream import Context, Depends

async def get_db():
    async with async_session() as session:
        yield session

@broker.subscriber("users.created")
async def handle_user(
    event: UserEvent,
    db = Depends(get_db),
    logger = Context(),  # Access logger from context
):
    await db.add(User(**event.dict()))
    logger.info(f"Created user {event.user_id}")

Multiple Brokers

from faststream import FastStream
from faststream.kafka import KafkaBroker
from faststream.rabbit import RabbitBroker

kafka = KafkaBroker("localhost:9092")
rabbit = RabbitBroker("amqp://localhost/")

app = FastStream(kafka, rabbit)

@kafka.subscriber("high-volume-events")
async def kafka_handler(data: dict):
    pass

@rabbit.subscriber("task-queue")
async def rabbit_handler(data: dict):
    pass

Testing

import pytest
from faststream.kafka import TestKafkaBroker

@pytest.fixture
def test_broker():
    return TestKafkaBroker(broker)

@pytest.mark.asyncio
async def test_order_handler(test_broker):
    async with test_broker:
        await test_broker.publish(
            {"order_id": "123", "customer_id": "456", "total": 99.99},
            topic="orders.created",
        )
        # Handler is automatically called
        # Add assertions based on side effects

AsyncAPI Documentation

FastStream auto-generates AsyncAPI documentation:

# Generate docs
from faststream.asyncapi import get_app_schema

schema = get_app_schema(app)
print(schema.to_yaml())

Access at http://localhost:8000/asyncapi when running with faststream run.

Kafka Patterns

Partition Key Selection

# Use aggregate ID for ordered events within same entity
await producer.publish(
    topic="orders",
    key=str(order.id),  # All order events on same partition
    value=event,
)

# Use customer ID for customer-centric ordering
await producer.publish(
    topic="customer-events",
    key=str(customer_id),  # All customer events ordered
    value=event,
)

# Use random key for maximum parallelism (no ordering)
import uuid
await producer.publish(
    topic="logs",
    key=str(uuid.uuid4()),  # Distributed across partitions
    value=log_event,
)

Consumer Group Patterns

from aiokafka import AIOKafkaConsumer

# Multiple consumers in same group = load balancing
consumer1 = AIOKafkaConsumer(
    "orders",
    group_id="order-processors",  # Same group
    bootstrap_servers="localhost:9092",
)

consumer2 = AIOKafkaConsumer(
    "orders",
    group_id="order-processors",  # Same group - partitions split
    bootstrap_servers="localhost:9092",
)

# Different groups = broadcast
analytics_consumer = AIOKafkaConsumer(
    "orders",
    group_id="analytics",  # Different group - gets all messages
    bootstrap_servers="localhost:9092",
)

Exactly-Once Semantics

producer = AIOKafkaProducer(
    bootstrap_servers="localhost:9092",
    acks="all",                    # Wait for all replicas
    enable_idempotence=True,       # Deduplicate on broker
    transactional_id="my-app-1",   # Enable transactions
)

await producer.start()

# Transactional produce
async with producer.transaction():
    await producer.send("topic1", value=b"msg1")
    await producer.send("topic2", value=b"msg2")
    # Both committed atomically or neither

Offset Management

consumer = AIOKafkaConsumer(
    "orders",
    group_id="processors",
    enable_auto_commit=False,  # Manual commit for reliability
    auto_offset_reset="earliest",  # Start from beginning if no offset
)

async for msg in consumer:
    try:
        await process(msg)
        # Commit AFTER successful processing
        await consumer.commit()
    except Exception:
        # Don't commit - message will be reprocessed
        pass

Topic Naming Conventions

<domain>.<entity>.<event-type>

Examples:
- orders.order.created
- orders.order.shipped
- customers.customer.registered
- inventory.stock.updated

RabbitMQ Patterns Reference

Detailed implementation patterns for RabbitMQ message queuing.

Exchange Types

Direct Exchange

Routes messages to queues based on exact routing key match.

# Declare direct exchange
await channel.declare_exchange(
    name="notifications",
    type=aio_pika.ExchangeType.DIRECT,
    durable=True
)

# Bind queue with specific routing key
await queue.bind(exchange="notifications", routing_key="email")
await queue.bind(exchange="notifications", routing_key="sms")

# Publish to specific routing key
await exchange.publish(
    message=Message(body=b"Send email"),
    routing_key="email"  # Only email queue receives this
)

Use cases:

• Task routing to specific workers
• Service-to-service direct communication
• Notifications by type (email, sms, push)

Topic Exchange

Routes based on pattern matching with wildcards.

# Declare topic exchange
await channel.declare_exchange(
    name="events",
    type=aio_pika.ExchangeType.TOPIC,
    durable=True
)

# Bind with wildcard patterns
await queue.bind(exchange="events", routing_key="order.created")
await queue.bind(exchange="events", routing_key="order.*")       # order.created, order.updated
await queue.bind(exchange="events", routing_key="*.created")     # order.created, user.created
await queue.bind(exchange="events", routing_key="audit.#")       # audit.*, audit.*.*, etc.

# Publish with specific routing key
await exchange.publish(message, routing_key="order.created.us")

Wildcards:

• * matches exactly one word
• # matches zero or more words

Use cases:

• Event-driven architectures
• Multi-tenant message routing
• Audit logging (capture all events)

Fanout Exchange

Broadcasts to all bound queues (ignores routing key).

# Declare fanout exchange
await channel.declare_exchange(
    name="broadcasts",
    type=aio_pika.ExchangeType.FANOUT,
    durable=True
)

# All bound queues receive every message
await queue1.bind(exchange="broadcasts")
await queue2.bind(exchange="broadcasts")

# Routing key is ignored
await exchange.publish(message, routing_key="")  # All queues get this

Use cases:

• Cache invalidation across services
• Real-time notifications to all subscribers
• System-wide announcements

---

Queue Configuration

Durable vs Transient

# Durable queue - survives broker restart
durable_queue = await channel.declare_queue(
    name="orders",
    durable=True,      # Queue definition persisted
    auto_delete=False  # Queue persists when consumers disconnect
)

# Transient queue - for temporary consumers
temp_queue = await channel.declare_queue(
    name="",           # Auto-generated name
    exclusive=True,    # Only this connection can consume
    auto_delete=True   # Deleted when connection closes
)

Queue Arguments

await channel.declare_queue(
    name="tasks",
    durable=True,
    arguments={
        # TTL for messages in queue
        "x-message-ttl": 86400000,  # 24 hours in ms

        # Max queue length
        "x-max-length": 10000,
        "x-overflow": "reject-publish",  # or "drop-head"

        # Dead letter exchange
        "x-dead-letter-exchange": "dlx",
        "x-dead-letter-routing-key": "failed.tasks",

        # Single active consumer
        "x-single-active-consumer": True,

        # Queue expiry (if unused)
        "x-expires": 3600000  # 1 hour
    }
)

Priority Queues

# Declare priority queue
await channel.declare_queue(
    name="priority-tasks",
    durable=True,
    arguments={"x-max-priority": 10}  # Priority levels 0-10
)

# Publish with priority
await exchange.publish(
    Message(
        body=b"Urgent task",
        priority=10  # Highest priority
    ),
    routing_key="tasks"
)

---

Dead Letter Queues

Setup DLX

# 1. Declare dead letter exchange
dlx = await channel.declare_exchange(
    name="dlx",
    type=aio_pika.ExchangeType.DIRECT,
    durable=True
)

# 2. Declare dead letter queue
dlq = await channel.declare_queue(
    name="failed-messages",
    durable=True
)
await dlq.bind(dlx, routing_key="failed")

# 3. Configure main queue to use DLX
main_queue = await channel.declare_queue(
    name="tasks",
    durable=True,
    arguments={
        "x-dead-letter-exchange": "dlx",
        "x-dead-letter-routing-key": "failed"
    }
)

DLQ Consumer for Analysis

async def process_dead_letters(message: aio_pika.IncomingMessage):
    async with message.process():
        # Extract failure metadata
        headers = message.headers
        death_info = headers.get("x-death", [{}])[0]

        original_queue = death_info.get("queue")
        reason = death_info.get("reason")  # rejected, expired, maxlen
        death_count = death_info.get("count", 1)

        # Log for analysis
        logger.error(
            "Dead letter received",
            original_queue=original_queue,
            reason=reason,
            death_count=death_count,
            body=message.body.decode()
        )

        # Optional: Store in database for dashboard
        await db.execute(
            "INSERT INTO dead_letters (queue, reason, body, created_at) VALUES ($1, $2, $3, NOW())",
            original_queue, reason, message.body.decode()
        )

Retry with DLX

async def setup_retry_topology(channel):
    """Create retry topology with exponential backoff."""

    # Main exchange
    main_exchange = await channel.declare_exchange("main", "direct", durable=True)

    # Retry exchanges with increasing delays
    for delay in [1000, 5000, 30000]:  # 1s, 5s, 30s
        retry_exchange = await channel.declare_exchange(
            f"retry.{delay}ms", "direct", durable=True
        )
        retry_queue = await channel.declare_queue(
            f"retry.{delay}ms.queue",
            durable=True,
            arguments={
                "x-message-ttl": delay,
                "x-dead-letter-exchange": "main"
            }
        )
        await retry_queue.bind(retry_exchange)

    # Final DLX for exhausted retries
    dlx = await channel.declare_exchange("dlx", "direct", durable=True)
    dlq = await channel.declare_queue("dead-letters", durable=True)
    await dlq.bind(dlx)

---

Consumer Acknowledgments

Manual Acknowledgment

async def consume_with_ack(queue_name: str, handler):
    queue = await channel.get_queue(queue_name)

    async with queue.iterator() as queue_iter:
        async for message in queue_iter:
            # Process without auto-ack
            async with message.process(requeue=False):
                try:
                    body = json.loads(message.body.decode())
                    await handler(body)
                    # Message auto-acked when context exits normally
                except Exception as e:
                    # Explicitly reject and send to DLX
                    await message.reject(requeue=False)
                    logger.error(f"Message rejected: {e}")

Batched Acknowledgment

async def consume_batched(queue_name: str, handler, batch_size: int = 100):
    """Acknowledge in batches for better throughput."""
    queue = await channel.get_queue(queue_name)
    pending_acks = []

    async with queue.iterator() as queue_iter:
        async for message in queue_iter:
            try:
                body = json.loads(message.body.decode())
                await handler(body)
                pending_acks.append(message)

                # Batch ack
                if len(pending_acks) >= batch_size:
                    await channel.basic_ack(
                        delivery_tag=pending_acks[-1].delivery_tag,
                        multiple=True  # Ack all messages up to this tag
                    )
                    pending_acks.clear()

            except Exception as e:
                await message.reject(requeue=False)

Prefetch (QoS)

# Set prefetch count to limit unacked messages
await channel.set_qos(prefetch_count=10)

# Higher prefetch = better throughput, but risk of message loss on crash
# Lower prefetch = safer, but slower
# Recommendation: Start with 10, tune based on monitoring

---

Best Practices

Message Design

# Good: Small, self-contained messages
message = {
    "event_id": str(uuid.uuid4()),
    "event_type": "order.created",
    "timestamp": datetime.now(timezone.utc).isoformat(),
    "data": {
        "order_id": "123",
        "customer_id": "456"
    }
}

# Bad: Large payloads
message = {
    "file_content": base64.encode(large_file)  # Use URL reference instead
}

Idempotent Consumers

async def idempotent_handler(message: dict):
    """Process message only if not already processed."""
    event_id = message["event_id"]

    # Check if already processed
    if await redis.get(f"processed:{event_id}"):
        logger.info(f"Skipping duplicate: {event_id}")
        return

    # Process message
    await process_order(message["data"])

    # Mark as processed (with TTL for cleanup)
    await redis.setex(f"processed:{event_id}", 86400, "1")

Connection Management

class RabbitMQConnection:
    """Robust connection with automatic reconnection."""

    def __init__(self, url: str):
        self.url = url
        self._connection = None
        self._channel = None

    async def connect(self):
        self._connection = await aio_pika.connect_robust(
            self.url,
            connection_class=aio_pika.RobustConnection,
            reconnect_interval=5,
            fail_fast=False
        )
        self._connection.add_close_callback(self._on_close)
        self._channel = await self._connection.channel()

    def _on_close(self, *args):
        logger.warning("RabbitMQ connection closed, will reconnect")

    async def close(self):
        if self._connection:
            await self._connection.close()