Восстановление соединения: reconnect

Q: Почему amqp091-go не переподключается сама?

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

Q: Куда положить цикл переподключения: в main или в обёртку?

В отдельный уровень: функция run(ctx) из урока — хороший образец. В main живёт только запуск и остановка, воркеры не знают о соединении вообще. Так reconnect тестируется отдельно — подменой dial, — а остановка из двадцать первого урока просто отменяет контекст, не ломая бизнес-код.

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

Практично нет: закрытие канала по ошибке протокола — почти всегда симптом ошибки кода, например long-lived канала из двадцать третьего урока, и чинить надо причину, а не подписываться на его NotifyClose. Правило из четвёртого урока: канал дёшев, живёт со своей задачей, и при обрыве соединения пересобирается вместе с ним.

У всех примеров курса до сих пор было общее допущение: соединение установилось — и живёт. В бою оно ложно: брокер рестартует под выкладку, сетевой полис по пути закрывает простаивающие TCP-сессии, под с клиентом переезжает на другую ноду. Обрыв — не авария, а штатное событие жизненного цикла, и код обязан быть к нему готов. Важно понимать, что amqp091-go за вас переподключаться не станет: библиотека низкоуровневая, reconnect не входит в её контракт, и об этом честно написано в её README — обрыв закрывает соединение, каналы и подписки, а восстановление — забота приложения. С другой стороны, топология durable-очередей из девятого урока живёт на брокере и переживает его рестарт, так что переподключение — это заново соединение, каналы и подписки, но не заново данные. Дизайн, который стоит принять с первого дня: соединение смертно, каждая горутина, работающая с ним, должна уметь выйти из цикла при закрытии потока, а уровень выше — уметь собрать всё заново. Ниже — как этот уровень устроен.

Rabbitmq reconnect: NotifyClose как сигнал#

Начинается rabbitmq reconnect с обнаружения, и главный источник сигнала — conn.NotifyClose: метод принимает Go-канал и шлёт в него *amqp.Error, когда соединение закрывается — по инициативе брокера, из-за ошибки протокола или сетевого сбоя. Тот же метод есть у канала: ch.NotifyClose сообщает о закрытии конкретного канала, что полезно, когда каналов много, а виноват один. Второй, пассивный источник — ваши циклы: при обрыве брокер закрывает потоки подписок, range завершается, и воркер из семнадцатого урока выйдет из цикла сам; закрытие потока доставки — это и есть уведомление для горутин, которые не подписывались на NotifyClose. Третий источник — ошибки операций: Ack и Publish на мёртвом соединении возвращают ошибку, обычно amqp.ErrClosed. Важно различать два вида закрытий: вы сами вызвали conn.Close при штатной остановке — тогда NotifyClose тоже сработает, и это не повод переподключаться; наоборот, обрыв при живом желании работать — повод. Разделяйте их флагом: закрытие, инициированное shutdown из двадцать первого урока, гасит цикл переподключения, всё остальное — запускает.

Цикл переподключения с бэкоффом

package main

import (
        "context"
        "log"
        "math/rand"
        "time"

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

func backoff(attempt int) time.Duration {
        d := time.Duration(1<<uint(attempt)) * time.Second // 1, 2, 4, 8...
        if d > 30*time.Second {
                d = 30 * time.Second
        }
        // джиттер: рассыпаем попытки, чтобы пул инстансов не бился в унисон
        return d/2 + time.Duration(rand.Int63n(int64(d/2)+1))
}

func run(ctx context.Context, url string) error {
        for attempt := 0; ; attempt++ {
                conn, err := amqp.Dial(url)
                if err != nil {
                        wait := backoff(attempt)
                        log.Println("dial:", err, "— ждём", wait)
                        select {
                        case <-time.After(wait):
                                continue
                        case <-ctx.Done():
                                return ctx.Err()
                        }
                }
                attempt = 0 // соединение живо — бэкофф сбрасывается

                if err := declareTopology(conn); err != nil {
                        conn.Close()
                        return err // ошибки топологии не лечатся обрывом
                }

                closed := conn.NotifyClose(make(chan *amqp.Error, 1))
                go consumeAll(ctx, conn) // подписки, воркеры, их каналы

                select {
                case err := <-closed:
                        log.Println("соединение закрыто:", err)
                case <-ctx.Done():
                        conn.Close() // штатная остановка: без переподключения
                        return ctx.Err()
                }
                conn.Close()
        }
}

Структура вечная: dial — топология — подписки — ожидание закрытия. Бэкофф удваивает паузу с потолком в тридцать секунд и добавляет джиттер, чтобы сотня инстансов после рестарта брокера не билась о порт в одну миллисекунду; успешный dial сбрасывает счётчик. NotifyClose даёт канал с буфером на одно значение: без него сигнал уйдёт в никому не читаемый канал, если goroutine уже вышла. Контекст разделяет два выхода: обрыв перезапускает цикл, отмена — останавливает процесс.

Что теряется при обрыве#

Что	Что происходит	Кто чинит
неподтверждённые доставки	возвращаются в очередь как requeue	никто: они снова уйдут живым воркерам
сообщения в durable-очереди	остаются на брокере и переживают рестарт	ничего чинить не нужно
каналы и подписки	закрываются вместе с соединением	приложение заново после dial
публикации в полёте	не долетели — считаются не отправленными	издатель: повтор или буфер
non-durable очереди	стираются рестартом брокера	объявление топологии при reconnect

Ack на мёртвом соединении

Кейс: воркер взял доставку, пошёл в базу, за это время соединение упало — брокер вернул сообщение в очередь, а воркер вернулся из базы и вызвал d.Ack(false). Ошибка amqp.ErrClosed — и вот тут развилка. Неправильная ветка: «ack не прошёл, значит задача не сделана» — и код отвечает requeue или алертом, хотя задание выполнено, а сообщение уже обрабатывает другой воркер: получается дубль работы и ложные срабатывания. Правильная ветка: ошибка на закрытом канале или соединении означает «моя сессия мертва», воркер прекращает цикл и выходит, а восстановление — забота уровня соединения из этого урока. Отличить легко по типу ошибки: amqp.ErrClosed и ошибки вида «channel/connection is not open» — сигнал сетевой, бизнес-ошибки обработки к ним не относятся. Симметричная история у издателя: публикация в мёртвое соединение тоже возвращает ErrClosed, и без подтверждений успешный вызов Publish означает лишь «принято в буфер» — повтор при обрыве делает издатель, осознанно принимая риск дубликатов.

Переподключение amqp: топология и подписки заново#

Сам dial — половина дела; переподключение amqp обязано восстановить рабочее состояние, и тут помогает дисциплина топологии из пятого урока: объявления идемпотентны, поэтому declareTopology вызывает те же QueueDeclare, ExchangeDeclare и QueueBind с теми же аргументами — на живом брокере это no-op, на свежем после сбоя — восстановление. Порядок сборки прежний: соединение, каналы, окна Qos, подписки с теми же тегами, воркеры из семнадцатого урока. Один нюанс: у воркеров, переживших старое соединение, range по старым потокам уже завершён — они вышли из циклов, и новые подписки запускают новые горутины; не пытайтесь переиспользовать старые Delivery-каналы, они закрыты. Тайм-аут на dial и объявление стоит задавать отдельно от бэкоффа: зависший TCP-хендшейк без тайм-аута подвесит цикл на минуты. И держите reconnect-уровень отдельно от бизнес-кода: воркеры не знают о соединении, уровень управляет жизнью, а метрики переподключений — счётчик рестартов и время вне связи — покажут, насколько брокер вам вообще стабилен.

Blocked-уведомления и NotifyReturn

// брокер приостановил публикации: память или диск превысили алармы
blocked := conn.NotifyBlocked(make(chan bool, 1))
go func() {
        for b := range blocked {
                if b {
                        log.Println("брокер заблокировал публикации — экономим буферы")
                } else {
                        log.Println("публикации разблокированы")
                }
        }
}()

// unroutable-сообщения возвращаются издателю (публикация с mandatory)
returned := conn.NotifyReturn(make(chan amqp.Return, 1))
go func() {
        for r := range returned {
                log.Printf("вернулось без маршрута: key=%s body=%s",
                        r.RoutingKey, r.Body)
        }
}()

NotifyBlocked шлёт true, когда брокер попал под аларм памяти или диска и перестал принимать публикации, и false при разблокировке: сигнализация здесь экономит часы расследований «куда делись сообщения». NotifyReturn ловит basic.return от публикаций с флагом mandatory из шестого урока: сообщение не нашло маршрута и вернулось издателю — их считают отдельно, потому что каждое такое возвращение означает ошибку маршрутизации, а не сбой сети.

Heartbeat: детектор полусмертных соединений

Не каждый обрыв приходит с FIN-пакетом: кабель выдернули, NAT умер, файрвол молча выкинул сессию — TCP выглядит живым, а данных нет, и range будет висеть, пока его не разбудит часами простоя. Для этого в протоколе есть heartbeat: обе стороны обмениваются кадрами с заданным периодом, и молчание дольше порога закрывает соединение — старый пассивный обрыв превращается в активное закрытие и попадает в NotifyClose. В amqp091-go период задаётся полем Heartbeat в amqp.Config при DialConfig; дефолтное значение библиотеки — десять секунд, и его стоит переопределять осознанно: слишком частый пульс тратит CPU и трафик, слишком редкий затягивает обнаружение. Heartbeat — не замена циклу переподключения, а его источник сигналов: чем короче пульс, тем быстрее срабатывает NotifyClose после реального обрыва.

Проверка reconnect: хаос по расписанию#

Цикл переподключения, который ни разу не срабатывал до продакшена, работает в продакшене плохо — он не отлажен. Поэтому reconnect проверяют как фичу: на стендах и в CI брокер рестартуют по расписанию, между воркером и брокером ставят прокси, умеющий рвать соединения, и сверяют поведение с ожиданиями — воркер восстановился, повторная обработка не удвоила побочных эффектов, лог понятен дежурному. Метрики делают проверку наглядной: счётчик переподключений, секунды вне связи, время до первой успешной публикации после сбоя. На них же видна деградация: рост пауз бэкоффа означает, что брокер недоступен дольше, а не что код сломан.

Blocked, Return и договорённости с издателем#

Два уведомления из кода выше — часть общего правила: соединение сообщает о состоянии, если его об этом спросить. NotifyBlocked сигнализирует, что брокер перестал принимать публикации из-за алармов памяти или диска: в этот момент издатели обязаны не пытаться выдавить сообщения в заблокированный сокет, а копить их у себя или временно приостанавливать; издавать в заблокированного брокера — значит рисковать потерей буферов при обрыве. NotifyReturn относится к отдельному механизму: публикации с флагом mandatory из шестого урока возвращаются отправителю, если маршрут не найден, и через conn.NotifyReturn приложение получает их для разбора — лог, метрика, парковка на месте. Оба сигнала переживают reconnect-логику только до обрыва: каналы уведомлений привязаны к соединению, и после пересборки их создают заново, поэтому регистрируют их в declare-фазе цикла, а не в main. Договорённость с издателем простая: включить mandatory на важных маршрутах, слушать returns, считать их метрикой качества топологии — рост возвратов почти всегда значит, что кто-то удалил привязку или переименовал ключ.

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

Reconnect проверяется жестоко и весело: поднимите контейнер из второго урока, запустите воркер из урока и рестартуйте контейнер — docker restart; воркер обязан восстановиться сам, а лог покажет пару закрытие-бэкофф-dial. В песочнице сети нет — там доступны fmt, strings, strconv, math, sort, errors и time, но арифметику бэкоффа и джиттера легко отрепетировать программой, а разброс попыток — оценить через sort по срезу длительностей. Интерактивы ниже закрепят порядок восстановления и поведение unacked.

Правила переподключения#

Соединение смертно по умолчанию: цикл dial — топология — подписки — ожидание закрытия пишут с первого дня.
NotifyClose отличает штатную остановку от обрыва: флаг shutdown гасит переподключение.
Бэкофф с потолком и джиттером; успешный dial сбрасывает счётчик попыток.
После reconnect заново объявляют топологию и подписки: старые Delivery-каналы уже не оживут.
Ошибка Ack или Publish вида ErrClosed — сигнал смерти сессии, а не бизнес-неудача.
Heartbeat обнаруживает тихие обрывы; NotifyBlocked и NotifyReturn говорят о состоянии между ними.

Опора урока: четвёртый урок про connection и channel объяснил, что и как закрывается, пятый про QueueDeclare — идемпотентность объявлений, на которой держится восстановление, а шестой про публикацию — mandatory и возвращённые сообщения. Данные, которые переживают обрыв, разобраны в девятом уроке про durable, пул воркеров, который пересобирается после reconnect, — в семнадцатом, штатная остановка, гасящая переподключение, — в двадцать первом. Карта курса — на странице курса.

Проверка: unacked при обрыве

У воркера были две неподтверждённые доставки, когда соединение оборвалось. Что с ними станет?

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

После рестарта брокера сто инстансов сервиса одновременно обнаруживают обрыв и начинают переподключаться. Что добавляет джиттер к бэкоффу?

Какие паузы выдаст бэкофф?

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

func backoff(attempt int) time.Duration {
        d := time.Duration(1<<uint(attempt)) * time.Second
        if d > 10*time.Second {
                d = 10 * time.Second
        }
        return d // джиттер отключили для наглядности
}

for i := 0; i < 5; i++ {
        fmt.Print(backoff(i), " ")
}

Соберите каркас reconnect

Допишите подписку на закрытие, переподключение и паузу бэкоффа.

closed := conn.(make(chan *amqp.Error, 1)) if err := <-closed; err != nil { log.Println("обрыв:", err) } time.(backoff(2)) conn, err = amqp.(url) if err != nil { log.Fatal(err) }

Соберите цикл восстановления

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

Строки

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

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

Карточки: reconnect

Термин

conn.NotifyClose

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

Определение

Канал, в который приходит *amqp.Error при закрытии соединения. Источник сигнала для цикла переподключения.

1 / 6

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

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

Вывод

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

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

Обрыв — штатное событие: amqp091-go не переподключается сама, цикл восстановления пишет приложение.
NotifyClose даёт сигнал закрытия; флаг shutdown отличает штатную остановку от сбоя.
Цикл reconnect: dial с бэкоффом и джиттером — идемпотентная топология — новые подписки.
Старые Delivery-каналы после обрыва закрыты: воркеры выходят, новые запускаются после dial.
ErrClosed в Ack или Publish — смерть сессии, а не бизнес-неудача; unacked возвращаются брокером.
Heartbeat превращает тихий обрыв в активное закрытие; период переопределяется в amqp.Config.
NotifyBlocked и NotifyReturn сообщают о блокировках публикаций и потерянных маршрутах между обрывами.

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

Соберите цикл из урока с бэкоффом и джиттером и воркером из семнадцатого. Запустите против контейнера из второго урока и проверьте три сценария: docker restart брокера — воркер восстановился сам; остановка контейнера на минуту — паузы бэкоффа видны в логах и потолок не превышен; штатный сигнал остановки — переподключения нет. Добавьте счётчик переподключений и метрику времени вне связи, затем включите heartbeat в amqp.Config и сравните, насколько быстрее обнаруживается тихий обрыв при выдернутом сетевом интерфейсе контейнера.

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

Почему amqp091-go не переподключается сама?