RPC над очередями: reply-to и correlation id

Весь курс строился вокруг идеи «отправил и пошёл дальше»: заказ положен в очередь, уведомления уйдут, когда дойдут. Но в каждом проекте есть вызовы с возвратом: приёмке заказа нужно знать остаток на складе, загрузке отчёта — идентификатор файла, бэкенду телефона — цену тарифа. HTTP-клиент решает это синхронным запросом, и там, где ответ нужен мгновенно, так и стоит делать. Очередь берёт на себя роль транспорта в трёх ситуациях: доступ к обработчику идёт через границы, где прямая сеть неудобна, — другой кластер, партнёрская интеграция; нужна балансировка пула обработчиков брокером без своего балансировщика; запрос не обязан умирать вместе с клиентом, и переживать его рестарт — полезное свойство. RPC над очередями стоит дороже синхронного вызова: два сообщения вместо одного запроса, отдельная очередь на клиента, совпадение по идентификатору. В этом уроке — конверт из ReplyTo и CorrelationId, пара клиент-сервер на amqp091-go и честный разбор, где async-поток из двенадцатого и тринадцатого уроков честнее.

Rpc очередь go: ReplyTo и CorrelationId#

Поля конверта заложены в сам протокол, и rpc очередь go собирается без единой кастомной структуры. Клиент перед публикацией открывает свою очередь ответов — обычно анонимную: QueueDeclare с пустым именем даёт брокерское имя вида amq.gen- и свойства exclusive с auto-delete, чтобы очередь жила ровно столько, сколько жив процесс клиента. В Publishing он ставит два поля: ReplyTo — имя этой очереди, CorrelationId — идентификатор запроса, сгенерированный на месте. Сервер читает запрос, считает ответ и публикует его в дефолтный обменник из одиннадцатого урока с ключом маршрутизации, равным ReplyTo, — дефолтный обменник доставляет сообщение в очередь с этим именем напрямую. CorrelationId сервер обязан скопировать из запроса в ответ: именно по нему клиент отличит ответ на свой запрос от чужого, ведь если клиентов много, очередь ответов у каждого своя, а в сценарии общих очередей идентификатор становится единственной ниточкой. Никаких новых полей в заголовках для этого не нужно: вы ничего не теряете от совместимости, а инструменты брокера видят обычные свойства сообщения.

Клиент RPC: запрос, ожидание, таймаут

package main

import (
        "context"
        "fmt"
        "log"
        "time"

        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)
        }

        // анонимная exclusive-очередь ответов
        q, err := ch.QueueDeclare("", false, false, true, false, nil)
        if err != nil {
                log.Fatal(err)
        }

        corrID := fmt.Sprintf("req-%d", time.Now().UnixNano())
        ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
        defer cancel()

        err = ch.PublishWithContext(ctx, "", "rpc.balance", false, false,
                amqp.Publishing{
                        ContentType:   "application/json",
                        CorrelationId: corrID,
                        ReplyTo:       q.Name,
                        Body:          []byte("{"account":"acc-77"}"),
                })
        if err != nil {
                log.Fatal(err)
        }

        msgs, err := ch.ConsumeWithContext(ctx, q.Name, "",
                true, false, false, false, nil)
        if err != nil {
                log.Fatal(err)
        }

        for d := range msgs {
                if d.CorrelationId == corrID {
                        log.Println("ответ:", string(d.Body))
                        return
                }
                log.Println("чужой ответ, пропускаю:", d.CorrelationId)
        }
        log.Println("таймаут: ответ не пришёл")
}

Очередь ответов объявлена анонимной и exclusive: она умрёт вместе с клиентом, и второй клиент её не увидит. CorrelationId генерируется из наносекунд времени — в демо этого достаточно, в бою берут UUID; тот же идентификатор клиент кладёт в запрос и ищет в ответах. Таймаут держит и публикацию, и подписку: на исходе контекста поток ответов закрывается, range завершается, и клиент уходит по ветке «ответ не пришёл» вместо вечного висения.

Части RPC-конверта#

Элемент	Роль	Кто задаёт
ReplyTo	имя очереди, куда сервер публикует ответ	клиент в каждом запросе
CorrelationId	связка запроса и ответа, фильтр чужих ответов	клиент, сервер копирует в ответ
очередь запросов	вход сервера, точка балансировки реплик	сервер, durable по политике
очередь ответов	анонимная exclusive-очередь клиента	клиент, живёт с процессом
таймаут	граница ожидания: публикация и подписка под контекстом	клиент по SLO обработчика

Ответы-сироты и тайм-ауты

Кейс: клиент обработал таймаут и ушёл с ошибкой, а сервер через минуту всё-таки посчитал и опубликовал ответ — в очередь, которую больше никто не читает, до её auto-delete. Ничего не сломалось, если клиент не делает повторов; но с повтором возникает уже знакомая тема из восьмого урока: сервер может обработать одну и ту же заявку дважды, потому что первый ответ клиент не дождался. Выводы честные: обработчик RPC делайте идемпотентным, а повторы запросов ограничивайте таймаутом и счётчиком попыток. Вторая ловушка — горутинная утечка: если клиент ждёт ответ в горутине с бесконечным range и без контекста, каждый таймаут оставляет после себя висящую горутину и живую подписку. Третья — общий фильтр: ответ с чужим CorrelationId означает протоколовую ошибку — его логируют и пропускают, а не отдают вызывающему.

Сервер RPC: ответ до ack#

Порядок действий на сервере определяет, что случится при его падении посреди обработки, поэтому rpc очередь go живёт по правилу «ответ публикуется до подтверждения». Воркер читает доставку из очереди запросов, выполняет бизнес-логику, публикует ответ в ReplyTo с CorrelationId из запроса и только затем делает d.Ack(false). Если процесс умирает после публикации, но до ack, запрос вернётся в очередь и его обработают заново — клиент получит два одинаковых ответа и отбросит второй по идентификатору; потерянного состояния нет. Если поменять порядок и подтвердить до публикации, падение между ack и публикацией потеряет ответ безвозвратно: запрос подтверждён, клиент висит до таймаута без шансов. Очередь запросов объявляет сервер, и durable она должна быть, если запросы обязаны пережить рестарт брокера; для краткожных расчётных запросов допустимы non-durable очереди — цена известна: рестарт брокера стирает невыполненные запросы, и клиенты уходят по таймауту. Prefetch на сервере ставится по тем же правилам десятого урока: тяжёлые вычисления — окно-единица.

Сервер RPC: обработка и ответ

for d := range msgs {
        resp, err := balance(d.Body)
        if err != nil {
                // неразрешимая ошибка: ответ с пометкой, ack
                resp = []byte("{"error":"unavailable"}")
        }

        ctx, cancel := context.WithTimeout(context.Background(), time.Second)
        err = ch.PublishWithContext(ctx, "", d.ReplyTo, false, false,
                amqp.Publishing{
                        ContentType:   "application/json",
                        CorrelationId: d.CorrelationId,
                        Body:          resp,
                })
        cancel()
        if err != nil {
                log.Println("ответ не ушёл:", err)
                d.Nack(false, true) // вернём в очередь для повторной попытки
                continue
        }
        d.Ack(false)
}

Дефолтный обменник с ключом d.ReplyTo — это и есть доставка в очередь ответов конкретного клиента. CorrelationId копируется из запроса без изменений. Неразрешимая ошибка не молчит: клиент получает явный ответ с error и сам решает, что делать; Nack с requeue=true оставлен только для сбоев публикации — технических, повтор которых имеет смысл.

Direct reply-to: без очереди на каждый запрос

RabbitMQ умеет обходиться без анонимных очередей: псевдоочередь amq.rabbitmq.reply-to используется так — клиент один раз подписывается с autoAck=true и в каждом запросе ставит ReplyTo в значение amq.rabbitmq.reply-to; брокер направляет ответ подписчику напрямую, не создавая настоящую очередь. Экономия: никаких QueueDeclare и подписок на каждый вызов, меньше объектов в брокере. Цена: работает только при autoAck=true и одном потребителе на канал, подтверждение доставки не требуется по построению. Проверьте поддержку на своём брокере и не смешивайте два механизма в одном сервисе без причины.

Много запросов в полёте: карта ожиданий#

Клиент из примера ждёт один ответ, а реальный сервис шлёт десятки запросов одновременно, и здесь начинается устройство поверх rpc. Один потребитель на очереди ответов читает доставки и раскладывает их по карте: ключ — CorrelationId, значение — Go-канал конкретного ждущего вызова; вызов публикует запрос, регистрирует канал в карте и читает из своего канала с тайм-аутом. Карту защищают мьютексом: подписчик и вызовы живут в разных горутинах. Просроченные записи по тайм-ауту удаляют, иначе карта растёт без предела на недошедших ответах. Бонус схемы — переиспользование: одна очередь ответов и одна подписка на процесс вместо пары объявлений на каждый вызов, но с собственным фильтром. Здесь rpc на очередях догоняет по удобству обычные HTTP-клиенты — и подходит к границе, за которой проще взять готовую библиотеку, чем поддерживать карту самому.

Rabbitmq rpc: где асинхронность честнее#

Паттерн rabbitmq rpc не универсален, и главные ошибки вокруг него — применение там, где ответ по сути никому не нужен. Если звонок логирует факт, а не спрашивает значение, — это событие, и fanout или topic из двенадцатого и тринадцатого уроков выразят его без ложной синхронности. Если задача длинная — минута и больше, — синхронное ожидание связывает клиента и ломает таймауты: честнее принять запрос, вернуть идентификатор сразу и уведомить о готовности событием, а статус отдать отдельным запросом. Если ответ нужен, но его допустимо получить с задержкой в секунды, схема «поставить задачу и проверить статус» дешевле RPC в инцидентах: у неё нет зависших клиентов, потому что никто не ждёт в очереди ответов. Остаются настоящие кейсы RPC: короткие расчёты, балансировка пула обработчиков, интеграции через границы сетей. Свойство очереди, которое тут работает на вас: несколько реплик сервера на одной очереди запросов делят нагрузку автоматически, и отказ одной реплики просто переносит её запросы живым.

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

RPC проверяется только на живом брокере: поднимите контейнер из второго урока, запустите сервер и два клиента и посмотрите в management UI, как появляются и умирают анонимные очереди ответов. В песочнице сети нет — там доступны fmt, strings, strconv, math, sort, errors и time; пару запрос-ответ можно разыграть горутинами и Go-каналом: одна горутина пишет корреляции, вторая возвращает ответы. Интерактивы ниже закрепят поля конверта и порядок ответ-до-ack.

Правила RPC над очередями#

ReplyTo — имя очереди ответов клиента, CorrelationId — связка пары: оба поля стандартные, кастомные заголовки не нужны.
Очередь ответов — анонимная и exclusive: рождается и умирает с процессом клиента.
Сервер публикует ответ в дефолтный обменник с ключом ReplyTo и только затем подтверждает запрос.
Клиент фильтрует ответы по CorrelationId: чужие и опоздавшие ответы логируют и пропускают.
Ожидание держит контекст с таймаутом: бесконечный range без контекста оставляет висящие горутины.
Длинные задачи и уведомления — не RPC: идентификатор сразу, готовность событием, статус отдельным запросом.

Материал опирается на соседей: шестой урок про PublishWithContext дал поля Publishing, среди которых ReplyTo и CorrelationId, седьмой про потребление — цикл сервера, восьмой про ack — порядок подтверждений. Анонимные и exclusive-очереди остались в пятом уроке, события вместо RPC — в двенадцатом и тринадцатом уроках про fanout и topic. Остановку сервера RPC с дожимом разбирает двадцать первый урок. Карта курса — на странице курса.

Проверка: маршрут ответа

Сервер получил запрос с ReplyTo=amq.gen-AbCd и CorrelationId=req-42. Как он должен опубликовать ответ?

Проверка: порядок на сервере

Сервер обработал запрос и подтвердил его d.Ack(false), а публикацию ответа запланировал сразу после — но процесс упал между этими шагами. Что произойдёт?

Каким будет имя очереди?

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

q, err := ch.QueueDeclare("", false, false, true, false, nil)
if err != nil {
        log.Fatal(err)
}
fmt.Println(len(q.Name) > 0, strings.HasPrefix(q.Name, "amq.gen-"))

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

Допишите поля Publishing и подтверждение на сервере.

err = ch.PublishWithContext(ctx, "", "rpc.balance", false, false, amqp.Publishing{ CorrelationId: corrID, : q.Name, Body: []byte(req), }) // на сервере, после обработки: err = ch.PublishWithContext(ctx, "", , false, false, amqp.Publishing{ CorrelationId: , Body: resp, }) d.(false)

Соберите цепочку клиента RPC

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

Строки

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

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

Карточки: RPC на очередях

Термин

ReplyTo

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

Определение

Поле Publishing с именем очереди ответов. Сервер публикует в дефолтный обменник с этим именем как ключом.

1 / 6

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

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

Вывод

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

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

RPC над очередями строится на стандартных полях: ReplyTo — очередь ответов, CorrelationId — связка пары.
Очередь ответов клиента — анонимная и exclusive: имя amq.gen-, живёт с процессом.
Сервер публикует ответ в дефолтный обменник с ключом ReplyTo и подтверждает запрос после публикации.
Клиент фильтрует ответы по CorrelationId: чужие и опоздавшие логируют и пропускают.
Ожидание живёт под контекстом с таймаутом; обработчик сервера — идемпотентный, повторы дают дубли ответов.
Direct reply-to amq.rabbitmq.reply-to убирает анонимные очереди ценой autoAck=true.
Факты, длинные задачи и уведомления — не RPC: события и статусы честнее.

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

Соберите пару из урока: сервер на очереди rpc.balance с ответом до ack и клиент с анонимной очередью, CorrelationId и таймаутом в три секунды. Проверьте фильтр: отправьте два запроса с разными идентификаторами и убедитесь, что каждый клиент берёт только свой ответ. Остановите сервер между публикацией ответа и ack вручную и проследите повторную обработку запроса и второй ответ с тем же CorrelationId. В конце замените анонимную очередь на direct reply-to и запустите тот же сценарий, отметив в комментариях, что изменилось в подписке.

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

Нужна ли durable-очередь запросов?

Зависит от контракта: если незавершённые запросы обязаны пережить рестарт брокера — durable вместе с persistent-сообщениями из девятого урока. Для краткожных расчётных RPC допустима non-durable очередь: после рестарта брокера клиенты уйдут по таймауту и повторят запрос, а потерянных данных нет, потому что ответы не были выданы.

Как выбрать таймаут RPC-запроса?

От SLO обработчика с запасом на две сетевые стороны и очередь: если сервер в среднем отвечает за сто миллисекунд, разумно полторы-две секунды. Таймаут меньше хвостовой задержки обработчика даёт ложные ошибки на пиках, а больше — зависших клиентов. Тюнить стоит по измерениям, а не по ощущениям, и держать таймаут и в публикации, и в подписке, как в коде урока.

Что делать с повторами запросов?

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

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

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