Graceful shutdown потребителя

Q: Нужно ли вручную ставить сообщения обратно в очередь при остановке?

Нет: неподтверждённые доставки брокер вернёт сам при закрытии соединения — как при обрыве из двадцатого урока. В дожиме вручную подтверждают только то, что доделано. Попытка Nack из уже закрывающегося канала вернёт ErrClosed и добавит в логи ложную ошибку, которую дежурный примет за реальную проблему.

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

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

Q: Останавливать потребителя и продюсера одним контекстом?

Удобно и в большинстве случаев достаточно: один NotifyContext, разные этапы с собственными дедлайнами. Если продюсер должен проработать дольше потребителя — например, дослать накопленный буфер, — выдайте ему отдельный контекст с собственным тайм-аутом, но сигнальный корень оставьте один: две шины остановки путают порядок закрытия.

Пуш-рассылка в двенадцать ноль-ноль, сто тысяч подписчиков, воркер держит три доставки — и деплой-пайплайн присылает SIGTERM. Что произойдёт дальше, процесс решает сам: подписка ConsumeWithContext из седьмого урока принимает контекст, и у потребителя появляется законный способ остановиться не на полуслове. Без этой механики всё заканчивается одинаково: процесс умирает мгновенно, неподтверждённые доставки возвращаются брокером, а полузаписанные строки в базе и отправленные наполовину письма остаются как след. Такой обрыв сообщения не теряет — очередь их вернёт, — зато дублирует работу и рвёт побочные эффекты, которые нельзя переиграть: письмо уже ушло, повторное списание пройдёт второй раз. Дожимом называют фазу, когда новые доставки больше не берутся, а взятые доедаются до ack. Дожим превращает SIGTERM из аварии в процедуру, которую можно читать по логам: сигнал, хвост задач, чистый выход. Ниже — минимальный потребитель с сигналами и правила, без которых дожим не работает.

Rabbitmq graceful shutdown: сигнал и контекст#

Ядро rabbitmq graceful shutdown — превращение Unix-сигнала в контекст. Функция signal.NotifyContext из стандартной библиотеки возвращает контекст, который отменяется по os.Interrupt — это Ctrl+C — и по syscall.SIGTERM, который шлют docker stop, systemd и оркестраторы, плюс функцию stop для снятия обработчика. Дальше сигнальный контекст работает как любая другая отмена: он передаётся в ConsumeWithContext, а отмена закрывает поток доставок — воркер выходит из range сам, без паники и без принудительного прерывания горутин, которого в Go не существует. Отмена здесь кооперативная, как в девятом уроке курса по Go: контекст сообщает о желании остановиться, а решения принимают горутины, которые этот контекст получили. Важная оговорка про SIGKILL: этот сигнал перехватить нельзя, процесс умирает на месте, и единственная защита от его последствий — идемпотентные обработчики, а не код остановки. Поэтому дожим проектируют под SIGTERM и укладывают в grace period, который даёт платформа: у docker stop это десять секунд по умолчанию, в Kubernetes — terminationGracePeriodSeconds, в systemd — TimeoutStopSec.

Потребитель с сигналами и дожимом

package main

import (
        "context"
        "log"
        "os"
        "os/signal"
        "sync"
        "syscall"
        "time"

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

func main() {
        ctx, stop := signal.NotifyContext(context.Background(),
                os.Interrupt, syscall.SIGTERM)
        defer stop()

        conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
        if err != nil {
                log.Fatal(err)
        }

        ch, err := conn.Channel()
        if err != nil {
                log.Fatal(err)
        }
        if err := ch.Qos(3, 0, false); err != nil {
                log.Fatal(err)
        }

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

        var wg sync.WaitGroup
        wg.Add(1)
        go func() {
                defer wg.Done()
                for d := range msgs { // после отмены ctx поток закроется
                        if err := process(d.Body); err != nil {
                                d.Nack(false, false) // в retry из урока 16
                                continue
                        }
                        d.Ack(false)
                }
        }()

        <-ctx.Done()
        log.Println("сигнал получен: новые доставки не берём, доедаем текущие")

        done := make(chan struct{})
        go func() {
                wg.Wait()
                close(done)
        }()
        select {
        case <-done:
                log.Println("дожим завершён, выходим")
        case <-time.After(8 * time.Second):
                log.Println("дожим затянулся: закрываем канал принудительно")
                ch.Close() // unacked вернутся брокеру
        }
        conn.Close()
}

func process(body []byte) error {
        time.Sleep(300 * time.Millisecond) // имитация работы
        log.Println("обработано:", string(body))
        return nil
}

Сигнальный контекст прошёл в ConsumeWithContext, поэтому после SIGTERM брокер перестаёт отдавать новые доставки, а текущий range доходит до конца. Дожим ограничен восемью секундами: WaitGroup ждёт в отдельной горутине, а select в main выбирает между успехом и тайм-аутом — без этого одна зависшая задача из двадцать третьего урока повесила бы остановку навсегда. При тайм-ауте канал закрывается принудительно, его unacked возвращаются в очередь: недоделанная работа будет выполнена заново, но хотя бы не потеряется.

Фазы штатной остановки#

Фаза	Что происходит	Кто отвечает
сигнал пришёл	signal.NotifyContext отменяет контекст	main процесса
подписка гаснет	ConsumeWithContext закрывает поток: новых доставок нет	отмена контекста
текущие доставки	воркеры дообрабатывают и зовут Ack или Nack	горутины-воркеры
дожим с дедлайном	WaitGroup плюс тайм-аут: зависшая задача не вечна	код остановки
выход	conn.Close после дожима, иначе ack не уйдут	main

Дожим длиннее grace period

Кейс: команда выставила тайм-аут дожима в сорок секунд, не посмотрев, сколько платформа даёт на остановку. Docker stop ждёт десять секунд и присылает SIGKILL: обработчики не вызываются вовсе, unacked возвращаются брокером, а задача, которая была наполовину сделана, повторится с нуля после рестарта. Проблема даже не в потере сообщений, а в двойных побочных эффектах: письмо уйдёт, повторное списание пройдёт второй раз. Правило простое: дожим обязан укладываться в grace period с запасом — его тайм-аут заметно меньше, чем TimeoutStopSec или terminationGracePeriodSeconds. Задача, которая не влезает, режется на шаги: чекпоинт после каждого, а не один ack в конце часа работы. Про второй потолок со стороны брокера тоже помнить полезно: consumer timeout в тридцать минут из восьмого урока вернёт доставку в requeue, даже если ваш дожим готов ждать вечно.

Остановка потребителя go: дожим по шагам#

По шагам остановка потребителя go выглядит так: сначала прекращают брать новое — контекст отменён, поток подписки закрыт, и очередь копит излишки сама; затем воркеры доедают взятые доставки — range завершается на закрытом потоке, каждая доставка получает Ack или Nack; после этого WaitGroup освобождает main, и только тогда закрываются канал и соединение. Порядок закрытия критичен: Ack на закрытом канале вернёт ErrClosed, а брокер посчитает доставку неподтверждённой — та самая ложная неудача из двадцатого урока, когда работа сделана, а брокер её повторит. Отдельный вопрос — издатели: потребитель ничего о них не знает, но останавливать стоит сначала приём работы, потом обработку, иначе во время дожима очередь продолжит расти. В веб-сервисе за это отвечает readiness-проба: инстанс забирают из балансировщика, дослушивают хвост входящих запросов и лишь затем отменяют подписки. Обратный порядок даёт заметный дефект: потребителя убили, HTTP ещё жив, новые задачи попадают в очередь, которую никто не ест.

Дожим пула: cancel и общий WaitGroup

// пул из семнадцатого урока + функция остановки
func stopPool(cancel context.CancelFunc, wg *sync.WaitGroup,
        timeout time.Duration) {
        cancel() // новые доставки больше не берутся

        done := make(chan struct{})
        go func() {
                wg.Wait()
                close(done) // закрытие вместо отправки: событие без значения
        }()

        select {
        case <-done:
                log.Println("воркеры дошли до конца")
        case <-time.After(timeout):
                log.Println("дожим не уложился в", timeout)
        }
}

Функция не знает, сколько воркеров внутри: WaitGroup общий, cancel один на всех — тот же корень отмены, что и у HTTP-сервера. Закрытие канала вместо отправки значения читается как «событие случилось» и безопасно при любом числе ожидающих. Тайм-аут передаётся снаружи: у процесса он равен разнице между grace period платформы и запасом на завершение остального кода — выгрузку метрик, закрытие базы, финальные логи.

Что станет с остатками окна

Неподтверждённые доставки, которые воркер не успел доесть до смерти процесса, вручную возвращать не нужно и не получится: брокер сделает это сам при закрытии соединения — тот же requeue, что при обрыве из двадцатого урока. Остаток очереди тоже остаётся на месте: durable-очередь из девятого урока переживает рестарты, и новый инстанс после dial начнёт ровно с того места, где старый умер. В дожиме вручную делают ровно одно: подтверждают те доставки, работа по которым завершилась успешно.

Ack на выходе и проверка руками#

Проверка остановки потребителя go руками занимает минуту и показывает больше, чем любая диаграмма. Поднимите контейнер из второго урока, запустите потребителя из этого урока, накидайте задач продюсером из шестого и пошлите процессу kill -TERM: в логах появится тройка сигнал — дожим — чистый выход, в management UI подписка исчезнет, а глубина очереди покажет остаток, который доедят при следующем запуске. Затем повторите то же с kill -9: три неподтверждённые доставки вернутся в очередь с флагом Redelivered из восьмого урока, и воркер получит их заново после старта — ровно так выглядит типичный дубль, о котором честно предупреждали в девятнадцатом уроке про идемпотентность. Разница между двумя прогонами — наглядный ответ на вопрос, зачем вообще писать дожим: SIGTERM даёт чистый выход без повторов, SIGKILL — повтор работы и след в метриках redelivered.

Порядок остановки всего сервиса#

Потребитель — часть сервиса, и у остановки целого процесса есть порядок, который стоит соблюдать не по догме, а из-за последствий. Сначала сервис перестаёт принимать новую работу: readiness-проба начинает отвечать «не готов», балансировщик уводит трафик, входящих запросов больше нет. Затем останавливают HTTP-приём: сервер дорабатывает активные соединения с дедлайном. После этого отменяют контекст потребителей и дожимают их — очередь в этот момент работает амортизатором: издатели уже притихли, а излишки лежат в durable-хвосте и не пропадут. Последними закрываются зависимости: соединения с базой, клиентские пулы, метрики. Обратный порядок ломает наблюдаемость и деньги: если сначала убить потребителя при живом HTTP, новые задачи продолжат попадать в очередь, которую никто не ест, и возраст сообщений из двадцать второго урока начнёт расти прямо во время релиза — дежурный получит алерт, который ничего не говорит о реальной аварии.

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

Полный потребитель с сигналами запускается локально против брокера из второго урока: os/signal и сеть в браузерной песочнице недоступны. В песочнице открыты fmt, strings, strconv, math, sort, errors и time — механику дожима можно отрепетировать горутинами и каналами: медленная горутина вместо воркера, тайм-аут через time.After, решение select между done и таймером, как в задании ниже. Интерактивы на странице закрепят порядок фаз остановки и поведение unacked.

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

Сигнал превращается в контекст: signal.NotifyContext, дальше только ctx.Done и отмена подписки — одна шина остановки на процесс.
Дожим ограничен тайм-аутом: WaitGroup в отдельной горутине плюс select с time.After, зависшая задача не держит процесс вечно.
Тайм-аут дожима меньше grace period платформы: docker stop, systemd и Kubernetes дают конечное время, после которого приходит SIGKILL.
Порядок выхода: дожим воркеров, затем канал, затем соединение — Ack на закрытом канале возвращает ErrClosed.
Издателей останавливают раньше потребителей: сначала приём работы, потом её обработка, иначе очередь растёт во время дожима.
От SIGKILL защищает только идемпотентность обработчиков: сигнал не перехватить, unacked вернёт брокер.

Каркас урока собран из соседних глав: цикл воркера и ConsumeWithContext дали седьмой урок про потребление, пул с WaitGroup — семнадцатый про worker pool, а окна Qos — десятый про prefetch. Почему unacked вернутся сами, показал двадцатый урок про reconnect, стойкость остатков объясняет девятый про durable, а маршрут nack в retry — шестнадцатый про dead letter. Кооперативная отмена контекста осталась в девятом уроке курса по Go, readiness перед остановкой — в курсе по Prometheus. Карта курса — на странице курса.

Проверка: отмена в середине задачи

Контекст потребителя отменён, а воркер как раз обрабатывает доставку. Что произойдёт дальше?

Проверка: дожим против платформы

Тайм-аут дожима выставлен в сорок секунд, а docker stop даёт контейнеру десять. Что произойдёт при деплое с незавершёнными задачами?

Дожим или тайм-аут?

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

done := make(chan struct{})
go func() {
        time.Sleep(300 * time.Millisecond)
        close(done)
}()

select {
case <-done:
        fmt.Println("дожим завершён")
case <-time.After(100 * time.Millisecond):
        fmt.Println("тайм-аут дожима")
}

Соберите сигнальный каркас

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

ctx, stop := signal.(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() msgs, err := ch.(ctx, "tasks", "worker-1", false, false, false, false, nil) if err != nil { log.Fatal(err) } <-ctx.() log.Println("сигнал получен, дожимаем")

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

Расставьте шаги от сигнала до закрытого процесса: приём работы, отмена, дожим воркера, ожидание, соединение, финал.

Строки

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

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

Карточки: graceful shutdown

Термин

signal.NotifyContext

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

Определение

Превращает SIGINT и SIGTERM в отмену контекста: одна шина остановки для подписок, HTTP-сервера и воркеров.

1 / 6

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

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

Вывод

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

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

SIGTERM превращается в отмену контекста через signal.NotifyContext и доходит до подписки через ConsumeWithContext.
Дожим: новых доставок нет, взятые доедаются до Ack или Nack, процесс ждёт WaitGroup с тайм-аутом.
Тайм-аут дожима обязан быть заметно меньше grace period платформы: после него платформа пришлёт SIGKILL.
При SIGKILL или тайм-ауте unacked возвращает брокер: обработчики проектируются идемпотентными.
Порядок выхода: воркеры, затем канал, затем соединение; издатели останавливаются раньше потребителей.
Проверка руками: kill -TERM даёт чистый выход, kill -9 — возврат и повтор работы с Redelivered.

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

Соберите потребителя из урока с пулом из семнадцатого урока и сигнальным каркасом из этого. Замерьте три сценария против контейнера из второго урока: kill -TERM при пустой очереди — чистый выход без логов дожима; kill -TERM под нагрузкой продюсера из шестого — в логах виден дожим, а глубина очереди после рестарта совпадает с ожиданиями; kill -9 — сравните поведение unacked и флаг Redelivered в новом прогоне. Подберите тайм-аут дожима так, чтобы он укладывался в docker stop -t 15, и добавьте в конец остановки корректное закрытие HTTP-сервера с собственным дедлайном.

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

Нужно ли вручную ставить сообщения обратно в очередь при остановке?