Как решал проблему ненадежности обработчиков в RabbitMQ?

Решение проблемы ненадежности обработчиков в RabbitMQ

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

Основные проблемы

1. Потеря сообщений — если рабочий упадет до подтверждения
2. Дублирование сообщений — если сообщение переобработано
3. Зависание рабочего — обработка не завершилась
4. Отравленные сообщения — сообщение вызывает ошибку в цикле

Решение 1: Manual Ack (Подтверждение вручную)

Вместо автоматического подтверждения, подтверждаем ПОСЛЕ успешной обработки:

import pika
import json

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

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

# Критично: auto_ack=False — требуем явное подтверждение
channel.basic_qos(prefetch_count=1)  # Даём рабочему по одному сообщению

def process_payment(ch, method, properties, body):
    try:
        payment = json.loads(body)
        print(f"Processing payment: {payment['id']}")
        
        # Симулируем обработку
        result = charge_card(payment['user_id'], payment['amount'])
        
        if result['success']:
            # ТОЛЬКО ПОСЛЕ успеха — подтверждаем
            ch.basic_ack(delivery_tag=method.delivery_tag)
            print(f"Payment {payment['id']} confirmed")
        else:
            # Если ошибка — negative ack (переопределяем в очередь)
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
            print(f"Payment {payment['id']} requeued")
    
    except Exception as e:
        print(f"Error processing payment: {e}")
        # Отправляем в очередь смерти или переопределяем
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
        # requeue=False = отправляет в dead letter queue

channel.basic_consume(
    queue='payment_queue',
    on_message_callback=process_payment,
    auto_ack=False  # Вручную подтверждаем
)

print('Waiting for messages...')
channel.start_consuming()

Решение 2: Prefetch Count (Контроль нагрузки)

Не даём рабочему слишком много сообщений одновременно:

# Неправильно — может забрать 1000 сообщений и упасть
channel.basic_qos(prefetch_count=1000)  # Берёт до 1000 сообщений

# Правильно — по одному сообщению
channel.basic_qos(prefetch_count=1)

# Если нужна пропускная способность
channel.basic_qos(prefetch_count=10)  # Максимум 10 одновременно

Решение 3: Timeout и Heartbeat

Отследить зависшего рабочего через heartbeat:

import pika
from pika import ConnectionParameters

# Настраиваем heartbeat для обнаружения мертвых соединений
credentials = pika.PlainCredentials('guest', 'guest')
params = ConnectionParameters(
    host='localhost',
    credentials=credentials,
    heartbeat=10,  # Проверяем каждые 10 секунд
    blocked_connection_timeout=30,  # Макс 30 сек на блокировку
    connection_attempts=3,  # Пытаемся 3 раза
    retry_delay=2,  # Задержка между попытками
)

connection = pika.BlockingConnection(params)

Решение 4: Dead Letter Queue (Очередь для отравленных сообщений)

Автоматически отправляем проблемные сообщения в отдельную очередь:

def setup_rabbitmq_with_dlq(channel):
    # Основная очередь с параметрами DLQ
    channel.queue_declare(
        queue='payment_queue',
        durable=True,
        arguments={
            'x-dead-letter-exchange': 'dlx_exchange',
            'x-message-ttl': 60000,  # 60 сек для переобработки
            'x-max-length': 10000,  # Максимум 10k сообщений
        }
    )
    
    # Очередь смерти для проблемных сообщений
    channel.exchange_declare(
        exchange='dlx_exchange',
        exchange_type='direct',
        durable=True
    )
    
    channel.queue_declare(
        queue='payment_queue_dlq',
        durable=True,
        arguments={
            'x-message-ttl': 86400000,  # 24 часа в DLQ
        }
    )
    
    channel.queue_bind(
        exchange='dlx_exchange',
        queue='payment_queue_dlq',
        routing_key='payment_queue'
    )

# Обработчик для DLQ — повторно анализирует проблемы
def process_dlq_message(ch, method, properties, body):
    payment = json.loads(body)
    print(f"DLQ: Analyzing failed payment {payment['id']}")
    
    # Алертим или логируем
    send_alert(f"Payment {payment['id']} failed 3 times")
    
    ch.basic_ack(delivery_tag=method.delivery_tag)

Решение 5: Retry с exponential backoff

Повторяем обработку с растущей задержкой:

import time
import json
from datetime import datetime, timedelta

class RabbitMQConsumer:
    def __init__(self, channel, queue_name, max_retries=3):
        self.channel = channel
        self.queue_name = queue_name
        self.max_retries = max_retries
    
    def process_with_retry(self, delivery_tag, body):
        # Читаем количество попыток из заголовков
        attempt = self.get_attempt_count()
        
        try:
            payment = json.loads(body)
            process_payment_critical(payment)
            
            # Успех — подтверждаем
            self.channel.basic_ack(delivery_tag=delivery_tag)
            
        except TransientError as e:
            # Ошибка которая может исчезнуть (например, сервис временно недоступен)
            if attempt < self.max_retries:
                # Отправляем обратно в очередь
                delay = 2 ** attempt  # 2, 4, 8, 16 секунд
                print(f"Retrying payment in {delay}s (attempt {attempt + 1})")
                
                self.channel.basic_nack(delivery_tag=delivery_tag, requeue=True)
                time.sleep(delay)
            else:
                # Исчерпали попытки — в DLQ
                self.channel.basic_nack(delivery_tag=delivery_tag, requeue=False)
                
        except PermanentError as e:
            # Ошибка которая не исправится (например, неверный платёж)
            print(f"Permanent error: {e}")
            self.channel.basic_nack(delivery_tag=delivery_tag, requeue=False)
    
    def get_attempt_count(self) -> int:
        # В реальности это читается из заголовков сообщения
        return 0

class TransientError(Exception):
    pass

class PermanentError(Exception):
    pass

Решение 6: Idempotent обработка (Идемпотентность)

Обработка одного сообщения дважды = один результат:

import hashlib
import json
from datetime import datetime, timedelta

class IdempotentPaymentProcessor:
    def __init__(self, cache_ttl_hours=24):
        self.processed_ids = {}  # В реальности это Redis
        self.cache_ttl = timedelta(hours=cache_ttl_hours)
    
    def process_payment(self, payment: dict) -> dict:
        # Генерируем идемпотентный ключ
        idempotency_key = payment.get('idempotency_key')
        
        if not idempotency_key:
            raise ValueError("idempotency_key is required")
        
        # Проверяем обработали ли это раньше
        if idempotency_key in self.processed_ids:
            result = self.processed_ids[idempotency_key]
            print(f"Already processed, returning cached result: {result}")
            return result
        
        # Обрабатываем в первый раз
        try:
            result = charge_card(
                user_id=payment['user_id'],
                amount=payment['amount'],
                idempotency_key=idempotency_key  # Передаём в платёжку систему
            )
            
            # Кэшируем результат
            self.processed_ids[idempotency_key] = result
            
            return result
        
        except Exception as e:
            # Если ошибка — не кэшируем, следующая попытка обработает
            raise

# Использование
def process_rabbit_message(ch, method, properties, body):
    payment = json.loads(body)
    processor = IdempotentPaymentProcessor()
    
    try:
        result = processor.process_payment(payment)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

Решение 7: Мониторинг и алерты

import logging
from prometheus_client import Counter, Histogram, Gauge
import time

# Метрики
processed_counter = Counter('rabbitmq_processed_total', 'Total processed messages')
failed_counter = Counter('rabbitmq_failed_total', 'Total failed messages')
queue_size = Gauge('rabbitmq_queue_size', 'Current queue size')
processing_time = Histogram('rabbitmq_processing_seconds', 'Processing time')

def process_payment_with_monitoring(ch, method, properties, body):
    start_time = time.time()
    
    try:
        payment = json.loads(body)
        
        # Обработка
        result = charge_card(payment['user_id'], payment['amount'])
        
        ch.basic_ack(delivery_tag=method.delivery_tag)
        processed_counter.inc()
        
    except Exception as e:
        logging.error(f"Processing failed: {e}", exc_info=True)
        failed_counter.inc()
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
    
    finally:
        duration = time.time() - start_time
        processing_time.observe(duration)
        
        # Алерт если обработка очень долгая
        if duration > 10:
            logging.warning(f"Slow processing: {duration}s")

Полный пример с best practices

import pika
import json
import logging
from functools import wraps

logger = logging.getLogger(__name__)

class RobustRabbitMQConsumer:
    def __init__(self, host='localhost', queue_name='payment_queue'):
        self.host = host
        self.queue_name = queue_name
        self.connection = None
        self.channel = None
    
    def connect(self):
        params = pika.ConnectionParameters(
            host=self.host,
            heartbeat=10,
            blocked_connection_timeout=30,
        )
        self.connection = pika.BlockingConnection(params)
        self.channel = self.connection.channel()
        
        # Инициализируем очереди
        self.setup_queues()
    
    def setup_queues(self):
        # Основная очередь
        self.channel.queue_declare(
            queue=self.queue_name,
            durable=True,
            arguments={
                'x-dead-letter-exchange': 'dlx',
            }
        )
        
        # Очередь для мертвых писем
        self.channel.exchange_declare('dlx', exchange_type='direct')
        self.channel.queue_declare(queue=f'{self.queue_name}_dlq', durable=True)
        self.channel.queue_bind(exchange='dlx', queue=f'{self.queue_name}_dlq')
        
        # Префетч для контроля нагрузки
        self.channel.basic_qos(prefetch_count=1)
    
    def start_consuming(self):
        self.channel.basic_consume(
            queue=self.queue_name,
            on_message_callback=self.handle_message,
            auto_ack=False
        )
        
        logger.info(f"Starting to consume from {self.queue_name}")
        self.channel.start_consuming()
    
    def handle_message(self, ch, method, properties, body):
        try:
            message = json.loads(body)
            logger.info(f"Processing message: {message.get('id')}")
            
            # Здесь ваша бизнес-логика
            self.process_payment(message)
            
            # Успех
            ch.basic_ack(delivery_tag=method.delivery_tag)
            logger.info(f"Message {message.get('id')} processed successfully")
        
        except Exception as e:
            logger.error(f"Error processing message", exc_info=True)
            # Отправляем обратно в очередь для retry
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
    
    def process_payment(self, payment: dict):
        # Реальная обработка платежа
        pass
    
    def close(self):
        if self.channel:
            self.channel.stop_consuming()
        if self.connection:
            self.connection.close()

# Использование
if __name__ == '__main__':
    consumer = RobustRabbitMQConsumer()
    try:
        consumer.connect()
        consumer.start_consuming()
    except KeyboardInterrupt:
        consumer.close()

Ключевые правила надежности

1. Manual ACK — подтверждаем только после успеха
2. Prefetch=1 — контролируем нагрузку
3. Heartbeat — обнаруживаем зависшие соединения
4. Dead Letter Queue — обрабатываем проблемные сообщения
5. Retry logic — повторяем с exponential backoff
6. Idempotency — одинаковый результат при повторе
7. Monitoring — логируем и алертим на ошибки