Модуль 13 ⏱ ~30 мин чтения 📐 3 Mermaid диаграмм

4. Acknowledgements и delivery guarantees

В предыдущей статье 3. Queues и bindings мы подробно изучили, как очереди RabbitMQ сохраняют данные на диске и в оперативной памяти. Но надежное хранение — это лишь половина пути. Фундаментальная задача любого брокера — передать полезную нагрузку потребителю (Consumer) и получить математически строгое подтверждение того, что бизнес-логика успешно отработала.

Распределенные системы несовершенны по своей природе. Сетевой кабель может моргнуть, процесс воркера на Go может внезапно завершиться panic из-за разыменования nil-указателя, а контейнер в Kubernetes может быть мгновенно уничтожен ядром Linux по сигналу SIGKILL (OOM Killer) прямо посреди финансовой транзакции.

Как брокер узнает, что сообщение действительно обработано и его можно стереть из буфера? Для решения этой дилеммы в протоколе AMQP 0-9-1 существует механизм Acknowledgements (Подтверждений).

Гарантии доставки (Delivery Guarantees)

Прежде чем погружаться в протокольные тонкости команд Ack и Nack, зафиксируем базовую классификацию гарантий доставки в распределенных системах:

flowchart TD
    subgraph Guarantees ["Модели гарантий доставки"]
        AMO["1. At-most-once<br>(Не более одного раза)"]
        ALO["2. At-least-once<br>(Не менее одного раза)"]
        EO["3. Exactly-once<br>(Строго один раз)"]
    end

    AMO -->|"Fire-and-Forget"| R1["0 или 1 доставка.<br>Дубли исключены, возможна потеря данных."]
    ALO -->|"Manual Ack + Retries"| R2["1 или более доставок.<br>Потеря исключена, возможны дубликаты."]
    EO -->|"At-least-once + Idempotency"| R3["Строго один логический эффект.<br>Достигается только связкой брокера и консьюмера."]

    classDef gBox fill:#1e293b,stroke:#38bdf8,stroke-width:2px,color:#f8fafc;
    classDef rBox fill:#312e81,stroke:#818cf8,stroke-width:1px,color:#f8fafc;

    class AMO,ALO,EO gBox;
    class R1,R2,R3 rBox;
  1. At-most-once (Не более одного раза): Сообщение доставляется 0 или 1 раз. При сбоях системы данные могут быть безвозвратно утеряны, но дублирование полностью исключено (паттерн Fire-and-Forget). Применяется для низкоприоритетной телеметрии и метрик, где потеря нескольких замеров не критична.
  2. At-least-once (Не менее одного раза): Сообщение гарантированно доставляется 1 или более раз. Потеря данных исключена, однако при сбоях сети или падении воркеров возможны повторные доставки (дубликаты). Это фундаментальный золотой стандарт RabbitMQ.
  3. Exactly-once (Строго один раз): Идеальное состояние распределенной системы. На уровне одного лишь брокера сообщений (будь то RabbitMQ, Kafka или NATS) эта гарантия физически недостижима из-за классической «задачи двух генералов» (Two Generals' Problem): если воркер успешно выполнил транзакцию в базе данных, но подтверждающий сетевой пакет Ack потерялся из-за сетевого сбоя, брокер сочтет воркера мертвым и переотправит сообщение другому узлу.
    • Единственный промышленный способ получить Exactly-once обработку: обеспечить доставку At-least-once на уровне RabbitMQ и реализовать строгую идемпотентность на стороне консьюмера (этому мы посвятим отдельную лекцию 10. Idempotency в message processing).

Consumer Acknowledgements (Подтверждения подписчика)

Когда ваше приложение на Go подключается к очереди через вызов ch.Consume(...), RabbitMQ может работать в одном из двух режимов подтверждения.

1. Auto-Ack (Fire and Forget)

Если при вызове метода подписки передан аргумент autoAck: true, брокер помечает сообщение как успешно выполненное и удаляет его из очереди ровно в тот момент, когда запишет байты фрейма в свой исходящий TCP-сокет.

2. Manual Ack (Ручное подтверждение)

В промышленных системах стандартным является использование autoAck: false. В этом режиме удаление сообщения из брокера происходит только по явной команде приложения.

sequenceDiagram
    autonumber
    participant Broker as "RabbitMQ (Erlang Queue Process)"
    participant Worker as "Go Consumer (Goroutine)"
    participant Storage as "Database (PostgreSQL)"

    Note over Broker: Очередь хранит сообщение (Ready)
    Broker->>Worker: Basic.Deliver (Payload, deliveryTag: 42)
    Note over Broker: Перенос сообщения в статус Unacked

    Worker->>Storage: Выполнение транзакции (BEGIN ... COMMIT)
    Storage-->>Worker: OK (Данные сохранены в БД)

    Worker->>Broker: Basic.Ack (deliveryTag: 42, multiple: false)
    Note over Broker: Окончательное удаление сообщения из RAM/диска

Механика статуса Unacked под капотом

Когда брокер передает сообщение клиенту, он не стирает его с диска или из оперативной памяти. Он переводит сообщение из статуса Ready во внутреннее состояние Unacked (ожидает подтверждения) и ассоциирует его с дескриптором конкретного amqp.Channel.

Каждому сообщению, летящему внутри канала, брокер присваивает монотонно возрастающее 64-битное целое число — deliveryTag. Этот тег уникален строго в рамках данного канала и служит адресом сообщения при отправке подтверждения.

Негативные подтверждения (Nack и Reject)

Что делать, если при обработке сообщения в Go что-то пошло не так? Например, сервис базы данных временно недоступен, внешний платежный шлюз вернул HTTP 500 или пришел битый JSON-документ?

В этом случае отправлять Ack категорически нельзя — иначе брокер сотрет бракованное сообщение, и задача будет потеряна. Для информирования брокера об ошибке в протоколе AMQP предусмотрены две инструкции:

  1. Basic.Reject: Базовый метод AMQP, позволяющий отклонить одно конкретное сообщение по его deliveryTag.
  2. Basic.Nack (Negative Acknowledgement): Расширение протокола от RabbitMQ. В отличие от Reject, поддерживает флаг multiple: true, позволяя отклонить сразу пачку сообщений вплоть до указанного тега.

Оба метода принимают критически важный булев аргумент — requeue.

1. Requeue = true (Опаснейшая ловушка: Poison Message Loop)

Если консьюмер выполняет вызов с флагом requeue: true, брокер возвращает отклоненное сообщение обратно в очередь.

graph TD
    subgraph DangerZone ["Poison Message Loop (100% CPU Churn)"]
        Q["Очередь RabbitMQ<br>(Head of Queue)"] -->|"1. Basic.Deliver"| W["Go Воркер"]
        W -->|"2. json.Unmarshal FAILED!<br>d.Nack(requeue=true)"| Q
    end

    classDef err fill:#450a0a,stroke:#f87171,stroke-width:2px,color:#fecaca;
    classDef q fill:#1e293b,stroke:#38bdf8,stroke-width:2px,color:#f8fafc;

    class W err;
    class Q q;

2. Requeue = false (Архитектурно правильный путь)

Если ошибка является фатальной (детерминированный баг валидации) либо счетчик повторных попыток исчерпан, воркер обязан передавать флаг отбрасывания — делать Nack(requeue=false).

В этом случае сообщение будет либо удалено навсегда, либо, что более правильно, сброшено в Dead Letter Exchange (DLX) для последующего анализа инженерами или отложенного ретрая (подробно разберем в лекции 7. Dead letter exchanges).

Publisher Confirms: Как отправителю спать спокойно?

На технических собеседованиях и при проектировании архитектуры инженеры часто сосредотачиваются на консьюмерах, забывая про сторону отправителя (Producer).

Для устранения этой уязвимости в AMQP 0-9-1 существует механизм Publisher Confirms (Подтверждения издателя):

  1. Канал отправителя переводится в режим подтверждений вызовом ch.Confirm(false).
  2. Библиотека начинает асинхронно слушать подтверждения от сервера.
  3. Брокер отправляет Basic.Ack публикатору только после того, как сообщение было маршрутизировано во все целевые очереди и физически записано на диск (в случае Quorum или Durable-очередей с DeliveryMode = Persistent).

Использование Publisher Confirms слегка увеличивает Round-Trip Time (RTT) публикации, но является единственным математически строгим способом гарантировать семантику At-least-once на стороне отправителя.

Идиоматичный Go-код консьюмера с Manual Ack

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

Go 
package main

import (
    "context"
    "encoding/json"
    "errors"
    "log"
    "time"

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

type OrderTask struct {
    OrderID   string `json:"order_id"`
    UserID    int64  `json:"user_id"`
    Amount    int64  `json:"amount"`
}

// processOrder выполняет бизнес-логику задачи
func processOrder(ctx context.Context, body []byte) error {
    var task OrderTask
    if err := json.Unmarshal(body, &task); err != nil {
        // Детерминированная ошибка: битый JSON ретраить бессмысленно
        return errors.New("malformed_json")
    }

    if task.Amount <= 0 {
        return errors.New("invalid_amount")
    }

    // Эмуляция полезной работы (запись в БД)
    log.Printf("Заказ %s успешно обработан для пользователя %d", task.OrderID, task.UserID)
    return nil
}

// startConsumer запускает цикл вычитывания сообщений с ручным Ack
func startConsumer(ctx context.Context, ch *amqp.Channel, queueName string) error {
    // 1. Обязательно регистрируем подписку с autoAck: false!
    msgs, err := ch.Consume(
        queueName,
        "orders-worker-1", // consumer tag
        false,             // autoAck (СТРОГО FALSE)
        false,             // exclusive
        false,             // noLocal
        false,             // noWait
        nil,               // args
    )
    if err != nil {
        return err
    }

    // 2. Вычитываем входящий канал доставки в отдельной горутине
    go func() {
        for {
            select {
            case <-ctx.Done():
                log.Println("Остановка консьюмера по контексту...")
                return
            case d, ok := <-msgs:
                if !ok {
                    log.Println("AMQP канал доставки закрыт брокером")
                    return
                }

                // Обработка сообщения
                processCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
                err := processOrder(processCtx, d.Body)
                cancel()

                if err != nil {
                    log.Printf("Ошибка обработки сообщения [tag=%d]: %v", d.DeliveryTag, err)

                    // Критическая ошибка валидации: Nack БЕЗ requeue
                    // Сообщение улетит в DLX (если настроен) или удалится
                    if nackErr := d.Nack(false, false); nackErr != nil {
                        log.Printf("Сбой отправки Nack: %v", nackErr)
                    }
                    continue
                }

                // 3. Успешная обработка: отправляем явный Ack
                if ackErr := d.Ack(false); ackErr != nil {
                    log.Printf("Сбой отправки Ack [tag=%d]: %v", d.DeliveryTag, ackErr)
                    // Если отправка сорвалась, соединение мертво;
                    // при переподключении брокер переотправит сообщение снова.
                }
            }
        }
    }()

    return nil
}

Итог

  1. Гарантия At-least-once: Достигается исключительно при совместном использовании Publisher Confirms на стороне отправителя и Manual Ack (autoAck: false) на стороне получателя.
  2. Управление состоянием в Erlang: Брокер непрерывно отслеживает сообщения в статусе Unacked. При разрыве TCP-сокета воркера брокер автоматически возвращает их в состояние Ready в голову очереди.
  3. Опасность Poison Messages: Никогда не выполняйте Nack(requeue=true) вслепую при сбоях валидации. Это порождает бесконечный цикл загрузки CPU. Используйте requeue=false и направляйте брак в Dead Letter Exchange.
  4. Опасность утечек Ack: Каждый путь исполнения в коде консьюмера обязан завершаться вызовом Ack либо Nack. Забытый Ack приводит к переполнению оперативной памяти брокера и аварийной остановке всего кластера.

Мы научились гарантировать доставку каждого сообщения. Но представьте: в очереди накопился миллион задач, и консьюмер на Go подключился с флагом autoAck: false. Сколько сообщений брокер мгновенно вывалит в сокет нашего бедного воркера? Как предотвратить затопление памяти приложения и справедливо распределить работу между горутинами?

Об этом мы поговорим в следующей лекции: 5. Prefetch и QoS.