--- title: "Потребление: ConsumeWithContext" description: "ConsumeWithContext в Go: подписка на очередь, канал amqp.Delivery, цикл воркера, цена autoAck и отмена потребления. Чтение очереди go с примерами кода." course: "RabbitMQ в Go: очереди сообщений на amqp091-go" url: "https://golangmaster.ru/courses/rabbitmq/lessons/07-potreblenie-soobschenij/" published: "2026-09-19" updated: "2026-09-21" minutes: 20 --- # Потребление: ConsumeWithContext *Курс «RabbitMQ в Go: очереди сообщений на amqp091-go», урок 7 из 25. ~20 минут. Обновлено 21.09.2026. Источник: https://golangmaster.ru/courses/rabbitmq/lessons/07-potreblenie-soobschenij/* Чтение очереди go через rabbitmq consume начинается не с команды «дай сообщение», а с подписки: ConsumeWithContext регистрирует потребителя и возвращает обычный Go-канал, из которого воркер читает доставки. В этом уроке разберём сигнатуру, поля amqp.Delivery, анатомию цикла воркера, цену autoAck и то, как корректно остановить потребление, не оставив очередь без присмотра. В шестом уроке мы накопили в очереди 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 из десятого урока. Всё, что воркер хочет сделать с сообщением, он делает по этим полям и по телу. ### Воркер: подписка и цикл с подтверждением ```go 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, и есть та ручка, за которую воркер останавливают из снаружи. ### Потребление с дедлайном ```go 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 второму. Дальше по маршруту: в [восьмом уроке](/courses/rabbitmq/lessons/08-ack-nack/) разбирают Ack, Nack и Reject — что происходит с неподтверждённым сообщением при падении воркера и как выглядит requeue. Гонку честности между воркерами выравнивает [десятый урок про prefetch](/courses/rabbitmq/lessons/10-prefetch/), а сборка нескольких воркеров в пул — [семнадцатый](/courses/rabbitmq/lessons/17-worker-pool/). Остановка сервиса без потерь — [двадцать первый урок про graceful shutdown](/courses/rabbitmq/lessons/21-graceful-rabbitmq/), рассылка всем подписчикам — [двенадцатый про fanout](/courses/rabbitmq/lessons/12-fanout-exchange/). Механика чтения из канала Go разобрана в [уроке про горутины](/courses/go-advanced/lessons/04-gorutiny/) — там же про range по закрытому каналу. Предыдущая ступень — публикация в [шестом уроке](/courses/rabbitmq/lessons/06-publikaciya-soobschenij/), карта курса — на [странице курса](/courses/rabbitmq/). ### Проверка: остановка потребления **Вопрос:** Контекст, переданный в ConsumeWithContext, отменили. Что произойдёт? - ✅ Поток доставок закроется, и цикл range воркера завершится - Соединение с брокером разорвётся - Воркер продолжит получать сообщения, но с флагом Redelivered - Брокер удалит очередь вместе с сообщениями Отмена контекста отменяет потребителя: канал Delivery закрывается, range завершается, воркер доходит до финализации. Соединение и сам канал продолжают жить — это и делает контекст аккуратным способом остановки. ### Проверка: два потребителя на очереди **Вопрос:** На очередь tasks подписаны два воркера. Кому достанется очередное сообщение? - Обоим воркерам — каждый получит копию - ✅ Ровно одному из них, раздача идёт по кругу - Тому, у кого меньше очередь незавершённых задач, если prefetch настроен - Никому: вторая подписка вернёт ACCESS_REFUSED Одно сообщение уходит ровно одному подписчику — очередь раздаёт задачи по кругу. Копии всем делают fanout-обменники с отдельными очередями, а ACCESS_REFUSED случится только при exclusive-подписке. Честность раздачи настраивается prefetch из десятого урока. ### Что выведет воркер? ```go 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") ``` **Что выведет программа?** - ✅ все сообщения из очереди, затем done — поток закрылся после cancel - только первое сообщение, затем зависание навсегда - сообщения, затем done, и снова сообщения: подписка возобновилась - паника: чтение из закрытого канала После cancel поток доставок закрывается штатно, range завершается — печатается done. Чтение из закрытого канала не паникует в range: это штатный выход из цикла, а не ошибка. ### Соберите подписку воркера Допишите вызов подписки, обращение к телу доставки и подтверждение. ```go 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, содержимое лежит в поле Body, подтверждение — метод Ack. Правильные ответы в пропусках: `ConsumeWithContext`, `Body`, `Ack` ### Соберите воркер по строкам Расставьте строки в порядке выполнения: соединение, проверки, закрытие, канал, подписка, цикл. Правильный порядок строк: 1. `conn, err := amqp.Dial(url)` 2. `if err != nil { log.Fatal(err) }` 3. `defer conn.Close()` 4. `ch, err := conn.Channel()` 5. `msgs, err := ch.ConsumeWithContext(ctx, "tasks", "", false, false, false, false, nil)` 6. `for d := range msgs { d.Ack(false) }` ### Карточки: ConsumeWithContext - **ConsumeWithContext** — Подписка на очередь: возвращает поток доставок <-chan Delivery. Контекст управляет остановкой потребителя, autoAck — подтверждением. - **amqp.Delivery** — Одна доставка: Body, ContentType, Headers, DeliveryTag, Redelivered, RoutingKey. Снимок сообщения в момент выдачи воркеру. - **consumer tag** — Метка подписки: видна в management UI и в уведомлениях. Пустая строка — брокер сгенерирует ctag-N сам. - **autoAck** — True: брокер считает сообщение доставленным в момент отправки. False: сообщение ждёт явного Ack и вернётся, если воркер не подтвердил. - **Redelivered** — Флаг повторной доставки: сообщение уже отдавалось потребителю и вернулось без подтверждения. Повод для лога и анализа. - **round-robin** — Раздача сообщений между подписчиками одной очереди по кругу: каждое сообщение получает ровно один потребитель. ## Коротко: что запомнить - ConsumeWithContext возвращает поток доставок: воркер — это цикл for range по каналу Delivery. - Поля Delivery: Body и ContentType для работы, DeliveryTag для подтверждений, Redelivered для ловли повторов. - autoAck=true удаляет сообщение в момент доставки — падение воркера теряет данные; боевой вариант false. - Остановка потребления — отмена контекста: поток закроется, range завершится, финализация выполнится. - Несколько потребителей на очереди получают сообщения по кругу, каждое — ровно одному. - Подписка — один на канал и горутину; параллелизм строят несколькими воркерами, а не горутинами поверх потока. ## Домашнее задание Запустите воркер из урока и продюсер из шестого урока в разных терминалах, затем добавьте вторую копию воркера и опубликуйте пять сообщений — убедитесь, что они распределились между воркерами. После этого замените таймаут в примере на контекст, отменяемый по Ctrl+C, и проверьте, что воркер печатает финальную строку и завершается сам. ## Вопросы и ответы по уроку ### Можно ли читать сообщения по одному без подписки? В amqp091-go есть ch.Get для однократного чтения, но для сервисов он не подходит: подписка держит поток открытым и не тратит сетевой рейс на каждое сообщение. Get годится для административных скриптов и отладки, воркеры пишут через ConsumeWithContext. ### Что произойдёт с непрочитанными сообщениями, если воркер остановить? Они останутся в очереди: при ручных подтверждениях неподтверждённое вернётся в очередь, а непрочитанное и не покидало её. При autoAck=true картина другая — всё, что брокер успел отдать, считается доставленным и исчезает из очереди вместе с процессом. ### Нужен ли отдельный канал на каждого потребителя? Да: подписка и её поток доставок живут на одном канале, а канал не предназначен для конкурентного использования горутинами. Практика — один воркер, один канал, один цикл; для параллелизма поднимают несколько подписок, что и делает пул воркеров из семнадцатого урока.