RabbitMQ в Go: публикация и обработка сообщений

Принять изображение можно быстро, подготовить несколько его размеров — уже нет. Эта работа занимает секунды и не должна удерживать HTTP-соединение открытым.

Разделим ответственность. media-api сохранит описание задачи и быстро ответит клиенту, RabbitMQ передаст задачу, а image-worker обработает файл в своём темпе. Проследим одну задачу — job-456 — от публикации до подтверждения.

media-api                 RabbitMQ                    image-worker
    │                         │                             │
    │      image_resize       │                             │
    │      {job_id: ...}      │                             │
    ├────────────────────────▶│                             │
    │                         ├────────────────────────────▶│
    │                         │ изменить размер и сохранить │
    │                         │◀─────────── Ack ────────────┤

К концу статьи этот поток будет работать, а затем мы намеренно остановим обработчик до Ack и увидим, почему RabbitMQ допускает повторную обработку.

Примеры используют amqp091-go версии 1.10.0. Нужен запущенный RabbitMQ и переменная AMQP_URL, например amqp://guest:guest@localhost:5672/ для локального окружения.

Что именно лежит в очереди

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

type ImageResizeJob struct {
	AssetID      string `json:"asset_id"`
	JobID        string `json:"job_id"`
	OriginalPath string `json:"original_path"`
}

Для нашей задачи JSON выглядит так:

{
  "asset_id": "asset-42",
  "job_id": "job-456",
  "original_path": "uploads/asset-42/original.jpg"
}

JobID станет идентификатором операции: если сообщение придёт повторно, обработчик сможет проверить состояние job-456, а не создавать второй независимый результат.

Собираем маршрут сообщения

media-api становится издателем сообщения (producer), а image-worker — его получателем (consumer). Издатель отправляет сообщение не прямо «в RabbitMQ», а в точку обмена (exchange). Она выбирает очередь по ключу маршрутизации (routing key). Для одной очереди достаточно стандартной точки обмена (default exchange): её имя — пустая строка, а ключ маршрутизации совпадает с именем очереди image_resize.

издатель → стандартная точка обмена → очередь image_resize → получатель

При старте откроем одно TCP-соединение, внутри него — легковесный AMQP-канал для протокольных операций, затем объявим очередь. Издатель и обработчик запускаются разными процессами, поэтому каждый выполняет этот код со своим соединением.

package messaging

import (
	"fmt"

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

type Client struct {
	conn  *amqp.Connection
	ch    *amqp.Channel
	queue string
}

func New(amqpURL, queueName string) (*Client, error) {
	conn, err := amqp.Dial(amqpURL)
	if err != nil {
		return nil, fmt.Errorf("connect to RabbitMQ: %w", err)
	}

	ch, err := conn.Channel()
	if err != nil {
		_ = conn.Close()
		return nil, fmt.Errorf("open channel: %w", err)
	}

	queue, err := ch.QueueDeclare(
		queueName,
		true,  // durable
		false, // auto-delete
		false, // exclusive
		false, // no-wait
		nil,
	)
	if err != nil {
		_ = ch.Close()
		_ = conn.Close()
		return nil, fmt.Errorf("declare queue: %w", err)
	}

	return &Client{conn: conn, ch: ch, queue: queue.Name}, nil
}

func (c *Client) Close() error {
	if err := c.ch.Close(); err != nil {
		_ = c.conn.Close()
		return err
	}
	return c.conn.Close()
}

durable: true сохраняет определение очереди после перезапуска RabbitMQ. Это ещё не обещание сохранить каждое сообщение: устойчивость очереди и сообщения настраивается отдельно.

Параметры повторного объявления должны совпадать. Если издатель создал image_resize как устойчивую очередь (durable: true), а обработчик объявляет то же имя с durable: false, RabbitMQ закроет канал с ошибкой конфигурации вместо молчаливого изменения очереди.

Публикуем задачу

Теперь сериализуем ImageResizeJob и отправим его через стандартную точку обмена:

func (c *Client) PublishImageResize(
	ctx context.Context,
	message ImageResizeJob,
) error {
	body, err := json.Marshal(message)
	if err != nil {
		return fmt.Errorf("marshal image resize job: %w", err)
	}

	err = c.ch.PublishWithContext(
		ctx,
		"",      // стандартная точка обмена
		c.queue, // ключ маршрутизации
		false,   // mandatory
		false,   // immediate
		amqp.Publishing{
			ContentType:  "application/json",
			DeliveryMode: amqp.Persistent,
			Body:         body,
		},
	)
	if err != nil {
		return fmt.Errorf("publish image resize job: %w", err)
	}
	return nil
}

Этот фрагмент продолжает пакет messaging; ему нужны импорты context и encoding/json. DeliveryMode: amqp.Persistent просит RabbitMQ сохранить сообщение на диск. Вместе с устойчивой очередью это лучше защищает от обычного перезапуска RabbitMQ, чем настройки по умолчанию.

Остановите обработчик и опубликуйте задачу job-456. В интерфейсе управления RabbitMQ у очереди image_resize должно появиться одно сообщение в состоянии Ready. После запуска обработчика во время работы оно находится в состоянии Unacked, а после подтверждения исчезает из очереди. Если обработка мгновенная, промежуточное состояние можно не успеть увидеть — временная пауза в локальном обработчике сделает его заметным.

Получаем по одной тяжёлой задаче

Получатель сначала задаёт prefetch. Значение 1 означает: не отправлять ему следующую задачу, пока текущая остаётся неподтверждённой.

Оба следующих фрагмента составляют тело метода Consume. Параметр handler — функция обработки задачи: она получает тело сообщения и сообщает, нужно ли вернуть неудачную доставку в очередь.

type Handler func(
	context.Context,
	[]byte,
) (requeue bool, err error)

// Следующие фрагменты находятся внутри:
// func (c *Consumer) Consume(ctx context.Context, handler Handler) error

if err := c.ch.Qos(1, 0, false); err != nil {
	return fmt.Errorf("set qos: %w", err)
}

deliveries, err := c.ch.ConsumeWithContext(
	ctx,
	c.queue,
	"",
	false, // ручной Ack
	false,
	false,
	false,
	nil,
)
if err != nil {
	return fmt.Errorf("start consumer: %w", err)
}

prefetch: 1 хорошо подходит последовательной обработке изображений. Это не универсальное «правильное значение»: быстрый обработчик или несколько параллельных горутин могут оправдать большее окно, измеренное под реальной нагрузкой.

Пройдём по доставкам. Наш обработчик возвращает requeue: true только для кратковременного сбоя, который действительно имеет смысл повторить сразу. Постоянную ошибку он возвращает с requeue: false. Для записи в журнал в этом фрагменте также нужен импорт log.

for delivery := range deliveries {
	log.Printf(
		"received image job redelivered=%t",
		delivery.Redelivered,
	)

	requeue, handleErr := handler(ctx, delivery.Body)
	if handleErr != nil {
		if err := delivery.Nack(false, requeue); err != nil {
			return fmt.Errorf("nack delivery: %w", err)
		}
		continue
	}

	if err := delivery.Ack(false); err != nil {
		return fmt.Errorf("ack delivery: %w", err)
	}
}

if err := ctx.Err(); err != nil {
	return err
}
return fmt.Errorf("delivery channel closed")

Ack отправляется после того, как уменьшенные изображения сохранены и состояние job-456 зафиксировано. Если подтвердить доставку раньше, а затем потерять процесс, RabbitMQ честно удалит сообщение и уже не сможет восстановить работу.

Подтверждение должно идти через тот же AMQP-канал, который получил доставку. При завершении контекста ConsumeWithContext отменяет получателя; закрытие соединения возвращает оставшиеся неподтверждённые сообщения в очередь.

Останавливаем обработчик до Ack

Теперь проверим гарантию, а не просто прочитаем о ней:

Добавьте перед завершением handler временную паузу и запись в журнал result saved for job-456.
Опубликуйте одну задачу и дождитесь этого лога.
Остановите процесс до того, как управление вернётся к delivery.Ack(false).
Запустите обработчик снова.

Второй запуск получит ту же задачу, а delivery.Redelivered станет true. RabbitMQ гарантирует доставку как минимум один раз (at least once): сообщение не должно пропасть из-за падения получателя, но иногда будет обработано повторно.

Это меняет проектирование обработчика. Перед записью результата он проверяет job_id и выполняет переход состояния идемпотентно. Повторная задача job-456 может перезаписать тот же объект или увидеть, что работа уже завершена, но не должна создать второй набор файлов и записей.

Разделяем три разных окна потери

Ручной Ack решает только сбой получателя. У издателя и RabbitMQ есть собственные границы:

Момент сбоя	Нужный механизм
обработчик завершился во время работы	ручной `Ack` и идемпотентность
RabbitMQ перезапустился после принятия сообщения	устойчивая очередь и сохранение сообщения на диск
соединение издателя оборвалось во время публикации	подтверждения публикации и идемпотентная публикация

В amqp091-go 1.10.0 имя PublishWithContext может ввести в заблуждение: контекст не прерывает саму запись в сетевое соединение. Кроме того, возврат nil означает, что клиент записал публикацию без локальной ошибки, а не то, что RabbitMQ надёжно принял её.

Если API нельзя отвечать успехом до подтверждения RabbitMQ, включите подтверждения публикации (publisher confirms) и дождитесь положительного результата через DeferredConfirmation.WaitContext. Детали и ограничения описаны в официальной документации RabbitMQ.

Даже подтверждения не превращают сеть в обработку ровно один раз (exactly once): при обрыве клиент иногда не знает, успел ли RabbitMQ принять сообщение. Стабильный job_id позволяет безопасно повторить попытку или применить транзакционный журнал исходящих событий (transactional outbox), если запись в БД и публикация должны образовать одну бизнес-операцию.

Ошибка не всегда означает немедленный возврат в очередь

Nack(false, true) возвращает сообщение в очередь сразу. Если JSON повреждён или исходный объект не существует, обработчик получит ту же постоянную ошибку и создаст горячий бесконечный цикл.

Разделяйте исходы:

успех — Ack(false);
кратковременная ошибка — ограниченная повторная попытка с задержкой;
постоянная ошибка — Nack(false, false) с настроенной точкой обмена недоставленных сообщений;
потеря процесса или соединения — доставка остаётся неподтверждённой и возвращается в очередь.

Для очереди недоставленных сообщений (dead-letter queue, DLQ) основная очередь должна быть связана с точкой обмена недоставленных сообщений (dead-letter exchange, DLX). Без этой настройки Nack(false, false) удалит сообщение. В рабочем окружении DLX удобнее задавать политикой RabbitMQ: её можно менять без переобъявления очереди.

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

Соединение тоже имеет жизненный цикл

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

После неожиданного разрыва процесс может перезапустить оркестратор. Если переподключение выполняет само приложение, оно должно ограничить частоту попыток и заново объявить точки обмена, очереди и привязки: новое соединение не наследует AMQP-канал старого.

Если один процесс и публикует, и получает сообщения, выделите ролям разные каналы. Библиотека сериализует конкурентные публикации, но общий канал связывает их ошибки протокола, закрытие и порядок подтверждений публикации. Базовую последовательность можно сравнить с официальным руководством RabbitMQ для Go.

Что проверить перед использованием

Издатель и обработчик одинаково объявляют устойчивую очередь image_resize.
Сообщение содержит координаты работы, ContentType: application/json и DeliveryMode: amqp.Persistent.
Обработчик использует ручной Ack, подходящий prefetch и подтверждает только сохранённый результат.
Повторная доставка job-456 не создаёт второй бизнес-результат.
Временные повторы ограничены, а постоянные ошибки попадают в реально настроенную DLQ.
Команда понимает, требуются ли подтверждения публикации или транзакционный журнал исходящих событий.
После переподключения топология объявляется заново.

Связь сообщения с HTTP-контрактом задачи продолжена в статье «Асинхронные задачи и опрос состояния в REST API».

RabbitMQ в Go: публикация и обработка сообщений

Что именно лежит в очереди

Собираем маршрут сообщения

Публикуем задачу

Получаем по одной тяжёлой задаче

Останавливаем обработчик до Ack

Разделяем три разных окна потери

Ошибка не всегда означает немедленный возврат в очередь

Соединение тоже имеет жизненный цикл

Что проверить перед использованием

Следующий шаг после статьи

Продолжить изучение

Похожие статьи

Redis, RabbitMQ и S3-хранилище в Go-проектах

Асинхронные задачи и опрос состояния в REST API

Redis в Go: клиент, кэширование и TTL