Паттерны использования и гарантии доставки в RabbitMQ

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


Work Queues: распределение нагрузки между потребителями

Work Queues (очереди задач) используются для распределения трудоемких задач между несколькими worker-процессами. Этот паттерн идеален для обработки фоновых задач, таких как генерация отчетов, обработка изображений или отправка email.

Ключевые характеристики:
Одна очередь, несколько потребителей
Каждое сообщение обрабатывается только одним потребителем
Балансировка нагрузки через настройку prefetch count

Реализация на Java с Spring AMQP

Producer:

@Service
public class TaskProducer {
    private final RabbitTemplate rabbitTemplate;
    
    @Value("${app.queues.task-queue}")
    private String taskQueue;
    
    public void sendTask(Task task) {
        rabbitTemplate.convertAndSend(taskQueue, task, message -> {
            // Установка приоритета задачи
            message.getMessageProperties().setPriority(task.getPriority());
            // Время жизни сообщения
            message.getMessageProperties().setExpiration("3600000"); // 1 час
            return message;
        });
    }
}

Consumer с ручными подтверждениями:

@Component
public class TaskWorker {
    
    @RabbitListener(queues = "${app.queues.task-queue}")
    public void processTask(Task task, Channel channel, 
                           @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            // Обработка задачи
            boolean success = executeTask(task);
            
            if (success) {
                // Подтверждение успешной обработки
                channel.basicAck(deliveryTag, false);
                log.info("Task {} processed successfully", task.getId());
            } else {
                // Отказ без повторной очереди (перемещение в DLQ)
                channel.basicNack(deliveryTag, false, false);
                log.error("Task {} failed, moved to DLQ", task.getId());
            }
        } catch (Exception e) {
            // При ошибке - повторная очередь
            channel.basicNack(deliveryTag, false, true);
            log.error("Error processing task {}, requeued", task.getId(), e);
        }
    }
    
    private boolean executeTask(Task task) {
        // Логика обработки задачи
        return true;
    }
}

Конфигурация для равномерного распределения:

@Configuration
public class WorkQueueConfig {
    
    @Bean
    public Queue taskQueue() {
        return QueueBuilder.durable("tasks.queue")
            .withArgument("x-max-priority", 10) // Поддержка приоритетов
            .withArgument("x-dead-letter-exchange", "dlx.exchange")
            .build();
    }
    
    @Bean
    public SimpleRabbitListenerContainerFactory workerFactory(
            ConnectionFactory connectionFactory) {
        
        SimpleRabbitListenerContainerFactory factory = 
            new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        
        // Критичная настройка для равномерного распределения
        factory.setPrefetchCount(1); // По одному сообщению на consumer
        factory.setConcurrentConsumers(3); // Три параллельных worker'а
        factory.setMaxConcurrentConsumers(10); // Автомасштабирование при нагрузке
        
        // Ручные подтверждения для контроля
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        
        return factory;
    }
}

#Java #middle #RabbitMQ

Publish/Subscribe: широковещательная рассылка

Publish/Subscribe (публикация/подписка) используется, когда одно сообщение должно быть доставлено множеству потребителей. Типичные сценарии: уведомления, аудит-логи, обновления кэша.

Архитектура:
Publisher отправляет сообщение в exchange
Exchange копирует сообщение во все привязанные очереди
Каждый consumer имеет свою собственную очередь

Реализация Fanout Exchange

Конфигурация обменника и очередей:

@Configuration
public class PubSubConfig {
    
    @Bean
    public FanoutExchange notificationsExchange() {
        return new FanoutExchange("notifications.fanout", true, false);
    }
    
    @Bean
    public Queue auditQueue() {
        return new Queue("notifications.audit.queue", true);
    }
    
    @Bean
    public Queue cacheQueue() {
        return new Queue("notifications.cache.queue", true);
    }
    
    @Bean
    public Queue analyticsQueue() {
        return new Queue("notifications.analytics.queue", true);
    }
    
    @Bean
    public Binding auditBinding() {
        return BindingBuilder.bind(auditQueue())
            .to(notificationsExchange());
    }
    
    @Bean
    public Binding cacheBinding() {
        return BindingBuilder.bind(cacheQueue())
            .to(notificationsExchange());
    }
    
    @Bean
    public Binding analyticsBinding() {
        return BindingBuilder.bind(analyticsQueue())
            .to(notificationsExchange());
    }
}

Publisher:

@Service
public class NotificationPublisher {
    private final RabbitTemplate rabbitTemplate;
    
    public void publishNotification(Notification notification) {
        // Отправка в fanout exchange - все подписчики получат копию
        rabbitTemplate.convertAndSend(
            "notifications.fanout", 
            "", // routing key игнорируется для fanout
            notification
        );
    }
}

Consumer для аудита (пример одного из многих):

@Component
public class AuditNotificationConsumer {
    
    @RabbitListener(queues = "notifications.audit.queue")
    public void handleNotification(Notification notification) {
        // Каждый consumer обрабатывает свою копию сообщения
        auditService.logEvent(notification);
    }
}

Routing/Topics: селективная подписка

Topic Exchange позволяет выполнять сложную маршрутизацию на основе routing key и patterns. Используется, когда потребители интересуются только определенными типами сообщений.

Pattern syntax:
* (звездочка) заменяет одно слово
# (решетка) заменяет ноль или более слов
Пример: stock.usd.* или stock.#

Реализация Topic Exchange

Конфигурация:

@Configuration
public class TopicRoutingConfig {
    
    @Bean
    public TopicExchange stockExchange() {
        return new TopicExchange("stock.topic", true, false);
    }
    
    @Bean
    public Queue usdStockQueue() {
        return new Queue("stock.usd.queue", true);
    }
    
    @Bean
    public Queue eurStockQueue() {
        return new Queue("stock.eur.queue", true);
    }
    
    @Bean
    public Queue allStockQueue() {
        return new Queue("stock.all.queue", true);
    }
    
    @Bean
    public Binding usdBinding() {
        // Будет получать: stock.usd.nyse, stock.usd.nasdaq
        return BindingBuilder.bind(usdStockQueue())
            .to(stockExchange())
            .with("stock.usd.*");
    }
    
    @Bean
    public Binding eurBinding() {
        // Будет получать: stock.eur.lse, stock.eur.euronext
        return BindingBuilder.bind(eurStockQueue())
            .to(stockExchange())
            .with("stock.eur.*");
    }
    
    @Bean
    public Binding allStocksBinding() {
        // Будет получать все сообщения о stock
        return BindingBuilder.bind(allStockQueue())
            .to(stockExchange())
            .with("stock.#");
    }
}

#Java #middle #RabbitMQ

Publisher с различными routing keys:

@Service
public class StockPublisher {
    private final RabbitTemplate rabbitTemplate;
    
    public void publishUSDPrice(String exchange, BigDecimal price) {
        String routingKey = "stock.usd." + exchange.toLowerCase();
        StockUpdate update = new StockUpdate("USD", exchange, price);
        
        rabbitTemplate.convertAndSend(
            "stock.topic", 
            routingKey, 
            update
        );
    }
    
    public void publishEURPrice(String exchange, BigDecimal price) {
        String routingKey = "stock.eur." + exchange.toLowerCase();
        StockUpdate update = new StockUpdate("EUR", exchange, price);
        
        rabbitTemplate.convertAndSend(
            "stock.topic", 
            routingKey, 
            update
        );
    }
}

RPC over AMQP: синхронные вызовы через асинхронный транспорт

Remote Procedure Call (RPC) поверх AMQP позволяет выполнять синхронные запросы-ответы через ассинхронную систему сообщений. Используется когда нужен немедленный ответ, но нельзя установить прямое соединение.

Механизм работы:
Клиент отправляет запрос с уникальным correlationId
Сервер обрабатывает запрос и отправляет ответ в очередь ответов
Клиент ожидает ответ с matching correlationId

Реализация RPC

Клиентская сторона:

@Service
public class CalculatorRpcClient {
    private final RabbitTemplate rabbitTemplate;
    
    public CalculatorRpcClient(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
        
        // Настройка reply listener
        this.rabbitTemplate.setReplyTimeout(30000);
        this.rabbitTemplate.setUseDirectReplyToContainer(false);
    }
    
    public BigDecimal calculate(CalculationRequest request) {
        // Создание correlation ID
        String correlationId = UUID.randomUUID().toString();
        
        // Отправка запроса и ожидание ответа
        CalculationResponse response = (CalculationResponse) 
            rabbitTemplate.convertSendAndReceive(
                "rpc.exchange",
                "calculator.rpc",
                request,
                message -> {
                    message.getMessageProperties()
                        .setCorrelationId(correlationId);
                    message.getMessageProperties()
                        .setReplyTo("amq.rabbitmq.reply-to"); // Временная очередь
                    return message;
                }
            );
        
        if (response == null) {
            throw new RuntimeException("RPC timeout");
        }
        
        return response.getResult();
    }
}

Серверная сторона:

@Component
public class CalculatorRpcServer {
    
    @RabbitListener(
        bindings = @QueueBinding(
            value = @Queue(value = "rpc.calculator.queue", durable = "true"),
            exchange = @Exchange(value = "rpc.exchange", type = ExchangeTypes.DIRECT),
            key = "calculator.rpc"
        )
    )
    public CalculationResponse calculate(CalculationRequest request,
                                        Message message) {
        
        String correlationId = message.getMessageProperties()
            .getCorrelationId();
        String replyTo = message.getMessageProperties()
            .getReplyTo();
        
        log.debug("Processing RPC request {} from {}", 
                 correlationId, replyTo);
        
        // Выполнение расчета
        BigDecimal result = performCalculation(request);
        
        // Возврат результата с сохранением correlationId
        CalculationResponse response = new CalculationResponse(result);
        
        // CorrelationId автоматически копируется в ответ
        return response;
    }
    
    private BigDecimal performCalculation(CalculationRequest request) {
        // Логика расчета
        return BigDecimal.ZERO;
    }
}

#Java #middle #RabbitMQ

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

At-most-once: риск потери сообщений

Механизм: Сообщение отправляется без подтверждений, сразу удаляется из очереди.

Использовать когда:
Потеря сообщений допустима
Высокая производительность критична
Система имеет механизмы компенсации

// Producer без подтверждений
@Bean
public RabbitTemplate atMostOnceTemplate(ConnectionFactory cf) {
    RabbitTemplate template = new RabbitTemplate(cf);
    template.setMandatory(false);
    
    // Отключение publisher confirms
    ((CachingConnectionFactory) cf)
        .setPublisherConfirmType(CachingConnectionFactory.ConfirmType.NONE);
    
    return template;
}

// Consumer с auto-ack
@RabbitListener(queues = "at.most.once.queue")
public void handleAutoAck(Message message) {
    // Сообщение уже удалено из очереди при доставке
    // При падении здесь - сообщение потеряно
    processMessage(message);
}

At-least-once: риск дублирования

Механизм: Гарантирует доставку минимум один раз, но возможны дубликаты.

Реализация с ручными подтверждениями:

@Component
public class AtLeastOnceConsumer {
    
    @RabbitListener(queues = "orders.queue")
    public void processOrder(Order order, Channel channel,
                            @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        
        try {
            // 1. Обработка заказа
            processOrderSafely(order);
            
            // 2. Подтверждение после успешной обработки
            channel.basicAck(tag, false);
            
            log.info("Order {} processed and acknowledged", order.getId());
            
        } catch (Exception e) {
            // 3. При ошибке - повторная очередь
            channel.basicNack(tag, false, true);
            log.error("Order processing failed, requeued", e);
            
            // Идемпотентность критична!
            // При повторной обработке должна быть сохранена консистентность
        }
    }
    
    private void processOrderSafely(Order order) {
        // Идемпотентная обработка
        if (orderRepository.existsById(order.getId())) {
            log.warn("Duplicate order {}, skipping", order.getId());
            return;
        }
        
        orderRepository.save(order);
        inventoryService.reserveItems(order);
    }
}

#Java #middle #RabbitMQ

Exactly-once: сложная реализация

Реальность: RabbitMQ не предоставляет exactly-once гарантий на уровне протокола. Достигается комбинацией механизмов.

Паттерн для pseudo-exactly-once:

@Service
@Transactional
public class ExactlyOnceProcessor {
    
    private final RabbitTemplate rabbitTemplate;
    private final OrderRepository orderRepository;
    
    public void processWithExactlyOnceSemantics(OrderMessage message) {
        // 1. Проверка идемпотентности
        if (processedMessageRepository.existsById(message.getId())) {
            return; // Уже обработано
        }
        
        // 2. Сохранение в outbox паттерн
        OrderOutbox outbox = new OrderOutbox();
        outbox.setMessageId(message.getId());
        outbox.setOrderData(message.getPayload());
        outbox.setStatus(OutboxStatus.PROCESSING);
        
        orderRepository.save(outbox);
        
        // 3. Бизнес-логика в транзакции
        Order order = createOrder(message);
        orderRepository.save(order);
        
        // 4. Обновление статуса в той же транзакции
        outbox.setStatus(OutboxStatus.PROCESSED);
        orderRepository.save(outbox);
        
        // 5. Отправка подтверждения (после коммита транзакции)
        // RabbitTemplate работает после завершения транзакции
        rabbitTemplate.convertAndSend(
            "order.processed.exchange",
            "order.processed",
            new OrderProcessedEvent(order.getId())
        );
        
        // 6. Идемпотентность на стороне получателя подтверждения
    }
    
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
    public void afterCommit(OrderProcessedEvent event) {
        // Только после успешного коммита
        rabbitTemplate.convertAndSend(
            "order.completed.queue",
            event
        );
    }
}

Механизмы надежности

Publisher Confirms

Механизм: Брокер подтверждает получение сообщения.

Реализация:

@Configuration
public class PublisherConfirmConfig {
    
    @Bean
    public CachingConnectionFactory confirmedConnectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        
        // Включение publisher confirms
        factory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        factory.setPublisherReturns(true);
        
        return factory;
    }
    
    @Bean
    public RabbitTemplate confirmedRabbitTemplate(
            CachingConnectionFactory connectionFactory) {
        
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        
        // Callback для подтверждений
        template.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                metrics.increment("publisher.confirms.success");
            } else {
                metrics.increment("publisher.confirms.failure");
                log.error("Message not confirmed: {}", cause);
                
                // Логика повторной отправки
                if (correlationData != null) {
                    retryService.scheduleRetry(correlationData.getId());
                }
            }
        });
        
        // Callback для возвращенных сообщений
        template.setReturnsCallback(returned -> {
            log.error("Message returned: {}", returned.toString());
            returnedMessageService.handleReturned(returned);
        });
        
        return template;
    }
}

#Java #middle #RabbitMQ

Consumer Acknowledgements

Типы подтверждений:

public class AcknowledgementExamples {
    
    // 1. Автоматическое подтверждение (ненадежно)
    @RabbitListener(queues = "auto.ack.queue")
    public void autoAck(Message message) {
        // Подтверждение происходит автоматически при возврате метода
    }
    
    // 2. Ручное подтверждение (рекомендуется)
    @RabbitListener(queues = "manual.ack.queue")
    public void manualAck(Order order, Channel channel,
                         @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        
        try {
            process(order);
            channel.basicAck(tag, false); // Положительное подтверждение
        } catch (BusinessException e) {
            // Не requeue для бизнес-ошибок
            channel.basicNack(tag, false, false);
        } catch (TechnicalException e) {
            // Requeue для технических ошибок
            channel.basicNack(tag, false, true);
        }
    }
    
    // 3. Подтверждение с транзакцией
    @Transactional
    @RabbitListener(queues = "transactional.queue")
    public void transactionalAck(Payment payment, Channel channel,
                                @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        
        // Сохранение в БД
        paymentRepository.save(payment);
        
        // Подтверждение произойдет только после коммита транзакции
        // При откате транзакции - сообщение вернется в очередь
    }
}

Transactional Channels

Использование транзакций:

@Service
public class TransactionalMessageService {
    
    private final RabbitTemplate rabbitTemplate;
    
    @Transactional
    public void processInTransaction(Order order) {
        // 1. Сохранение в базу данных
        orderRepository.save(order);
        
        // 2. Отправка сообщения в той же транзакции
        rabbitTemplate.execute(channel -> {
            // Начало транзакции RabbitMQ
            channel.txSelect();
            
            try {
                // Отправка сообщения
                channel.basicPublish(
                    "orders.exchange",
                    "order.created",
                    null,
                    serialize(order)
                );
                
                // Коммит транзакции RabbitMQ
                channel.txCommit();
                
                return null;
            } catch (Exception e) {
                // Откат транзакции RabbitMQ
                channel.txRollback();
                throw new RuntimeException("Transaction failed", e);
            }
        });
        
        // 3. Если произойдет исключение здесь, 
        // откатятся и БД и сообщение RabbitMQ
        inventoryService.updateStock(order);
    }
}

#Java #middle #RabbitMQ

Dead Letter Exchanges: обработка проблемных сообщений

Dead Letter Exchange (DLX) — специальный exchange, куда перенаправляются сообщения, которые не могут быть обработаны.

Типичные сценарии:
Сообщение отбраковано (nack без requeue)
Превышено максимальное количество попыток обработки
Истек TTL сообщения
Очередь заполнена (при overflow поведении)

Реализация DLX

Конфигурация основной очереди с DLX:

@Configuration
public class DeadLetterConfig {
    
    // Основной DLX
    @Bean
    public DirectExchange dlxExchange() {
        return new DirectExchange("dlx.exchange", true, false);
    }
    
    // Очередь для мертвых писем
    @Bean
    public Queue dlQueue() {
        return QueueBuilder.durable("dead.letter.queue")
            .withArgument("x-max-length", 10000) // Ограничение размера
            .withArgument("x-message-ttl", 86400000) // 24 часа хранения
            .build();
    }
    
    @Bean
    public Binding dlBinding() {
        return BindingBuilder.bind(dlQueue())
            .to(dlxExchange())
            .with("#"); // Все routing keys
    }
    
    // Рабочая очередь с настройкой DLX
    @Bean
    public Queue orderProcessingQueue() {
        return QueueBuilder.durable("orders.processing.queue")
            .withArgument("x-dead-letter-exchange", "dlx.exchange")
            .withArgument("x-dead-letter-routing-key", "orders.failed")
            .withArgument("x-max-retries", 3) // Кастомный аргумент
            .build();
    }
}

Обработчик проблемных сообщений:

@Component
public class DeadLetterProcessor {
    
    @RabbitListener(queues = "dead.letter.queue")
    public void handleDeadLetter(Message failedMessage,
                                Channel channel,
                                @Header(AmqpHeaders.DELIVERY_TAG) long tag,
                                @Header(AmqpHeaders.RECEIVED_ROUTING_KEY) String routingKey,
                                @Header(AmqpHeaders.RECEIVED_EXCHANGE) String exchange,
                                @Header("x-death") List<Map<String, Object>> deaths) {
        
        // Анализ причины попадания в DLQ
        String reason = analyzeFailureReason(deaths);
        
        log.error("Dead letter received: routingKey={}, reason={}, deaths={}", 
                 routingKey, reason, deaths);
        
        // Логика обработки в зависимости от причины
        if (isRecoverable(failedMessage, deaths)) {
            handleRecoverableMessage(failedMessage);
        } else {
            handlePermanentFailure(failedMessage);
        }
        
        // Подтверждение обработки DLQ
        channel.basicAck(tag, false);
    }
    
    private String analyzeFailureReason(List<Map<String, Object>> deaths) {
        if (deaths != null && !deaths.isEmpty()) {
            Map<String, Object> lastDeath = deaths.get(0);
            return (String) lastDeath.get("reason");
        }
        return "unknown";
    }
    
    private void handleRecoverableMessage(Message message) {
        // Пример: повторная отправка после задержки
        String originalQueue = extractOriginalQueue(message);
        long delay = calculateRetryDelay(message);
        
        retryService.scheduleRetry(message, originalQueue, delay);
    }
    
    private void handlePermanentFailure(Message message) {
        // Архивирование неудачных сообщений
        archiveService.archiveFailedMessage(message);
        
        // Уведомление команды поддержки
        alertService.notifySupportTeam(message);
    }
}

#Java #middle #RabbitMQ

Продвинутая обработка с задержкой повторных попыток:

@Configuration
public class DelayedRetryConfig {
    
    // Exchange для отложенных повторных попыток
    @Bean
    public CustomExchange delayedExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        
        return new CustomExchange(
            "delayed.retry.exchange",
            "x-delayed-message",
            true,
            false,
            args
        );
    }
    
    // Очередь для отложенных повторных попыток
    @Bean
    public Queue delayedRetryQueue() {
        return new Queue("delayed.retry.queue", true);
    }
    
    @Bean
    public Binding delayedBinding() {
        return BindingBuilder.bind(delayedRetryQueue())
            .to(delayedExchange())
            .with("retry.key")
            .noargs();
    }
    
    // Сервис для отложенных повторных попыток
    @Service
    public class DelayedRetryService {
        private final RabbitTemplate rabbitTemplate;
        
        public void scheduleRetry(Message message, int attempt) {
            long delay = calculateExponentialBackoff(attempt);
            
            rabbitTemplate.convertAndSend(
                "delayed.retry.exchange",
                "retry.key",
                message,
                m -> {
                    // Установка задержки
                    m.getMessageProperties()
                        .setHeader("x-delay", delay);
                    
                    // Сохранение номера попытки
                    m.getMessageProperties()
                        .setHeader("retry-attempt", attempt);
                    
                    return m;
                }
            );
        }
        
        private long calculateExponentialBackoff(int attempt) {
            return (long) Math.pow(2, attempt) * 1000; // Экспоненциальная задержка
        }
    }
}

Практические рекомендации

Выбор паттерна

Work Queues — для фоновой обработки задач с балансировкой нагрузки
Publish/Subscribe — для широковещательных уведомлений
Routing/Topics — для сложной маршрутизации сообщений
RPC — для синхронных запросов в асинхронной среде

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

At-most-once — только для non-critical данных
At-least-once — стандартный выбор для большинства систем
Exactly-once — достигается через идемпотентность и транзакции

#Java #middle #RabbitMQ

Обработка ошибок и мониторинг RabbitMQ

В современных системах мониторинг — это не просто сбор метрик, а целостная система наблюдаемости, состоящая из трёх столпов: метрики, логи и трассировка. RabbitMQ, как критический компонент инфраструктуры, требует особого внимания ко всем трём аспектам.


Мониторинг: многоуровневый подход

RabbitMQ Management UI: первый рубеж защиты

Management UI — это визуальный интерфейс, но его настоящая ценность в оперативном обнаружении аномалий.

Критические метрики в реальном времени:
Message rates — скорость публикации и потребления сообщений
Queue depths — глубина очередей (backlog)
Consumer count — количество активных потребителей
Connection churn — частота переподключений

Паттерны для наблюдения:
Внезапный рост глубины очереди указывает на отставание потребителей
Падение количества потребителей сигнализирует о проблемах с deployment
Увеличение частоты переподключений говорит о сетевых проблемах

Prometheus метрики: систематический мониторинг

Prometheus предоставляет количественные данные для анализа трендов и настройки алертов. RabbitMQ экспортирует метрики через встроенный плагин, но важнее понимать, что отслеживать:

Базовые метрики для любого кластера:

rabbitmq_queue_messages_ready        # Сообщения, готовые к доставке
rabbitmq_queue_messages_unacked      # Сообщения в процессе обработки
rabbitmq_queue_consumers             # Количество потребителей
rabbitmq_process_open_fds            # Открытые файловые дескрипторы
rabbitmq_erlang_gc_reclaimed_bytes   # Память, освобождённая сборщиком мусора

Производственные алерты должны отслеживать:
Глубину очереди, превышающую разумные лимиты
Отсутствие потребителей для критических очередей
Необычные паттерны в скорости обработки сообщений
Потребление памяти, приближающееся к лимитам


Health Checks в Spring Boot Actuator: проверка жизнеспособности

Spring Boot Actuator предоставляет готовые health checks, но в production их нужно расширять:

Три уровня проверок:
Liveness — проверка, что приложение запущено
Readiness — проверка готовности принимать трафик
Startup — мониторинг процесса запуска

Кастомные health checks для RabbitMQ:

// Концептуальный пример расширенной проверки
@Component
public class RabbitMQBusinessHealthIndicator {
    
    public Health checkBusinessReadiness() {
        // Проверяем не только соединение, но и:
        // 1. Наличие критических очередей
        // 2. Наличие активных потребителей
        // 3. Разумную глубину очередей
        // 4. Скорость обработки сообщений
    }
}

Важный принцип: Health checks должны проверять не только доступность RabbitMQ, но и способность вашего приложения эффективно с ним взаимодействовать.

Стратегии обработки ошибок

Классификация ошибок: знай своего врага

Транзиентные ошибки (временные):
Сетевые проблемы
Временная недоступность сервиса
Кратковременные таймауты

Бизнес-ошибки:
Невалидные данные в сообщении
Нарушение бизнес-правил
Конфликты данных

Системные ошибки:
Потеря соединения с базой данных
Недостаток ресурсов
Баги в коде


Retry с экспоненциальной задержкой

Повторные попытки — стандартный подход для транзиентных ошибок, но важно избегать retry storms:

Принципы правильного retry:
Экспоненциальная задержка между попытками
Ограничение максимального количества попыток
Исключение бизнес-ошибок из retry логики
Использование jitter для распределения нагрузки

# Конфигурация Spring Retry
spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true
          max-attempts: 3
          initial-interval: 1s
          multiplier: 2
          max-interval: 10s

Circuit Breaker: защита от каскадных отказов

Автоматический выключатель предотвращает зацикливание вызовов при постоянных ошибках:

Три состояния Circuit Breaker:
Closed — запросы проходят нормально
Open — запросы сразу отклоняются
Half-Open — пробные запросы для проверки восстановления

Использование с RabbitMQ: Circuit breaker следует применять для операций, которые могут вызывать каскадные отказы, например, при вызове внешних сервисов из обработчиков сообщений.

#Java #middle #RabbitMQ

Dead Letter Queues: изоляция проблемных сообщений

DLQ — это не мусорка, а система диагностики:

Что отправлять в DLQ:
Сообщения, превысившие лимит попыток обработки
Сообщения с истёкшим TTL
Сообщения, отброшенные из-за переполнения очереди

Архитектура обработки DLQ:
Анализ — изучение причин попадания в DLQ
Классификация — разделение на исправимые и неисправимые ошибки
Восстановление — повторная обработка после исправления
Архивация — сохранение неисправимых сообщений для анализа


Логирование и трассировка

Correlation IDs: сквозная идентификация запросов

Correlation ID — это уникальный идентификатор, который проходит через все компоненты системы, участвующие в обработке запроса.

Реализация в Spring Boot:

// Фильтр для HTTP запросов
@Component
public class CorrelationIdFilter implements Filter {
    
    public void doFilter(ServletRequest request, ServletResponse response, 
                         FilterChain chain) {
        // Извлекаем или генерируем correlation ID
        // Помещаем в MDC для логирования
        // Добавляем в заголовки RabbitMQ сообщений
    }
}

// MessagePostProcessor для RabbitMQ
@Component  
public class CorrelationIdMessagePostProcessor implements MessagePostProcessor {
    
    public Message postProcessMessage(Message message) {
        // Добавляем correlation ID из MDC в заголовки сообщения
        return message;
    }
}

Важно: Correlation ID должен передаваться через все асинхронные границы — HTTP запросы, сообщения RabbitMQ, вызовы внешних API.

Структурированное логирование: от текста к данным

Структурированные логи (JSON, Logstash) позволяют автоматически анализировать и агрегировать данные.

Ключевые поля для каждого log entry:
timestamp
level
logger
message
correlationId
traceId/spanId (для трассировки)
Дополнительный контекст (queue, messageId, userId)

Конфигурация Logback для JSON:

<appender name="JSON" class="ch.qos.logback.core.ConsoleAppender">
    <encoder class="net.logstash.logback.encoder.LogstashEncoder">
        <customFields>{"application":"${APP_NAME}"}</customFields>
    </encoder>
</appender>

Паттерны логирования для RabbitMQ:
При публикации сообщения: логируем messageId, routingKey, размер сообщения
При получении сообщения: логируем deliveryTag, очередь, consumer
При обработке: логируем длительность, результат, ошибки
При подтверждении: логируем успешность, время обработки


Распределённая трассировка с OpenTelemetry

Трассировка показывает путь запроса через все микросервисы, включая асинхронные взаимодействия через RabbitMQ.

Интеграция RabbitMQ с OpenTelemetry:
Инъекция trace context в заголовки сообщений
Создание spans для операций публикации и потребления
Связывание spans через очередь сообщений

Концептуальный подход:

HTTP Request → [Span A] → RabbitMQ Publish → [Span B]
                                  ↓
                           (сообщение в очереди)
                                  ↓
                          RabbitMQ Consume → [Span C] → DB Call → [Span D]

Важно: Даже при асинхронной коммуникации через RabbitMQ можно сохранить контекст трассировки, передавая traceId и spanId в заголовках сообщений.

Практические рекомендации

Уровни мониторинга

Инфраструктурный уровень: доступность RabbitMQ, использование ресурсов
Уровень приложения: health checks, метрики Spring Boot
Бизнес-уровень: скорость обработки заказов, количество ошибок

Стратегия алертинга

Приоритеты алертов:
P0: Полная недоступность RabbitMQ или критичных очередей
P1: Быстрый рост глубины очереди (> 1000 сообщений/минуту)
P2: Отсутствие потребителей для критичных очередей
P3: Ухудшение производительности (> 95 перцентиль latency)

Паттерны для production

Всегда используйте Publisher Confirms для гарантированной доставки
Реализуйте идемпотентность на стороне потребителя
Настройте разумные TTL для сообщений
Мониторьте не только RabbitMQ, но и своё приложение
Тестируйте сценарии отказа в staging среде


#Java #middle #RabbitMQ