Есть ли ограничения в RabbitMQ?

Ограничения RabbitMQ

Да, RabbitMQ имеет ряд ограничений и подводных камней, о которых нужно знать при работе с ним. Это важные знания для production систем.

1. Объем памяти и дисковое пространство

Memory leak при очереди из одного consumer

# Проблема: Если consumer медленнее, чем producer
import pika

connection = pika.BlockingConnection()
channel = connection.channel()

channel.queue_declare(queue='heavy_work', durable=True)

# Producer быстро добавляет сообщения
for i in range(100000):
    channel.basic_publish(
        exchange='',
        routing_key='heavy_work',
        body=f'Message {i}' * 1000  # Большие сообщения
    )

# Consumer работает медленно
def slow_callback(ch, method, properties, body):
    import time
    time.sleep(5)  # Обработка 5 секунд
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(
    queue='heavy_work',
    on_message_callback=slow_callback
)

# Результат: RabbitMQ кушает память, очередь раздувается

Решение: Используйте prefetch для ограничения

# Ограничиваем, сколько сообщений может держать consumer одновременно
channel.basic_qos(prefetch_count=1)  # По одному сообщению за раз

# Теперь RabbitMQ не будет загружать новые сообщения,
# пока consumer не обработает текущее

2. Максимальный размер сообщения

FRAME_MAX

# По умолчанию в RabbitMQ максимальный размер фрейма — 131KB
# Если попытаться отправить больше — ошибка

# Плохо: слишком большое сообщение
large_data = b'x' * (200 * 1024)  # 200KB

try:
    channel.basic_publish(
        exchange='',
        routing_key='queue',
        body=large_data
    )
except pika.exceptions.MethodNotImplemented:
    print('Сообщение слишком большое!')

Решение: Увеличьте FRAME_MAX при запуске RabbitMQ

# docker-compose.yml
services:
  rabbitmq:
    environment:
      RABBITMQ_FRAME_MAX: "2097152"  # 2MB вместо 131KB

Ор используйте сжатие:

import gzip
import json

data = json.dumps({'huge': 'object'}).encode()
compressed = gzip.compress(data)

channel.basic_publish(
    exchange='',
    routing_key='queue',
    body=compressed,
    properties=pika.BasicProperties(
        content_encoding='gzip'
    )
)

3. Гарантии доставки

No guarantee без подтверждения

# ❌ Неправильно: нет гарантии доставки

def callback(ch, method, properties, body):
    print(f'Обработал: {body}')
    # Забыли подтвердить! Если процесс упадет —
    # сообщение потеряется

channel.basic_consume(queue='queue', on_message_callback=callback)

Правильно: подтверждение

# ✅ Правильно: explicit acknowledgment

def callback(ch, method, properties, body):
    try:
        process_message(body)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        # Если ошибка — отправляем в dead letter или переочередь
        ch.basic_nack(
            delivery_tag=method.delivery_tag,
            requeue=True  # Вернуть в очередь
        )

channel.basic_consume(queue='queue', on_message_callback=callback)

4. Ограничение числа очередей

# Проблема: Если создать миллионы очередей
# RabbitMQ будет использовать больше памяти

# Плохо: динамическое создание очередей для каждого пользователя
for user_id in range(1_000_000):
    channel.queue_declare(queue=f'user_{user_id}_queue')
    # Миллион очередей! RabbitMQ упадет

# Хорошо: используйте topic exchanges с routing keys
# Одна очередь с pattern matching

channel.exchange_declare(
    exchange='user_events',
    exchange_type='topic',
    durable=True
)

# Подписываемся на все события пользователя
# Вместо миллиона очередей
channel.queue_bind(
    queue='notifications',
    exchange='user_events',
    routing_key='user.*.notifications'
)

5. Retention и TTL

Сообщения остаются в памяти

# Проблема: Если никто не слушает, сообщения копятся

# Решение 1: TTL на очередь
channel.queue_declare(
    queue='ephemeral_queue',
    arguments={
        'x-message-ttl': 60000  # 60 секунд
    }
)

# Решение 2: TTL на сообщение
channel.basic_publish(
    exchange='',
    routing_key='queue',
    body=b'message',
    properties=pika.BasicProperties(
        expiration='60000'  # Это сообщение живет 60 сек
    )
)

6. Dead Letter Exchange (для обработки ошибок)

# Проблема: Некорректные сообщения зависают

# Решение: Dead Letter Exchange для problematic messages

channel.exchange_declare(
    exchange='dlx',
    exchange_type='direct'
)

channel.queue_declare(
    queue='dead_letters',
    durable=True
)

channel.queue_bind(
    exchange='dlx',
    queue='dead_letters',
    routing_key='failed'
)

# Основная очередь отправляет отклоненные сообщения в DLX
channel.queue_declare(
    queue='main_queue',
    arguments={
        'x-dead-letter-exchange': 'dlx',
        'x-dead-letter-routing-key': 'failed'
    }
)

def callback(ch, method, properties, body):
    try:
        process(body)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception:
        # Нежелательное сообщение отправляется в DLX
        ch.basic_nack(
            delivery_tag=method.delivery_tag,
            requeue=False  # Не возвращать в основную очередь
        )

7. Количество connections

# Проблема: Слишком много открытых соединений

# ❌ Плохо: создаем новое соединение для каждого сообщения

def send_message(message):
    connection = pika.BlockingConnection()  # Новое соединение!
    channel = connection.channel()
    channel.basic_publish(...)
    connection.close()

for msg in messages:
    send_message(msg)  # Создаем 1000 соединений

# ✅ Хорошо: переиспользуем соединение

connection = pika.BlockingConnection()
channel = connection.channel()

for msg in messages:
    channel.basic_publish(...)

connection.close()

8. Persisted vs in-memory

# Проблема: Если RabbitMQ упадет, данные потеряются

# ❌ Неправильно: очередь не persisted
channel.queue_declare(queue='temp_queue')  # durable=False

# ✅ Правильно: durable очередь на диске
channel.queue_declare(
    queue='important_queue',
    durable=True  # Сохраняется на диск
)

# И durable сообщения
channel.basic_publish(
    exchange='',
    routing_key='queue',
    body=b'message',
    properties=pika.BasicProperties(
        delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE
    )
)

Практический пример production setup

import pika
from datetime import datetime

connection = pika.BlockingConnection()
channel = connection.channel()

# Durable очередь
channel.queue_declare(
    queue='notifications',
    durable=True,
    arguments={
        'x-message-ttl': 86400000,  # 24 часа
        'x-max-length': 100000,      # Максимум 100k сообщений
        'x-dead-letter-exchange': 'notifications_dlx'
    }
)

channel.basic_qos(prefetch_count=10)  # Обработаем 10 одновременно

def process_notification(ch, method, properties, body):
    try:
        print(f'Processing: {body}')
        # Обработка
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        print(f'Error: {e}')
        ch.basic_nack(
            delivery_tag=method.delivery_tag,
            requeue=False  # В DLX для анализа
        )

channel.basic_consume(
    queue='notifications',
    on_message_callback=process_notification
)

print('Listening...')
channel.start_consuming()

Ключевые выводы

• Prefetch count — критично для управления памятью
• Durable queues и persistent messages — для надежности
• TTL и max-length — для контроля размера
• Dead Letter Exchange — для обработки ошибок
• Reuse connections — не создавайте новое соединение для каждого сообщения
• Подтверждения — используйте explicit acknowledgments