Потребление: ConsumeWithContext

Q: Можно ли читать сообщения по одному без подписки?

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

Q: Что произойдёт с непрочитанными сообщениями, если воркер остановить?

Они останутся в очереди: при ручных подтверждениях неподтверждённое вернётся в очередь, а непрочитанное и не покидало её. При autoAck=true картина другая — всё, что брокер успел отдать, считается доставленным и исчезает из очереди вместе с процессом.

Q: Нужен ли отдельный канал на каждого потребителя?

Да: подписка и её поток доставок живут на одном канале, а канал не предназначен для конкурентного использования горутинами. Практика — один воркер, один канал, один цикл; для параллелизма поднимают несколько подписок, что и делает пул воркеров из семнадцатого урока.

В шестом уроке мы накопили в очереди tasks несколько сообщений — теперь их пора забрать. Наивное ожидание: есть метод вроде «вытащи одно сообщение». В amqp091-go механика другая: потребитель подписывается на очередь один раз, а дальше брокер сам доставляет каждое новое сообщение в поток, который вы читаете как обычный Go-канал. Из этого следует всё остальное: воркер — это цикл по каналу, остановка потребления — это закрытие потока, а конкуренция воркеров — это несколько подписок на одну очередь. Сместился сам центр тяжести: не вы ходите за сообщениями, они приходят к вам. Разберём, как устроена подписка, что приходит в доставке, где у цикла воркера конец и почему флаг autoAck, который так удобно ставить в true для быстрых примеров, в продакшене стоит дороже, чем кажется.

Чтение очереди go: подписка ConsumeWithContext#

Сигнатура ch.ConsumeWithContext(ctx context.Context, queue, consumer string, autoAck, exclusive, noLocal, noWait bool, args amqp.Table) (<-chan Delivery, error) возвращает поток доставок, и разбор rabbitmq consume go логично идти по аргументам. Queue — имя очереди, объявленной заранее, как в пятом уроке. Consumer — тег потребителя, человекочитаемая метка, по которой видны подписки в management UI; пустая строка просит брокер сгенерировать имя вида ctag-N. AutoAck — подтверждать ли сообщение автоматически в момент доставки; здесь только упомянем, что для надёжного потребителя нужен false, а целиком вопрос гарантий — тема восьмого урока. Exclusive — эксклюзивная подписка: второй потребитель на ту же очередь получит ошибку канала ACCESS_REFUSED, вариант для очереди, обрабатываемой ровно одним воркером. NoLocal — флаг из спецификации, который серверы RabbitMQ не поддерживают и игнорируют, поэтому пишут false. NoWait экономит ответ брокера, как и в объявлении очереди. Args — карта особых свойств вроде x-priority для приоритета потребителя; в первых подписках передают nil. Ошибка возвращается, если очередь не существует или подписка нарушает эксклюзивность.

amqp.Delivery: что приходит воркеру#

Каждое значение из потока — amqp.Delivery, сообщение с сопроводительными данными протокола. Body — тот самый []byte, который продюсер положил в шестом уроке; ContentType и Headers приезжают с ним же, поэтому потребитель может ветвиться по формату и атрибутам, не разбирая содержимое. DeliveryTag — номер доставки в рамках канала, он понадобится для подтверждений: ack ссылается на сообщение по этому номеру. Redelivered — флаг повторной доставки: true значит, что сообщение уже отдавалось кому-то и вернулось в очередь без подтверждения; по нему ловят циклы перезапусков и ядовитые сообщения, о которых пойдёт речь в шестнадцатом уроке. RoutingKey и Exchange напоминают, откуда и по какому маршруту сообщение приехало, MessageId — поле, которое продюсер мог задать для дедупликации. Полезно держать в голове: Delivery — это снимок сообщения в момент доставки, а не живая ссылка; изменение структуры на месте ничего не меняет в брокере, а размер потоков ограничивается настройкой prefetch из десятого урока. Всё, что воркер хочет сделать с сообщением, он делает по этим полям и по телу.

Воркер: подписка и цикл с подтверждением

package main

import (
        "context"
        "fmt"
        "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.QueueDeclare("tasks", true, false, false, false, nil)
        if err != nil {
                log.Fatal(err)
        }

        ctx, cancel := context.WithCancel(context.Background())
        defer cancel()

        msgs, err := ch.ConsumeWithContext(ctx, "tasks", "receipt-worker",
                false, false, false, false, nil)
        if err != nil {
                log.Fatal(err)
        }

        for d := range msgs {
                fmt.Println("обрабатываю:", string(d.Body))
                if err := d.Ack(false); err != nil {
                        log.Println("ack failed:", err)
                }
        }
        log.Println("поток доставок закрыт, воркер завершается")
}

Подписка с autoAck=false и явным d.Ack(false) после обработки — минимально честный воркер; механика ack целиком разбирается в следующем уроке. Цикл for range живёт, пока открыт поток доставок: пока брокер присылает сообщения и контекст не отменён. Запустите воркер и в другом терминале продюсер из шестого урока — сообщения напечатаются по мере публикации.

Цикл воркера: анатомия for range по Delivery#

Цикл воркера — самое короткое описание потребителя: получил, обработал, подтвердил. Внутри итерации живут три вопроса, которые определяют качество воркера. Первый — обработка: тяжёлая работа делается здесь же, в этой горутине; распараллеливание достигается не запуском горутин из цикла на одном канале, а несколькими подписками — по одному каналу и циклу на воркера, как требует четвёртый урок; рабочий пул целиком собирают в семнадцатом. Второй — подтверждение: ack в конце итерации, а не в начале; что случится при панике посреди обработки, разберём в следующем уроке вместе с recover. Третий — выход: range завершится, когда поток доставок закроется — по отмене контекста, закрытию канала или обрыву соединения; после цикла воркер спокойно доисполняет финализацию и возвращает управление. Форма с ok-проверкой, d, ok := <-msgs, нужна там, где после закрытия потока делают отдельный шаг: дожимают незавершённые задачи, пишут метрику остановки. В простом воркере достаточно range и строки после него — она и есть сигнал, что подписка кончилась.

Поля amqp.Delivery, нужные воркеру#

Поле	Тип	Зачем воркеру
Body	[]byte	содержимое сообщения, то, что обрабатываем
DeliveryTag	uint64	номер доставки для Ack, Nack и Reject
Redelivered	bool	признак повторной доставки: ловим яды и циклы
ContentType	string	формат тела без заглядывания внутрь
Headers	amqp.Table	атрибуты продюсера: трассировка, версия схемы
RoutingKey, Exchange	string	откуда и по какому маршруту приехало сообщение

autoAck=true: сообщение умирает вместе с процессом

Кейс: воркер рассылок работал с autoAck=true — брокер считает сообщение доставленным в момент отправки и удаляет его из очереди. Процесс обновили в час пик, воркер упал на середине партии. Часть сообщений была уже прочитана и не обработана, а в очереди их больше нет: подтверждать нечего, восстанавливать неоткуда. Нашли потерю по сверке с базой — писем отправлено меньше, чем задач создано. С autoAck=false непрочитанное осталось бы в очереди unacknowledged и вернулось бы новому воркеру автоматически. Дорога это граблина в один шаг — но обходить её стоит с первого дня.

Отмена потребления: контекст вместо обрыва#

ConsumeWithContext привязывает подписку к контексту: отмена ctx приводит к отмене потребителя, поток доставок закрывается, и range воркера завершается штатно. Это самый аккуратный способ остановить чтение — без закрытия канала и соединения, которые понадобятся ещё для метрик и финальных публикаций. Регистрация подписки сама уважает дедлайн: если брокер не отвечает, вызов вернётся с ошибкой по контексту, а не повиснет. Сравните с грубыми способами: ch.Close закрывает канал целиком, вместе со всеми подписками и возможностью публиковать, а обрыв соединения роняет и их, и всё незавершённое. Полный рецепт остановки сервиса — сигналы, дожим последних задач, ack на выходе — соберём в двадцать первом уроке про graceful shutdown; здесь важно, что механизм у него уже есть: контекст, переданный в ConsumeWithContext, и есть та ручка, за которую воркер останавливают из снаружи.

Потребление с дедлайном

ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()

msgs, err := ch.ConsumeWithContext(ctx, "tasks", "short-live",
        false, false, false, false, nil)
if err != nil {
        log.Fatal(err)
}

for d := range msgs {
        fmt.Println("получено:", string(d.Body))
}
fmt.Println("30 секунд вышли — потребление отменено")

Через полминуты контекст отменится, поток доставок закроется, и range завершится — печатается финальная строка. Такой воркер удобно запускать для локальных экспериментов: он сам заканчивается, не забирая очередь под себя. В боевом коде вместо таймаута стоит контекст, отменяемый по сигналу завершения сервиса.

Несколько потребителей: кому достанется сообщение#

Подпишите на одну очередь двух воркеров — и брокер начнёт раздавать сообщения по кругу: первое первому, второе второму, и так далее; каждое сообщение уходит ровно одному подписчику, а не всем. Механика та же, что лежит под масштабированием: подняли второй контейнер с воркером — очередь сама начала делиться, никаких изменений кода. Три уточнения к картинке. Раздача ведётся по готовности: воркер, который ещё не подтвердил предыдущее сообщение, при ручных ack с prefetch — тема десятого урока — новые получать не будет, и «по кругу» станет честнее. Если нужно, чтобы каждое сообщение видели все подписчики, очередь не подходит — это работа fanout-обменника из двенадцатого урока и отдельных очередей на подписчика. И последнее: теги потребителей видны в management UI на вкладке очереди — по ним проверяют, что воркеры действительно подписались, а ConsumerCount из QueueDeclare показывает их число без UI.

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

Воркеру нужен живой брокер: в браузерной песочнице доступны только fmt, strings, strconv, math, sort, errors и time — без сети, поэтому ConsumeWithContext в браузере не запустится. Соберите воркер и продюсер из шестого урока в двух терминалах и посмотрите, как сообщения перетекают между ними; поднимите вторую копию воркера и убедитесь, что брокер сам делит очередь. Интерактивы ниже проверят сигнатуру, поведение autoAck и порядок строк воркера.

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

autoAck=false в боевом воркере: подтверждение только после обработки, механика — в восьмом уроке.
Один канал и один цикл range на воркера; параллелизм — количеством подписок, не горутинами поверх канала.
Останов через отмену контекста: поток закроется, range завершится, финализация дойдёт до конца.
Redelivered=true — повод залогировать и присмотреться: сообщение уже кто-то не осилил.
Тег потребителя делайте осмысленным: по нему ищут подписку в management UI и логах.
Эксклюзивная подписка — когда очередь обслуживает ровно один воркер; иначе ACCESS_REFUSED второму.

Дальше по маршруту: в восьмом уроке разбирают Ack, Nack и Reject — что происходит с неподтверждённым сообщением при падении воркера и как выглядит requeue. Гонку честности между воркерами выравнивает десятый урок про prefetch, а сборка нескольких воркеров в пул — семнадцатый. Остановка сервиса без потерь — двадцать первый урок про graceful shutdown, рассылка всем подписчикам — двенадцатый про fanout. Механика чтения из канала Go разобрана в уроке про горутины — там же про range по закрытому каналу. Предыдущая ступень — публикация в шестом уроке, карта курса — на странице курса.

Проверка: остановка потребления

Контекст, переданный в ConsumeWithContext, отменили. Что произойдёт?

Проверка: два потребителя на очереди

На очередь tasks подписаны два воркера. Кому достанется очередное сообщение?

Что выведет воркер?

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

ctx, cancel := context.WithCancel(context.Background())
msgs, _ := ch.ConsumeWithContext(ctx, "tasks", "w",
        false, false, false, false, nil)

go func() {
        time.Sleep(50 * time.Millisecond)
        cancel()
}()

for d := range msgs {
        fmt.Println("got:", string(d.Body))
}
fmt.Println("done")

Соберите подписку воркера

Допишите вызов подписки, обращение к телу доставки и подтверждение.

msgs, err := ch.("tasks", "worker", false, false, false, false, nil) if err != nil { log.Fatal(err) } for d := range msgs { fmt.Println(string(d.)) d.(false) }

Соберите воркер по строкам

Расставьте строки в порядке выполнения: соединение, проверки, закрытие, канал, подписка, цикл.

Строки

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

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

Карточки: ConsumeWithContext

Термин

ConsumeWithContext

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

Определение

Подписка на очередь: возвращает поток доставок <-chan Delivery. Контекст управляет остановкой потребителя, autoAck — подтверждением.

1 / 6

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

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

Вывод

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

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

ConsumeWithContext возвращает поток доставок: воркер — это цикл for range по каналу Delivery.
Поля Delivery: Body и ContentType для работы, DeliveryTag для подтверждений, Redelivered для ловли повторов.
autoAck=true удаляет сообщение в момент доставки — падение воркера теряет данные; боевой вариант false.
Остановка потребления — отмена контекста: поток закроется, range завершится, финализация выполнится.
Несколько потребителей на очереди получают сообщения по кругу, каждое — ровно одному.
Подписка — один на канал и горутину; параллелизм строят несколькими воркерами, а не горутинами поверх потока.

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

Запустите воркер из урока и продюсер из шестого урока в разных терминалах, затем добавьте вторую копию воркера и опубликуйте пять сообщений — убедитесь, что они распределились между воркерами. После этого замените таймаут в примере на контекст, отменяемый по Ctrl+C, и проверьте, что воркер печатает финальную строку и завершается сам.

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

Можно ли читать сообщения по одному без подписки?