Ошибки потребителя: retry и dead letter

У обработки из предыдущих уроков осталось две ветки, и обе плохие для ошибок. Requeue=true возвращает сообщение в голову очереди — воркер тут же получает его снова, снова падает и снова возвращает: цикл, который горит на полной скорости, пока его не заметит человек. Requeue=false просто снимает сообщение с учёта — данные ушли, заказ потерялся, и никто об этом не узнает, потому что ошибок в логах нет, а очередь пустая. Третий путь называется dead lettering: сообщение, от которого потребитель отказался, брокер сам перепубликует в другой обменник, где его подхватывает отдельная очередь. Это штатный механизм AMQP, а не причуда брокера: отказ от доставки — тоже событие, и у него может быть подписчик. Так в архитектуре появляется парковка для проблемных сообщений: трафик идёт дальше, поломка изолирована, а дежурный разбирает карантин в спокойном режиме. В этом уроке — как включить механизм и как построить на нём повтор с лимитом попыток.

Rabbitmq dead letter: куда уходят отказы#

Rabbitmq dead letter — это не отдельная очередь, а правило перепубликации: когда сообщение покидает очередь не через доставку потребителю, брокер отправляет его в exchange, указанный в настройках этой очереди. Поводы три. Первый — отказ: Nack или Reject с requeue=false, то есть именно та ветка, которую потребитель выбирает, когда повтор бессмыслен. Второй — истечение TTL: у очереди или сообщения есть срок жизни, и просроченное сообщение тоже считается покинувшим очередь штатно. Третий — переполнение: очередь с лимитом длины x-max-length выталкивает старейшее или новейшее сообщение при превышении. Перепубликация — это обычная публикация внутри брокера: тело сохраняется, свойства в основном сохраняются, но добавляется заголовок x-death со следами маршрута, а ключ маршрутизации заменяется на x-dead-letter-routing-key, если он задан, или остаётся прежним. Именно поэтому под dead letter логично брать direct-обменник из одиннадцатого урока: он раскладывает отказы по очередям по ключу так же, как боевой трафик.

Каркас: DLX, DLQ и очередь с настройкой

package main

import (
        "context"
        "log"

        amqp "github.com/rabbitmq/amqp091-go"
)

func main() {
        conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
        if err != nil {
                log.Fatal(err)
        }
        defer conn.Close()

        ch, err := conn.Channel()
        if err != nil {
                log.Fatal(err)
        }

        // обменник отказов и очередь-карантин
        err = ch.ExchangeDeclare("orders.dlx", "direct", true, false, false, false, nil)
        if err != nil {
                log.Fatal(err)
        }
        _, err = ch.QueueDeclare("orders.dead", true, false, false, false, nil)
        if err != nil {
                log.Fatal(err)
        }
        err = ch.QueueBind("orders.dead", "dead", "orders.dlx", false, nil)
        if err != nil {
                log.Fatal(err)
        }

        // боевая очередь со ссылкой на обменник отказов
        _, err = ch.QueueDeclare("orders", true, false, false, false,
                amqp.Table{
                        "x-dead-letter-exchange":    "orders.dlx",
                        "x-dead-letter-routing-key": "dead",
                })
        if err != nil {
                log.Fatal(err)
        }

        msgs, err := ch.ConsumeWithContext(context.Background(), "orders", "worker",
                false, false, false, false, nil)
        if err != nil {
                log.Fatal(err)
        }

        for d := range msgs {
                if err := handle(d.Body); err != nil {
                        log.Println("отказ:", err)
                        if err := d.Nack(false, false); err != nil {
                                log.Println("nack failed:", err)
                        }
                        continue
                }
                d.Ack(false)
        }
}

func handle(body []byte) error {
        // здесь валидация и бизнес-логика из пятнадцатого урока
        return nil
}

Порядок объявления идёт от хвоста: сначала обменник отказов и карантинная очередь с привязкой, затем боевая очередь, в аргументах которой указан x-dead-letter-exchange. Обменник выбран direct, ключ маршрутизации задан явно через x-dead-letter-routing-key — иначе брокер взял бы исходный ключ публикации. Nack с requeue=false в цикле воркера и есть триггер: брокер перепубликует доставку в orders.dlx, и сообщение оседает в orders.dead.

Пути в dead letter#

Повод	Что происходит	Кто инициирует
Nack / Reject с requeue=false	отказ от доставки: брокер перепубликует сообщение в DLX	код потребителя
истечение TTL	сообщение с x-message-ttl или из очереди с x-message-ttl просрочено	настройка очереди или издатель
x-max-length	очередь с лимитом длины выталкивает сообщения при переполнении	настройка очереди
reject-publish при переполнении	режим очереди: новые сообщения отбраковываются, а не выталкивают старые	настройка очереди

Requeue=true и ядовитое сообщение

Кейс: в orders прилетело тело, которое json.Unmarshal из пятнадцатого урока не переваривает, а воркер отвечает requeue=true, потому что так было написано в примере. Сообщение возвращается в голову очереди и доставляется мгновенно — десятки тысяч циклов в секунду, ядра заняты, лог раздувается одной строкой, а стоящие за битым сообщением задачи не двигаются. Management UI покажет красный флаг redelivered и взрыв скорости доставки. Диагноз — по паре «глубина не падает, потребитель занят на сто процентов». Лечение: ошибку разбора считать окончательной — Nack(false, false) и путь в DLX; повторным возвратом имеет смысл отвечать только на временные сбои вроде недоступности базы, и для них лучше очередь повторов с задержкой, о которой ниже.

Dlx очередь go: аргументы QueueDeclare#

В Go-коде вся настройка dead letter живёт в пятом аргументе QueueDeclare — таблице amqp.Table, и разбирать dlx очередь go стоит по этим ключам. X-dead-letter-exchange — имя обменника, куда брокер перепубликует покидающие очередь сообщения; сам обменник — обычный, подойдёт direct или topic из тринадцатого урока. X-dead-letter-routing-key — ключ, который подставляется при перепубликации; без него уходит оригинальный ключ, что удобно, когда карантинная очередь привязана тем же ключом. Объявление идемпотентно, но сверяет аргументы: попытка переобъявить orders без таблицы или с другой таблицей закроет канал ошибкой 406 PRECONDITION_FAILED — тот же контракт, что в пятом уроке, поэтому настройки DLX зашивают в код топологии с самого начала: сменить аргументы у существующей очереди нельзя, придётся переносить данные в новую. Учитывая это, DLX и DLQ объявляют теми же durable-настройками, что и боевую очередь, — иначе после рестарта брокера ссылка окажется на несуществующий обменник, и перепубликация тихо отбросит сообщение.

Retry с лимитом: TTL-очередь повторов#

DLQ сама по себе отвечает на вопрос «куда класть сломанное», но временные сбои — недоступная база, тайм-аут внешнего API — хочется не парковать, а повторить позже. На DLX собирается паттерн очереди повторов: боевая очередь ссылается на обменник retry, очередь retry объявлена с аргументом x-message-ttl, скажем, тридцать секунд, и собственным x-dead-letter-exchange, ведущим обратно в боевую очередь. Воркер при временной ошибке делает Nack(false, false) — сообщение уходит в retry, ждёт TTL и возвращается в боевую очередь как dead letter. Потребитель отличит круг не по телу, а по заголовку x-death: брокер пишет туда историю покидания очередей, и счётчик count в записи растёт с каждым кругом. Лимит попыток — это условие в коде воркера: count выше порога — отказ окончательный, Nack в конечную DLQ с другим ключом. Задержку задаёт TTL всей очереди, поэтому у паттерна есть честная оговорка: сообщения внутри retry выходят в порядке очереди, и короткая задержка для одной задачи означает такую же для остальных кругов.

Воркер с лимитом попыток

func attempts(d amqp.Delivery) int {
        deaths, ok := d.Headers["x-death"].([]interface{})
        if !ok {
                return 0
        }
        total := 0
        for _, entry := range deaths {
                if e, ok := entry.(amqp.Table); ok {
                        if c, ok := e["count"].(int64); ok {
                                total += int(c)
                        }
                }
        }
        return total
}

// в цикле воркера
for d := range msgs {
        err := process(d.Body)
        if err == nil {
                d.Ack(false)
                continue
        }
        if isTemporary(err) && attempts(d) < 3 {
                log.Println("повтор:", err)
                d.Nack(false, false) // в очередь повторов по её DLX
                continue
        }
        log.Println("в карантин:", err)
        d.Nack(false, false) // TTL-очередь сама вернёт или DLQ примет
}

Заголовок x-death приходит как срез интерфейсов, каждый элемент — amqp.Table с полями count, queue, exchange, reason, time; поля брокер кладёт в int64, поэтому приведение типов двойное. Порог на три попытки — константа политики, её выводят в конфиг вместе с TTL очереди повторов: суммарное время ожидания равняется попыткам, умноженным на TTL.

Что лежит в x-death

Каждая запись x-death описывает одно покидание очереди: count — сколько раз сообщение покидало именно эту очередь, exchange и routing-keys — откуда оно пришло, queue — имя очереди, reason — повод (rejected, expired, maxlen), time — момент последнего события. При повторных кругах через ту же очередь брокер не плодит записи, а увеличивает count и обновляет time. Поле count имеет тип int64 — приведение к int напрямую падает в рантайме, это самая частая ошибка разбора заголовка.

Ядовитые сообщения: карантин и разбор#

Конечная DLQ — это не мусорка, а рабочий инструмент дежурного, и её содержимое требует того же внимания, что глубина боевой очереди. Практика зрелых систем: отдельная DLQ на каждую боевую очередь с именем вида orders.dead, алерт на рост глубины из двадцать второго урока и правило «карантин пустой по умолчанию». Ядовитое сообщение разбирают руками: в management UI из второго урока его можно открыть и увидеть тело, заголовки и x-death — чаще всего виноваты битое тело из пятнадцатого урока или невалидные бизнес-данные, которые валидация пропустила. Для возврата исправленных сообщений пишут маленький replay-инструмент: он читает карантин и перепубликует тела в боевую очередь, не удаляя их автоматически. Автоматический replay опасен: сообщение, ломавшее потребителя, вернётся и сломает его снова, создав цикл между конвейером и карантином. Разбор начинается с reason в x-death и счётчика попыток — они отвечают, где сообщение застряло и сколько раз о нём спотыкались.

Код урока — локально, интерактивы — на странице

Dead lettering исполняет брокер, поэтому каркас DLX проверяется локально против контейнера из второго урока: опубликуйте битый заказ и посмотрите в management UI, как он осел в orders.dead с заголовком x-death. В браузерной песочнице сети и брокера нет — там доступны fmt, strings, strconv, math, sort, errors и time, так что разбор заголовков и счётчика попыток можно потренировать программой, подложив x-death руками. Интерактивы ниже закрепят аргументы QueueDeclare и ветки воркера.

Правила обработки ошибок потребителя#

Сначала классифицируйте ошибку: временная — повтор, окончательная — карантин; requeue=true по умолчанию — ловушка.
Битое тело и невалидные данные — окончательный отказ: Nack с requeue=false в очередь с DLX.
X-dead-letter-exchange и x-dead-letter-routing-key задают при объявлении очереди; смена аргументов живой очереди невозможна.
Очередь повторов — это TTL и собственный DLX обратно в боевую очередь; лимит попыток считает count в x-death.
Count в x-death имеет тип int64: приведение типов пишите в два шага.
DLQ мониторят как боевую очередь: рост карантин — повод для алерта, а не для молчаливой чистки.

Опора этого урока: восьмой урок про ack и nack дал механику requeue, пятнадцатый про JSON — валидацию, которая отделяет битые тела, а одиннадцатый про direct exchange — обменники, на которых строятся DLX и очередь повторов. Свойства очередей и идемпотентность объявления остались в пятом уроке, а двадцать третий урок про антипаттерны держит requeue-цикл в коллекции граблей — отсюда он вынесен архитектурно. Карта курса — на странице курса.

Проверка: куда уйдёт отказ

Очередь orders объявлена с x-dead-letter-exchange=orders.dlx и x-dead-letter-routing-key=dead. Воркер вызвал d.Nack(false, false) на доставке из orders. Что произойдёт?

Проверка: очередь повторов

Очередь retry объявлена с x-message-ttl=30000 и x-dead-letter-exchange, ведущим обратно в боевую очередь. Воркер сделал Nack(false, false), и сообщение ушло в retry. Что случится через тридцать секунд?

Что ответит брокер на переобъявление?

Прочитайте код и выберите, что он выведет:

// топология объявлена и работает:
_, err := ch.QueueDeclare("orders", true, false, false, false,
        amqp.Table{
                "x-dead-letter-exchange": "orders.dlx",
        })
log.Println("первый:", err == nil)

// другой запуск: та же очередь, но без аргументов
_, err = ch.QueueDeclare("orders", true, false, false, false, nil)
fmt.Println("второй:", err)

Соберите каркас dead letter

Допишите объявление обменника, ссылку на него в аргументах очереди и отказ воркера.

err = ch.ExchangeDeclare("orders.dlx", , true, false, false, false, nil) if err != nil { log.Fatal(err) } _, err = ch.QueueDeclare("orders", true, false, false, false, amqp.Table{ "x-dead-letter-exchange": , }) // ошибка обработки в цикле воркера: d.(false, false)

Соберите топологию с DLX

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

Строки

Ваша программа

Нажимайте на строки слева, чтобы собрать программу

Карточки: dead letter

Термин

dead lettering

нажмите, чтобы перевернуть

Определение

Перепубликация сообщения, покинувшего очередь не через доставку: отказ потребителя, TTL или переполнение. Сообщение уходит в DLX очереди.

1 / 6

Песочница урока

Редактируйте код и запускайте его прямо здесь — интерпретатор работает в браузере, ничего устанавливать не нужно. За каждый запуск — +2 XP.

Вывод

Запустите код, чтобы увидеть вывод.

Коротко: что запомнить

Dead letter — правило перепубликации: отказ потребителя, TTL и переполнение уводят сообщение в DLX очереди.
X-dead-letter-exchange задаётся при объявлении очереди; смена аргументов живой очереди невозможна — канал закроется ошибкой 406.
Requeue=true на постоянной ошибке — цикл на полной скорости: битые тела уходят с Nack(false, false).
Очередь повторов — это TTL и собственный DLX обратно; лимит попыток воркер считает по x-death.
Count в x-death имеет тип int64: приведение к int напрямую паникует.
DLQ — рабочий инструмент дежурного: мониторинг глубины, разбор тел в UI, ручной replay без автоматики.
Классификация ошибок важнее настройки брокера: временная — пауза в retry, окончательная — карантин.

Домашнее задание

Соберите каркас из урока: обменник отказов, карантинную очередь orders.dead и боевую orders с аргументами DLX. Опубликуйте пять валидных заказов и одно битое тело из пятнадцатого урока, убедитесь, что битое оказалось в orders.dead с заголовком x-death, а валидные обработались. Затем постройте очередь повторов retry с TTL в десять секунд и DLX обратно в orders и посчитайте в воркере попытки по count: после третьего круга отправьте сообщение в конечный карантин. В конце попробуйте переобъявить orders без аргументов и объясните в комментариях ошибку 406.

Вопросы и ответы по уроку

Можно ли добавить DLX к уже созданной очереди?

Нет: аргументы очереди фиксируются при объявлении, и попытка переобъявить с новыми аргументами закрывает канал ошибкой PRECONDITION_FAILED. Путь миграции — новая очередь с нужными аргументами, перелив сообщений и переключение потребителей; либо выгрузить сообщения через shovel-плагин. Поэтому DLX закладывают в топологию с первого дня, даже если карантин пока кажется лишним.

Чем отличается Reject от Nack?

Reject подтверждает отказ одной доставки, Nack — умеет requeue сразу несколько: первый аргумент multiple. Для одиночных доставок в цикле воркера разницы в поведении dead lettering нет — оба с requeue=false отправляют сообщение в DLX. Nack берут ради multiple при батч-обработке, Reject — когда семантика нужна в одну доставку.

Какой TTL ставить очереди повторов?

Из характера сбоев: недоступная база поднимается за десятки секунд, тайм-аут внешнего API может жить минуты. Стартовая точка — тридцать секунд при трёх попытках, суммарно полторы минуты; дальше цифра тюнится по метрикам и инцидентам. Помните про честную оговорку паттерна: TTL — свойство всей очереди, и одна просрочка сдвигает весь круг.

Термины урока

Ключевые понятия из этого урока — короткие определения в глоссарии. Наведите на термин, чтобы вспомнить, откройте карточку для подробностей и ссылки на практику.