Перейти к основному содержимому
Нормативная версия документации - английская

Эта страница переведена с английского языковой моделью. При любом расхождении верен английский оригинал.

Запрос и ответ

Это руководство показывает, как один сервис вызывает другой через RabbitMQ и ждёт ответа.

Сервис заказов, прежде чем принять заказ, спрашивает у сервиса склада, есть ли товар в наличии. Вызов идёт через direct reply-to RabbitMQ: объявлять очередь ответов не нужно, и вести вручную таблицу ожидающих вызовов тоже не нужно.

Отвечающая сторона​

src/main.rs
/// The stock of one SKU. An RPC reply goes where the request asked, so the type names no
/// destination.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Stock {
    sku: String,
    available: bool,
}

#[subscriber("inventory.check", reply("inventory.check.unanswered"))]
async fn check_stock(request: &CheckStock) -> Stock {
    tracing::info!(request.sku, request.quantity, "stock checked");
    Stock {
        sku: request.sku.clone(),
        available: request.quantity <= 10,
    }
}

/// The inventory service's routes: each reply goes to the requester's own address.
fn inventory() -> Router<LapinBroker, impl RouterDef<LapinBroker>> {
    Router::new()
        .include(check_stock)
        .out_reply(Publish::default())
        .transform(DirectReplyTo)
        .build()
}

Сервис склада отвечает обычным обработчиком, который возвращает ответ. DirectReplyTo в издателе ответов отправляет каждый ответ по адресу, который запрос указал в свойстве reply-to, и копирует в ответ correlation-id запроса. Stock не называет адреса назначения, потому что каждый ответ уходит туда, куда попросил его запрос. Имя в reply("inventory.check.unanswered") - это адрес ответа на запрос без reply-to, и именно его показывает документ AsyncAPI сервиса.

Маршруты собраны в Router, поэтому сервис склада монтирует их в собственном приложении через b.include_router(inventory()). Пример запускает оба сервиса в одном процессе и монтирует оба роутера.

Запрашивающая сторона​

src/main.rs
/// How long an order waits for the inventory service.
const RPC_TIMEOUT: Duration = Duration::from_secs(2);
/// How long an order waits before it asks again when the inventory service did not answer.
const RETRY: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Order {
    id: u64,
    sku: String,
    quantity: u32,
}

/// Asks the inventory service and waits for its answer.
async fn ask_stock(
    inventory: &impl RequestReply,
    order: &Order,
) -> Result<Stock, Box<dyn Error + Send + Sync>> {
    let request = JsonCodec.encode(&CheckStock {
        sku: order.sku.clone(),
        quantity: order.quantity,
    })?;
    let reply = inventory
        .request(
            OutgoingMessage::new("inventory.check", request.as_ref()),
            RPC_TIMEOUT,
        )
        .await?;
    Ok(JsonCodec.decode(reply.payload())?)
}

#[subscriber("orders")]
async fn place_order(order: &Order, Out(inventory): Out<impl RequestReply>) -> HandlerOutcome {
    match ask_stock(inventory, order).await {
        Ok(stock) if stock.available => {
            tracing::info!(order.id, "order accepted");
            HandlerOutcome::ack()
        }
        Ok(_) => {
            tracing::warn!(order.id, "order rejected, out of stock");
            HandlerOutcome::drop()
        }
        Err(err) => {
            tracing::warn!(order.id, %err, "inventory unavailable, asking again later");
            HandlerOutcome::retry_after(RETRY)
        }
    }
}

/// The orders service's routes: the requester is a policy, paired with the connection at startup.
fn orders() -> Router<LapinBroker, impl RouterDef<LapinBroker>> {
    Router::new()
        .include(place_order)
        .out(DefaultSlot, Request::default())
        .build()
}

#[ruststream::app]
fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    // The example runs both services in one process; each service mounts its own router.
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(broker, |b| {
        b.include_router(orders());
        b.include_router(inventory());
    })
}

place_order ограничивает свой слот типом Out<impl RequestReply>, а точка монтирования связывает его с политикой Request, то есть LapinRequest. Каждый запрос уходит в обменник по умолчанию с ключом маршрутизации inventory.check и поэтому попадает в одноимённую очередь. В его свойстве reply-to стоит amq.rabbitmq.reply-to RabbitMQ, а correlation-id каждый раз новый. Ответ с тем же идентификатором завершает вызов, а ответ с другим идентификатором отбрасывается. Здесь запрос и ответ - байты, поэтому ask_stock кодирует и декодирует их через JsonCodec.

RPC_TIMEOUT ограничивает ожидание. Если вовремя никто не ответил, вызов возвращает ошибку AmqpError::RequestTimeout. Тогда place_order не угадывает остаток, а просит обработать заказ снова через 30 секунд с retry_after. Отсутствие товара - ответ бизнес-логики, поэтому такой заказ отбрасывается.

Политика принимает настройки публикации: exchange, priority и expiration. По умолчанию запрос не сохраняется на диск: после истечения времени ожидания его никто не ждёт, и переживать перезапуск ему незачем. persistent(true) делает его постоянным.

Гарантии доставки​

Direct reply-to работает по схеме at-most-once. Адрес ответа существует только в канале запрашивающей стороны на одном узле RabbitMQ, и ничего не ставится в устойчивую очередь. Ответ, который был в пути, когда канал закрылся, теряется, и вызывающая сторона получает истечение времени ожидания. Сам запрос - обычное сообщение в обычной очереди: пока сервис склада не работает, запрос ждёт в inventory.check, и ответ на него может прийти, когда вызывающая сторона уже перестала ждать. Поэтому запрос к складу можно безопасно повторить: второй ответ ничего не меняет.

Запрашивающая сторона читает ответы в собственном канале, который открывает её первый запрос.

Другой стек на другой стороне​

Обе стороны говорят на обычном AMQP 0.9.1. Отвечающая сторона на другом клиенте читает очередь запросов, публикует ответ в обменник по умолчанию с ключом маршрутизации из reply-to и копирует correlation-id. Отвечающая сторона на RustStream с DirectReplyTo так же отвечает запрашивающей стороне на другом клиенте.

Тестирование​

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