Модуль 9 ⏱ ~24 мин чтения 📐 2 Mermaid диаграмм

27. Очереди задач

В программировании параллельных систем горутины в оперативной памяти подобны стикерам с заметками, наклеенным на край монитора: они мгновенны, удобны и ничего не стоят. Но если в серверной внезапно отключится питание или процесс завершится по ошибке сегментации, все ваши стикеры с незавершенными задачами исчезнут без следа.

Очереди задач (Message Queues) представляют собой принципиально иной уровень надежности — это централизованное почтовое отделение с несгораемыми сейфами и расписками о получении. Они разделяют отправителя (producer) и получателя (consumer) не только в пространстве, но и во времени. Если ваш сервис заказов генерирует 10 000 транзакций в секунду в «черную пятницу», а сервис генерации бухгалтерских PDF-счетов способен переваривать только 200 документов в секунду, очередь задач принимает удар на себя, аккуратно сглаживая пиковую нагрузку и спасая инфраструктуру от каскадного коллапса.

Философия очередей задач

В отличие от фоновых горутин, которые живут исключительно в памяти процесса и исчезают вместе с ним, очереди задач выносят выполнение за пределы основного сервиса. Они обеспечивают персистентность, горизонтальное масштабирование потребителей, изоляцию сбоев и детерминированный контроль доставки. В production-системах очереди становятся нервной системой микросервисной архитектуры, связывая события, команды и асинхронные процессы без блокирующих вызовов.

1. Архитектура и гарантии доставки

Базовый паттерн включает три компонента: продюсер (генерирует задачу), брокер (хранит и маршрутизирует), консьюмер (обрабатывает). Выбор гарантии доставки определяет архитектуру обработки:

  • At-most-once («не более одного раза»): сообщение может потеряться при сетевом сбое, но гарантированно никогда не будет доставлено дважды. Быстро, дешево, но рискованно для критичного бизнеса.
  • At-least-once («хотя бы один раз»): сообщение доставляется минимум один раз. Стандарт де-факто для распределенных систем. Если обработчик упадет до отправки подтверждения, брокер доставит сообщение повторно. Это требует обязательной идемпотентности на стороне обработчика.
  • Exactly-once («строго однократно»): математический идеал, строгая однократная доставка. Крайне дорого в реализации, требует координации брокера и обработчика через распределенные транзакции или идемпотентные ключи с дедупликацией.
flowchart LR
    A["Продюсер генерирует задачу"] --> B["Брокер сохраняет в очередь"]
    B --> C["Консьюмер получает сообщение"]
    C --> D{"Обработка успешна?"}
    D -->|"Да"| E["Отправка ACK удаление из очереди"]
    D -->|"Нет"| F["Отправка NACK или Requeue"]
    F --> G["Повторная доставка или перемещение в DLQ"]

    classDef normal fill:#1e293b,stroke:#38bdf8,color:#f8fafc;
    classDef decision fill:#334155,stroke:#f59e0b,color:#f8fafc;
    classDef ack fill:#0f172a,stroke:#10b981,color:#f8fafc;
    classDef nack fill:#4c0519,stroke:#f43f5e,color:#ffffff;
    class A,B,C normal;
    class D decision;
    class E ack;
    class F,G nack;

2. Под капотом. Механика брокеров и протоколов

Брокеры сообщений радикально отличаются своим внутренним архитектурным устройством, что напрямую определяет выбор клиентских библиотек в Go:

  • RabbitMQ: Построен на Erlang VM. Использует классический протокол AMQP 0-9-1 с фреймовой передачей данных. Очереди хранятся преимущественно в оперативной памяти с поддержкой ленивой выгрузки на диск. Гарантии доставки обеспечиваются через подтверждения (acks) и устойчивые хэши. Концепция: «умный брокер, глупый клиент» — брокер сам отслеживает, кому и сколько сообщений выдано.
  • Kafka: Распределенный журнал транзакций (append-only log). Сообщения записываются строго последовательно на диск с использованием системного механизма mmap для zero-copy чтения через sendfile. Консьюмеры сами читают данные по смещению (offset), что позволяет перепотреблять или пропускать данные. Концепция: «глупый брокер, умный клиент».
  • NATS / JetStream: Сверхлегковесный брокер, целиком написанный на Go. Использует бинарный протокол, оптимизированный под экстремально низкую латентность. Модуль JetStream добавляет персистентность и распределенный стриминг поверх in-memory ядра.

3. Идиоматичный потребитель в Go

Правильный потребитель в Go — это управляемый цикл, который реагирует на отмену контекста, явно подтверждает обработку и не блокирует тред ОС ожиданием.

Go 
package consumer

import (
    "context"
    "fmt"
    "log"

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

func StartWorker(ctx context.Context, conn *amqp.Connection, queueName string, handler func(ctx context.Context, payload []byte) error) error {
    ch, err := conn.Channel()
    if err != nil {
        return fmt.Errorf("open channel: %w", err)
    }
    defer ch.Close()

    // autoAck: false — обязательное явное подтверждение
    msgs, err := ch.Consume(queueName, "", false, false, false, false, nil)
    if err != nil {
        return fmt.Errorf("register consumer: %w", err)
    }

    for {
        select {
        case <-ctx.Done():
            log.Println("consumer shutting down via context cancellation")
            return ctx.Err()
        case d, ok := <-msgs:
            if !ok {
                return fmt.Errorf("consumer channel closed unexpectedly")
            }

            // Обработка в отдельной горутине не рекомендуется без пула
            // Здесь синхронная обработка для простоты контроля ACK/NACK
            if err := handler(ctx, d.Body); err != nil {
                log.Printf("handler error: %v", err)
                // Reject без requeue, если ошибка фатальна для сообщения
                // или с requeue для временных сбоев
                if shouldRequeue(err) {
                    d.Nack(false, true)
                } else {
                    d.Nack(false, false)
                }
                continue
            }

            // Подтверждение после успешной обработки
            d.Ack(false)
        }
    }
}

func shouldRequeue(err error) bool {
    // В реальности проверяем тип ошибки: сетевые сбои -> true, валидация -> false
    return true
}

4. Управление конкурентностью и Prefetch

Без ограничений брокер будет пушить сообщения быстрее, чем консьюмер успевает их обрабатывать. Это приводит к росту очереди в памяти клиента, увеличению потребления RAM и риску OOM.

Prefetch (QoS) ограничивает количество неподтвержденных сообщений, выдаваемых консьюмеру до получения ACK. Это создает естественное обратное давление (backpressure) и балансирует нагрузку между подами.

Go 
// Установка prefetch перед start consuming
err = ch.Qos(
    10,    // prefetch count: макс 10 неподтвержденных сообщений на воркер
    0,     // prefetch size: игнорируем (не всегда поддерживается)
    false, // global: false = только для этого консьюмера
)
if err != nil {
    return fmt.Errorf("set qos: %w", err)
}

5. Обработка ошибок, DLQ и повторные попытки

Бесконечный requeue при фатальных ошибках создает «ядовитые» сообщения (poison pills), которые блокируют очередь, так как консьюмер будет постоянно падать на одном и том же сообщении.

Правильная стратегия:

  1. Ограниченное количество повторных попыток с экспоненциальной задержкой.
  2. Перемещение в Dead Letter Queue (DLQ) после N попыток.
  3. Асинхронный разборщик (replayer) для ручной обработки или исправления данных в DLQ.
flowchart TD
    A["Сообщение с ошибкой"] --> B{"Попытка < MaxRetries?"}
    B -->|"Да"| C["Задержка с backoff<br>повторная отправка в очередь"]
    C --> D["Следующий консьюмер берет задачу"]
    B -->|"Нет"| E["Публикация в DLQ<br>с метаданными ошибки"]
    E --> F["Асинхронный разбор или ручное вмешательство"]

    classDef normal fill:#1e293b,stroke:#38bdf8,color:#f8fafc;
    classDef decision fill:#334155,stroke:#f59e0b,color:#f8fafc;
    classDef dlq fill:#4c0519,stroke:#f43f5e,color:#ffffff;
    class A,C,D,F normal;
    class B decision;
    class E dlq;

Для реализации в Go без сложных внешних механизмов используйте паттерн из статьи 29. Retry и backoff, комбинируя его с публикацией в отдельный dead_letter_queue через тот же канал брокера.

6. Производительность и Mechanical Sympathy

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

  • TCP Backpressure: Брокер использует SO_RCVBUF и TCP window scaling. Если консьюмер не читает данные, окно закрывается, брокер замедляет выдачу. В Go net.Conn.Read() блокируется, пока ядро не скопирует данные из буфера сокета в user space.
  • GC Pressure: Каждое сообщение создает []byte и обертку. При 10k msg/sec без пулов это ~50-100 МБ аллокаций/сек. sync.Pool для временных структур или прямая работа с io.Reader (например, в Kafka sarama с Fetch API) снижает нагрузку.
  • Syscall Overhead: Каждый ACK/NACK — отдельный фрейм по сети. Группировка подтверждений (batch acks) или увеличение prefetch снижает количество системных вызовов.
  • Goroutine Scheduler: Если обработчик выполняет длительную CPU-задачу без syscall или sync точки, планировщик может не передать управление другим горутинам. Используйте runtime.Gosched в циклах или выносите тяжелые вычисления в отдельные воркеры.

7. Ловушки и антипаттерны

8. Итог

  1. Очереди задач обеспечивают персистентность, масштабирование и изоляцию сбоев, но требуют явного управления подтверждением доставки.
  2. At-least-once — стандарт для распределенных систем. Exactly-once достигается через идемпотентность обработчика.
  3. Настраивайте Prefetch для балансировки нагрузки и предотвращения OOM клиента. Значение должно соответствовать скорости обработки и задержке сети.
  4. Используйте context.Context для graceful shutdown потребителя и контроля таймаутов.
  5. Реализуйте Dead Letter Queue для обработки ядовитых сообщений, избегая бесконечного requeue.
  6. Минимизируйте аллокации сообщений через sync.Pool и прямую работу с буферами для снижения GC-давления.
  7. Группируйте ACK или увеличивайте prefetch для снижения network syscall overhead.

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

Следующая статья: 28. Idempotency