跳到主要内容
英文版本为准

本页译自英文，由模型翻译。若与英文原文存在差异，以英文原文为准。

请求与回复

本指南让一个服务通过 RabbitMQ 调用另一个服务，并等待回复。

订单服务在接受订单之前，先询问库存服务某件商品是否有货。 调用经由 RabbitMQ 的 direct reply-to 进行：不需要声明回复队列，也不需要手工维护一张待处理调用表。

应答方​

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 设为 RabbitMQ 的 amq.rabbitmq.reply-to，并带有一个新生成的 correlation-id。 id 相同的回复完成这次调用；id 不同的回复被丢弃。 这里的请求和回复都是字节，所以 ask_stock 用 JsonCodec 对它们编码和解码。

RPC_TIMEOUT 限定等待时间。 没有人及时应答时，调用返回 AmqpError::RequestTimeout 错误。 这时 place_order 用 retry_after 要求 30 秒后再投递这个订单，而不是去猜测库存。 商品缺货是业务上的答复，所以订单被丢弃。

这个策略接受发布所用的设置：exchange、priority 和 expiration。 请求默认是非持久化的：超时之后没人再等它，所以它在重启后存活下来也没有意义。 persistent(true) 让它持久化。

投递保证​

direct reply-to 是至多一次的。 回复地址存在于请求方位于某个 RabbitMQ 节点上的信道里，没有任何东西被持久地排入队列。 信道关闭时仍在途中的回复会丢失，调用方看到的是超时。 请求本身是普通队列里的一条普通消息：库存服务停机时，它在 inventory.check 中等待，可能在调用方放弃之后才得到应答。 所以库存查询可以安全地重复：应答两次不会改变任何东西。

请求方在自己的信道上读取回复，这个信道由它的第一个请求打开。

另一端使用其他技术栈​

双方说的都是普通的 AMQP 0.9.1。 用其他客户端编写的应答方消费请求队列，把回复发布到默认交换机，路由键取自 reply-to，并复制 correlation-id。 带 DirectReplyTo 的 RustStream 应答方也以同样的方式回复用其他客户端编写的请求方。

测试​

进程内 RabbitMQ 属于连接它的那个应用，所以请求方和应答方必须运行在同一个应用里才能通信。 测试启动示例中挂载了两个路由器的应用，检查有货的订单被接受、缺货的订单被丢弃。 第三个测试在暂停的时钟上只挂载订单路由器：没有人应答，请求超时，订单稍后再次询问。 进程内 RabbitMQ 像服务器一样为每个请求分配自己的回复地址，并按关联 id 匹配回复。 测试 RabbitMQ 服务讲解了测试套件。