Какие знаешь стратегии для обнаружения ошибок в RabbitMQ?

Стратегии обнаружения ошибок в RabbitMQ

RabbitMQ — критическая система обмена сообщениями, требующая надёжного мониторинга и обработки ошибок. Вот основные стратегии:

1. Message Acknowledgments (Подтверждения)

Auto-Ack (Небезопасно)

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

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

def callback(ch, method, properties, body):
    try:
        print(f"Processing: {body}")
        # Если здесь ошибка, сообщение потеряется!
    except Exception as e:
        print(f"Error: {e}")

# auto_ack=True — опасно!
channel.basic_consume(queue='tasks', on_message_callback=callback, auto_ack=True)
channel.start_consuming()

Manual Ack (Безопасно)

def callback(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=True)

channel.basic_consume(queue='tasks', on_message_callback=callback, auto_ack=False)
channel.start_consuming()

2. Dead Letter Exchange (DLX)

Перенаправление ошибочных сообщений:

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Основная очередь
channel.exchange_declare(exchange='main_exchange', exchange_type='direct', durable=True)
channel.queue_declare(queue='main_queue', durable=True, arguments={
    'x-dead-letter-exchange': 'dlx_exchange',
    'x-message-ttl': 60000,  # 60 сек TTL
})
channel.queue_bind(exchange='main_exchange', queue='main_queue', routing_key='task')

# Dead Letter Exchange
channel.exchange_declare(exchange='dlx_exchange', exchange_type='direct', durable=True)
channel.queue_declare(queue='dlx_queue', durable=True)
channel.queue_bind(exchange='dlx_exchange', queue='dlx_queue', routing_key='task')

# Обработка ошибок
def error_handler(ch, method, properties, body):
    print(f"Failed message: {body}")
    # Логирование, алерты, повторная обработка

channel.basic_consume(queue='dlx_queue', on_message_callback=error_handler, auto_ack=False)
channel.start_consuming()

3. Retry Logic (Повторные попытки)

import pika
import json

MAX_RETRIES = 3

def process_message(ch, method, properties, body):
    try:
        data = json.loads(body)
        # Обработка
        result = risky_operation(data)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        retry_count = properties.headers.get('x-retry-count', 0) if properties.headers else 0
        
        if retry_count < MAX_RETRIES:
            # Переотправить с увеличенным счётчиком
            new_headers = (properties.headers or {}).copy()
            new_headers['x-retry-count'] = retry_count + 1
            
            channel.basic_publish(
                exchange='main_exchange',
                routing_key='task',
                body=body,
                properties=pika.BasicProperties(
                    headers=new_headers,
                    delivery_mode=2  # persistent
                )
            )
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
        else:
            # Отправить в DLX
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

4. Publisher Confirms

Убедиться что сообщение приняла очередь:

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Включить Publisher Confirms
channel.confirm_delivery()

try:
    channel.basic_publish(
        exchange='main_exchange',
        routing_key='task',
        body='data',
        properties=pika.BasicProperties(delivery_mode=2)
    )
    # Будет исключение если отправка не подтверждена
    print("Message sent successfully")
except pika.exceptions.UnroutableError:
    print("Message could not be routed!")
except pika.exceptions.AMQPChannelError as e:
    print(f"Channel error: {e}")

5. Connection Resilience

import pika
from pika import connection

class RabbitMQConnection:
    def __init__(self, host='localhost'):
        self.host = host
        self.connection = None
        self.channel = None
        self.connect()
    
    def connect(self):
        try:
            credentials = pika.PlainCredentials('guest', 'guest')
            params = pika.ConnectionParameters(
                host=self.host,
                credentials=credentials,
                socket_timeout=5,
                connection_attempts=3,
                retry_delay=2,
                heartbeat=600  # 10 минут
            )
            self.connection = pika.BlockingConnection(params)
            self.channel = self.connection.channel()
            print("Connected to RabbitMQ")
        except pika.exceptions.AMQPConnectionError as e:
            print(f"Connection failed: {e}")
            raise
    
    def close(self):
        if self.connection and not self.connection.is_closed:
            self.connection.close()

6. Consumer Prefetch (Qos)

# Обрабатывай по одному сообщению за раз
channel.basic_qos(prefetch_count=1)

# Для высоконагруженных сценариев
channel.basic_qos(prefetch_count=10)

7. Monitoring и Logging

import logging
from pythonjsonlogger import jsonlogger

logger = logging.getLogger()
logHandler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter()
logHandler.setFormatter(formatter)
logger.addHandler(logHandler)

def monitored_callback(ch, method, properties, body):
    logger.info('message_received', extra={
        'delivery_tag': method.delivery_tag,
        'message': body.decode(),
        'queue': method.routing_key
    })
    
    try:
        process_message(body)
        ch.basic_ack(delivery_tag=method.delivery_tag)
        logger.info('message_processed')
    except Exception as e:
        logger.error('message_failed', extra={'error': str(e)})
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

8. Circuit Breaker Pattern

from circuitbreaker import circuit

@circuit(failure_threshold=5, recovery_timeout=60)
def process_with_rabbitmq(message):
    # Если 5 ошибок подряд, circuit открывается на 60 сек
    return risky_operation(message)

def callback(ch, method, properties, body):
    try:
        process_with_rabbitmq(body)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

9. Проверка здоровья

import pika
import time

def health_check():
    try:
        connection = pika.BlockingConnection(pika.ConnectionParameters('localhost', socket_timeout=2))
        connection.close()
        return True
    except Exception as e:
        print(f"RabbitMQ unreachable: {e}")
        return False

# Регулярная проверка
while True:
    if not health_check():
        send_alert("RabbitMQ is down!")
    time.sleep(30)

Лучшие практики

• Всегда используй Manual Ack вместо auto_ack
• Настроь TTL и DLX для обработки вечных ошибок
• Логируй все ошибки с контекстом
• Используй Publisher Confirms для критических операций
• Мониторь queue depth и connection errors
• Настроь heartbeat для долгоживущих соединений
• Используй circuit breaker для graceful degradation

Вывод: надёжная система на RabbitMQ требует многоуровневого подхода к обработке ошибок.