Что делать, если одно сообщение вызывает падение демона и блокирует обработку последующих сообщений?

Проблема “плохого” сообщения возникает, когда входные данные или логика обработки приводят к исключению, и сообщение повторно доставляется снова и снова.

Почему очередь может блокироваться

Если используется prefetch=1 или малое число воркеров:

1. сообщение доставляется

2. обработка падает

3. сообщение возвращается в очередь

4. тот же воркер снова получает его

В итоге прогресс не происходит.

Практические меры

1) Ограничение количества попыток

Обычно применяют счетчик попыток:

• сохраняют число ретраев в заголовке сообщения

• после N попыток отправляют в DLQ

2) Использование Dead Letter Queue

Сообщения, которые не удалось обработать:

• не блокируют основную очередь

• могут анализироваться отдельно

3) Изоляция обработки

Подходы:

• обработка в отдельном процессе

• worker pool, где падение одного процесса не влияет на другие

4) Обработка ошибок внутри воркера

Обработчик должен перехватывать исключения:

def on_message(ch, method, properties, body):
    try:
        process(body)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception:
        # отправка в DLQ или retry
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

Дополнительные техники

1. Валидация входных данных до тяжелой обработки

2. Таймауты на внешние вызовы

3. Идемпотентная обработка

Вывод
Если одно сообщение падает постоянно, его нужно изолировать: ограничить ретраи, отправить в DLQ и не позволять ему блокировать очередь.