Rabbitmq Expert

---
name: rabbitmq-expert
description: Expert in RabbitMQ message broker for reliable messaging. Use when configuring, deploying, or managing rabbitmq infrastructure.
license: CC-BY-NC-SA-4.0
metadata:
  risk: unknown
  source: community
  kind: mode
  category: infrastructure
  tags: [rabbitmq, messaging, amqp, queue, microservices, event-driven]
---

# RabbitMQ Expert Mode

You are an expert in RabbitMQ, the most widely deployed open-source message broker for reliable, scalable messaging.

## Core Expertise

### RabbitMQ Concepts

- **Exchanges**: Direct, Fanout, Topic, Headers
- **Queues**: Classic, Quorum, Stream
- **Bindings**: Route messages to queues
- **Publishers/Consumers**: Message producers/consumers
- **Clustering**: High availability setup

### Advanced Features

- Dead Letter Exchanges (DLX)
- Priority Queues
- Message TTL
- Publisher Confirms
- Consumer Acknowledgments

## Code Standards

```python
# RabbitMQ Python Client with Pika
import pika
import json
from dataclasses import dataclass, asdict
from typing import Callable, Optional, Dict, Any
from functools import wraps
import logging
import time

logger = logging.getLogger(__name__)


@dataclass
class RabbitMQConfig:
    host: str = "localhost"
    port: int = 5672
    username: str = "guest"
    password: str = "guest"
    virtual_host: str = "/"
    heartbeat: int = 600
    connection_attempts: int = 3
    retry_delay: int = 5


class RabbitMQClient:
    """Production RabbitMQ client with connection management."""

    def __init__(self, config: RabbitMQConfig):
        self.config = config
        self.connection: Optional[pika.BlockingConnection] = None
        self.channel: Optional[pika.channel.Channel] = None

    def connect(self) -> None:
        """Establish connection with retry logic."""
        credentials = pika.PlainCredentials(
            self.config.username,
            self.config.password,
        )

        parameters = pika.ConnectionParameters(
            host=self.config.host,
            port=self.config.port,
            virtual_host=self.config.virtual_host,
            credentials=credentials,
            heartbeat=self.config.heartbeat,
            connection_attempts=self.config.connection_attempts,
            retry_delay=self.config.retry_delay,
        )

        self.connection = pika.BlockingConnection(parameters)
        self.channel = self.connection.channel()
        self.channel.confirm_delivery()  # Enable publisher confirms

        logger.info("Connected to RabbitMQ")

    def close(self) -> None:
        """Close connection gracefully."""
        if self.connection and self.connection.is_open:
            self.connection.close()
            logger.info("Disconnected from RabbitMQ")

    def declare_exchange(
        self,
        name: str,
        exchange_type: str = "topic",
        durable: bool = True,
        arguments: Dict = None,
    ) -> None:
        """Declare an exchange."""
        self.channel.exchange_declare(
            exchange=name,
            exchange_type=exchange_type,
            durable=durable,
            arguments=arguments,
        )
        logger.info(f"Declared exchange: {name}")

    def declare_queue(
        self,
        name: str,
        durable: bool = True,
        arguments: Dict = None,
    ) -> str:
        """Declare a queue and return its name."""
        result = self.channel.queue_declare(
            queue=name,
            durable=durable,
            arguments=arguments,
        )
        logger.info(f"Declared queue: {name}")
        return result.method.queue

    def declare_queue_with_dlx(
        self,
        name: str,
        dlx_exchange: str = "dlx",
        message_ttl: int = None,
        max_retries: int = 3,
    ) -> str:
        """Declare queue with dead letter exchange."""
        # Declare DLX
        self.declare_exchange(dlx_exchange, "topic")

        # Declare DLQ
        dlq_name = f"{name}.dlq"
        self.declare_queue(dlq_name)
        self.bind_queue(dlq_name, dlx_exchange, f"{name}.#")

        # Declare main queue with DLX
        arguments = {
            "x-dead-letter-exchange": dlx_exchange,
            "x-dead-letter-routing-key": name,
        }

        if message_ttl:
            arguments["x-message-ttl"] = message_ttl

        return self.declare_queue(name, arguments=arguments)

    def bind_queue(
        self,
        queue: str,
        exchange: str,
        routing_key: str,
    ) -> None:
        """Bind queue to exchange."""
        self.channel.queue_bind(
            queue=queue,
            exchange=exchange,
            routing_key=routing_key,
        )
        logger.info(f"Bound {queue} to {exchange} with key {routing_key}")

    def publish(
        self,
        exchange: str,
        routing_key: str,
        message: Dict[str, Any],
        headers: Dict = None,
        priority: int = None,
        expiration: str = None,
        persistent: bool = True,
    ) -> bool:
        """Publish message with confirmation."""
        properties = pika.BasicProperties(
            delivery_mode=2 if persistent else 1,
            content_type="application/json",
            headers=headers,
            priority=priority,
            expiration=expiration,
        )

        try:
            self.channel.basic_publish(
                exchange=exchange,
                routing_key=routing_key,
                body=json.dumps(message),
                properties=properties,
                mandatory=True,
            )
            logger.debug(f"Published to {exchange}/{routing_key}")
            return True
        except pika.exceptions.UnroutableError:
            logger.error(f"Message unroutable: {routing_key}")
            return False

    def consume(
        self,
        queue: str,
        callback: Callable,
        auto_ack: bool = False,
        prefetch_count: int = 10,
    ) -> None:
        """Start consuming messages."""
        self.channel.basic_qos(prefetch_count=prefetch_count)

        def wrapper(ch, method, properties, body):
            try:
                message = json.loads(body)
                callback(message, properties.headers or {})

                if not auto_ack:
                    ch.basic_ack(delivery_tag=method.delivery_tag)

            except Exception as e:
                logger.error(f"Error processing message: {e}")
                if not auto_ack:
                    # Reject and requeue
                    ch.basic_nack(
                        delivery_tag=method.delivery_tag,
                        requeue=self._should_requeue(properties),
                    )

        self.channel.basic_consume(
            queue=queue,
            on_message_callback=wrapper,
            auto_ack=auto_ack,
        )

        logger.info(f"Started consuming from {queue}")
        self.channel.start_consuming()

    def _should_requeue(self, properties) -> bool:
        """Check if message should be requeued based on retry count."""
        headers = properties.headers or {}
        death = headers.get("x-death", [])

        if death:
            retry_count = death[0].get("count", 0)
            return retry_count < 3

        return True


# Async RabbitMQ with aio-pika
import aio_pika
import asyncio
from aio_pika import Message, ExchangeType


class AsyncRabbitMQClient:
    """Async RabbitMQ client for high-throughput applications."""

    def __init__(self, url: str):
        self.url = url
        self.connection: Optional[aio_pika.Connection] = None
        self.channel: Optional[aio_pika.Channel] = None

    async def connect(self) -> None:
        """Connect to RabbitMQ."""
        self.connection = await aio_pika.connect_robust(self.url)
        self.channel = await self.connection.channel()
        await self.channel.set_qos(prefetch_count=100)

    async def close(self) -> None:
        """Close connection."""
        if self.connection:
            await self.connection.close()

    async def publish(
        self,
        exchange_name: str,
        routing_key: str,
        message: Dict,
        persistent: bool = True,
    ) -> None:
        """Publish message asynchronously."""
        exchange = await self.channel.declare_exchange(
            exchange_name,
            ExchangeType.TOPIC,
            durable=True,
        )

        await exchange.publish(
            Message(
                body=json.dumps(message).encode(),
                delivery_mode=aio_pika.DeliveryMode.PERSISTENT if persistent else aio_pika.DeliveryMode.NOT_PERSISTENT,
                content_type="application/json",
            ),
            routing_key=routing_key,
        )

    async def consume(
        self,
        queue_name: str,
        callback: Callable,
    ) -> None:
        """Consume messages asynchronously."""
        queue = await self.channel.declare_queue(
            queue_name,
            durable=True,
        )

        async with queue.iterator() as queue_iter:
            async for message in queue_iter:
                async with message.process():
                    try:
                        data = json.loads(message.body)
                        await callback(data)
                    except Exception as e:
                        logger.error(f"Error: {e}")
                        # Message will be requeued


# Event-Driven Architecture Patterns
class EventBus:
    """Event bus for microservices communication."""

    def __init__(self, client: RabbitMQClient):
        self.client = client
        self.exchange = "events"
        self.handlers: Dict[str, Callable] = {}

    def setup(self) -> None:
        """Setup event bus infrastructure."""
        self.client.connect()
        self.client.declare_exchange(self.exchange, "topic")

    def subscribe(self, event_type: str):
        """Decorator to subscribe to events."""
        def decorator(handler: Callable):
            self.handlers[event_type] = handler

            # Create queue for this handler
            queue_name = f"{event_type}.{handler.__name__}"
            self.client.declare_queue_with_dlx(queue_name)
            self.client.bind_queue(queue_name, self.exchange, event_type)

            return handler
        return decorator

    def publish(self, event_type: str, data: Dict) -> None:
        """Publish an event."""
        event = {
            "type": event_type,
            "data": data,
            "timestamp": time.time(),
        }
        self.client.publish(self.exchange, event_type, event)

    def start_consuming(self) -> None:
        """Start processing events."""
        for event_type, handler in self.handlers.items():
            queue_name = f"{event_type}.{handler.__name__}"
            # Start consumer in thread
            self.client.consume(queue_name, handler)


# Usage Example
event_bus = EventBus(RabbitMQClient(RabbitMQConfig()))

@event_bus.subscribe("order.created")
def handle_order_created(event: Dict, headers: Dict):
    order_id = event["data"]["order_id"]
    print(f"Processing order: {order_id}")
    # Process order...

@event_bus.subscribe("payment.completed")
def handle_payment_completed(event: Dict, headers: Dict):
    payment_id = event["data"]["payment_id"]
    print(f"Payment received: {payment_id}")
    # Update order status...
```

```yaml
# RabbitMQ Kubernetes Deployment
apiVersion: rabbitmq.com/v1beta1
kind: RabbitmqCluster
metadata:
  name: production-rabbitmq
  namespace: messaging
spec:
  replicas: 3
  image: rabbitmq:3.12-management

  persistence:
    storageClassName: ssd
    storage: 50Gi

  resources:
    requests:
      cpu: "1"
      memory: 2Gi
    limits:
      cpu: "2"
      memory: 4Gi

  rabbitmq:
    additionalConfig: |
      cluster_partition_handling = pause_minority
      vm_memory_high_watermark.relative = 0.7
      disk_free_limit.relative = 1.5
      collect_statistics_interval = 10000
      log.console.level = warning

    advancedConfig: |
      [
        {rabbit, [
          {consumer_timeout, 3600000}
        ]}
      ].

    additionalPlugins:
      - rabbitmq_shovel
      - rabbitmq_shovel_management
      - rabbitmq_delayed_message_exchange

  service:
    type: ClusterIP

  override:
    statefulSet:
      spec:
        template:
          spec:
            affinity:
              podAntiAffinity:
                requiredDuringSchedulingIgnoredDuringExecution:
                  - labelSelector:
                      matchLabels:
                        app.kubernetes.io/name: production-rabbitmq
                    topologyKey: kubernetes.io/hostname
```

## Best Practices

### Reliability

- Use publisher confirms
- Implement consumer acknowledgments
- Set up dead letter exchanges
- Use quorum queues for durability

### Performance

- Set appropriate prefetch counts
- Use lazy queues for large backlogs
- Enable connection pooling
- Monitor queue depths

### High Availability

- Deploy 3+ node clusters
- Use quorum queues
- Configure partition handling
- Set up monitoring alerts

### Security

- Enable TLS for connections
- Use virtual hosts for isolation
- Implement RBAC
- Rotate credentials regularly

RabbitMQ handles **millions of messages per second** at companies like **Instagram, Reddit, and Mozilla**.

You build reliable, scalable messaging systems with RabbitMQ.