Connection и Channel: модель соединений

Сервис вырос: пять HTTP-хендлеров публикуют события, три воркера потребляют задачи, фоновый чистильщик время от времени что-то удаляет. Все они хотят говорить с брокером одновременно — и тут возникает вопрос архитектуры соединений. Открывать каждому своё TCP-соединение расточительно: рукопожатие, аутентификация, согласование параметров, память на обеих сторонах. Делить один поток всем подряд опасно: кадры разных задач перемешаются. AMQP 0-9-1 решает это двумя уровнями: соединение — дорогое, единичное, а каналы поверх него — дешёвые и многочисленные. В первом уроке мы уже видели conn := amqp.Dial и ch := conn.Channel(); теперь разберём, что именно разделяют эти два вызова, где проходят лимиты и какая модель использования не развалится под нагрузкой. Речь не про красоту кода: перепутанные уровни дают гонки, которые в тестах не воспроизводятся, а на проде закрывают каналы брокером. К концу урока у вас будет чек-лист раскладки соединений для своего сервиса и понимание, за чем следит management UI на вкладке Channels.

Connection и Channel: модель соединений#

Connection — это одно TCP-соединение с брокером, установленное после рукопожатия из прошлого урока: с аутентификацией, согласованным размером кадра и heartbeat. Цена высокая — и в миллисекундах установки, и в ресурсах: у брокера на каждое соединение живёт процесс-читатель. Channel — логическая сессия поверх соединения, лёгкая как структура данных: через неё идут все рабочие команды — объявления очередей, публикации, подписки, подтверждения. Мультиплексирование работает так: кадры разных каналов вперемешку летят по одному TCP-потоку, а брокер и клиент раскладывают их по номерам каналов. Поэтому rabbitmq connection channel читают как иерархию: одно соединение на процесс, десятки каналов поверх, каждый канал обслуживает свою задачу. Открыть канал — это вызов conn.Channel() без сетевого обмена, закрыть — тоже дёшево. Отсюда правило: соединения копят годами и переиспользуют, каналы создают под задачу и закрывают, не жалея.

Одно соединение, два канала

conn, err := amqp.Dial(url)
if err != nil {
        log.Fatal(err)
}
defer conn.Close()

pub, err := conn.Channel()
if err != nil {
        log.Fatal(err)
}

sub, err := conn.Channel()
if err != nil {
        log.Fatal(err)
}

// pub публикует, sub потребляет — задачи не пересекаются

Два вызова conn.Channel() дают два независимых канала поверх одного TCP-соединения: pub для публикаций и sub для потребления. Каждый канал ведёт свой журнал неподтверждённых сообщений и своё состояние, поэтому ошибка в одном не тянет за собой другой. Закрытие соединения в defer закроет и оба канала — отдельно закрывать их в main не обязательно, но в долгоживущем сервисе канал закрывают явно, когда задача закончилась.

Почему один amqp канал на горутину#

Канал в amqp091-go не потокобезопасен, и это не недоработка, а отражение протокола: порядок кадров внутри канала важен, брокер обрабатывает их строго по очереди. Горутина, публикующая сообщение, и горутина, подтверждающая доставку, могут одновременно писать в один канал — кадры перемешаются, и брокер получит последовательность, которой не существует. Чаще всего это выглядит как внезапное закрытие канала с ошибкой протокола на ровном месте, без внятного стека. Правило из документации простое: один amqp канал на горутину, а обмен данными — через свои каналы Go или очереди задач. Если горутин много, канал создают в горутине и держат при ней; если горутина короткоживущая, канал берут из ограниченного набора. Читателям урока про горутины схема знакома: шарить изменяемое состояние опасно, и здесь опасность лишь переносится на уровень протокола. Мьютекс вокруг канала технически работает, но сериализует всё — быстрее поднять второй канал.

Общий канал из горутин: гонка, которую не видно в тесте

Кейс: воркер обрабатывает сообщения и публикует результаты, а всё через один канал, потому что «так проще». Десять сообщений в секунду — работает месяц. Под нагрузкой брокер закрывает канал ошибкой протокола, все обработки и подтверждения падают разом, а локальный тест с двумя сообщениями гонку не воспроизводит никогда. Диагноз ставится по management UI: вкладка Channels показывает каналы в состоянии закрытия с exception. Лечение одно — канал на горутину; общий канал допустим только под мьютексом и только для редких вызовов.

Лимиты: channel_max и heartbeat#

Параметры обоих уровней согласуются при рукопожатии. Сервер предлагает channel_max — максимальное число каналов на соединение; по умолчанию свежие версии RabbitMQ дают 2047, клиент может попросить меньше, и упереться в потолок можно только экзотикой вроде тысяч каналов в одном процессе. Heartbeat — интервал, которым стороны проверяют живость соединения: клиент amqp091-go по умолчанию предлагает десять секунд, и если данные по соединению не идут, стороны обмениваются пустыми кадрами; соединение без обмена и без heartbeat брокер закроет как мёртвое. Именно heartbeat объясняет, почему «спящий» продюсер иногда обнаруживает своё соединение закрытым после долгой паузы: сеть могла умереть, а клиент узнал об этом по пропущенным пульсам. Размер кадра frame_max ограничивает, сколько байт сообщения уходит одним куском, — для больших полезных нагрузок это скорее повод задуматься о размере сообщений, чем параметр для тюнинга. Все три числа живут в amqp.Config, который мы использовали через DefaultDial.

Connection против Channel#

Характеристика	Connection	Channel
Уровень	TCP-соединение с брокером	логическая сессия поверх соединения
Цена открытия	рукопожатие, аутентификация, процесс у брокера	вызов без сетевого обмена
Сколько держать	один на процесс, переиспользуют годами	под задачу или горутину, закрывают смело
Потокобезопасность	безопасен, каналы открывают из любой горутины	не потокобезопасен: один канал — одна горутина
Падение	ломает все каналы поверх	закрывает только свою сессию

NotifyClose: узнавать о закрытии

closeCh := conn.NotifyClose(make(chan *amqp.Error, 1))

select {
case err := <-closeCh:
        log.Println("соединение закрыто брокером:", err)
case <-time.After(time.Minute):
        log.Println("минута прошла спокойно")
}

NotifyClose у соединения и у канала возвращает Go-канал, в который приходит причина закрытия. Это способ узнать о проблемах без опроса: потребитель ждёт на select и при закрытии перезапускает цикл подписки или соединение целиком — с этим кодом вернёмся в уроке про reconnect. Канал делают буферизованным на единицу, как рекомендует документация библиотеки, чтобы уведомление не потерялось, пока ваша горутина занята.

Ошибки канала: коды протокола и что они значат#

Ошибки канала приходят не строками, а amqp.Error с полями: числовой код протокола, человекочитаемая причина и флаг Recover — можно ли продолжать работу на этом соединении. Три кода стоит знать наизусть. PRECONDITION_FAILED с кодом 406 — вы объявили очередь с флагами, которые не совпадают с существующей; канал при этом закрывается брокером, и это главный герой следующего урока. NOT_FOUND, код 404 — обращение к несуществующей очереди или обменнику: например, публикация в очередь, которую никто не объявил. CHANNEL_ERROR, код 504 — недопустимая последовательность кадров, та самая гонка из общего канала. Логируйте код и причину, а не только «что-то сломалось»: по коду мгновенно видно, чей это баг — прикладной логики, схемы или конкурентности. И не путайте эти ошибки с нулевой ошибкой в публикациях: молчаливая потеря сообщения не закрывает канал и не возвращает ошибку — про этот отдельный разговор будет урок про PublishWithContext.

Сколько соединений нужно приложению#

Базовая раскладка зрелого сервиса: одно соединение на процесс, каналы по потребностям — отдельный канал для публикатора, по каналу на воркера потребления, служебный канал для разовых задач. Когда одного соединения мало? Первый случай — блокировки: если брокер приостановит соединение из-за нехватки места на диске, встают все каналы, и критичные публикации отделяют в собственное соединение. Второй — приоритеты: тяжёлые выгрузки и быстрые пользовательские события разводят по разным соединениям, чтобы по таймаутам и очередям кадров они не мешали друг другу. Третий — разные учётные данные или vhost: соединение привязано к одной паре логин-vhost, перескочить нельзя, нужен второй Dial. Но не путайте «мало» с «хочется побыстрее»: десяток соединений из каждой горутины — путь к исчерпанию лимитов и ловле throttling на брокере, о чём подробно скажем в уроке про антипаттерны.

Код урока запускается локально

Брокер по-прежнему живёт в Docker, Go-код запускается через go run на вашей машине: песочница сайта умеет только подмножество стандартной библиотеки и не держит TCP-соединений, поэтому conn.Channel() в браузере не выполнить. Интерактивы ниже прорабатывают модель соединений на уровне сигнатур и порядка вызовов — то, что чаще всего и путают. Если под рукой нет брокера, даже чтение вкладки Channels на живом контейнере с двумя-тремя каналами даст больше, чем перечитывание статей.

Правила соединений и каналов#

Одно соединение на процесс: открывают при старте, переиспользуют, закрывают при остановке.
Канал — на горутину: шарить канал между горутинами без мьютекса нельзя.
Каналы дешёвые: под каждую независимую задачу — свой канал.
Публикации и потребления разводят по разным каналам.
NotifyClose у соединения и каналов — источник правды о закрытии.
Heartbeat держат включённым: мёртвое соединение должно умирать быстро.
Гонку общего канала тест на два сообщения не ловит: нужен нагрузочный прогон.

Модель соединений ясна, и на ней уже видно, почему половина багов новичков — это не про очереди, а про уровни: канал, который делится между горутинами; соединение, которое открывают на каждый вызов; NotifyClose, который никто не слушает. Дальше по плану — первый объект уровня канала: в уроке про очередь объявим её через QueueDeclare и разберём флаги, от которых зависит, переживёт ли очередь рестарт брокера. Если хочется сразу увидеть сообщения в management UI, откройте вкладку Connections после запуска кода урока: там будет ваше соединение и два канала с именами задач, которые вы им доверили. Домашнее задание ниже закрывает круг: соединение, два канала, живой UI — на этой раскладке стоит весь оставшийся курс.

Проверка: цена открытия

Что дешевле открыть: новое соединение или новый канал?

Проверка: канал и горутины

Две горутины публикуют сообщения через один канал без синхронизации. Что хуже всего может случиться?

Что выведет программа?

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

ch, err := conn.Channel()
if err != nil {
        log.Fatal(err)
}

ch.Close()

err = ch.PublishWithContext(context.Background(),
        "", "tasks", false, false,
        amqp.Publishing{Body: []byte("hi")})

fmt.Println("ошибка:", err)

Соберите монитор закрытия канала

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

closeCh := ch.(make(chan *amqp., 1)) if err != nil { log.Fatal(err) } defer conn.()

Соберите настройку двух каналов

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

Строки

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

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

Карточки: модель соединений

Термин

Connection

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

Определение

TCP-соединение с брокером после полного рукопожатия. Тяжёлое: одно на процесс, переиспользуют годами.

1 / 6

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

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

Вывод

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

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

Connection — тяжёлое TCP-соединение, Channel — лёгкая сессия поверх; иерархия держит весь обмен.
Один connection на процесс, каналы — по задачам и горутинам.
Канал не потокобезопасен: шарить его между горутинами — путь к закрытию с ошибкой протокола.
channel_max, frame_max и heartbeat согласуются при рукопожатии и живут в amqp.Config.
NotifyClose — способ узнать о закрытии без опроса; канал уведомлений делают буферизованным.
Падение соединения роняет все каналы, падение канала — только свою сессию.

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

Возьмите клиент из прошлого урока и доведите до двух каналов: pub и sub. Проверьте в management UI, что соединение одно, а каналов два, затем закройте pub из кода и убедитесь, что sub продолжает работать, а NotifyClose pub-канала вернул причину закрытия.

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

Можно ли открывать канал на каждый HTTP-запрос?

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

Почему соединение закрылось само после долгой паузы?

Скорее всего сработал heartbeat: стороны перестали обмениваться кадрами, брокер счёл соединение мёртвым и закрыл его. Проверьте сеть и файрвол между процессами; в коде такое закрытие ловится через NotifyClose и обрабатывается переподключением, а не паникой.

Мьютекс вокруг канала — это нормально?

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

Термины урока

Ключевые понятия из этого урока — короткие определения в глоссарии. Наведите на термин, чтобы вспомнить, откройте карточку для подробностей и ссылки на практику.