教程：构建你的第一个服务¶

英文版本为准

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

本教程从零开始构建一个订单服务，并逐块讲解。服务运行在内存 Broker 上，不需要额外启动任何外部服务。换成真正的 Broker 只是一行改动，第 7 步会讲到。

1. 创建 crate¶

cargo new orders-service
cd orders-service

Cargo.toml

[package]
name = "orders-service"
version = "0.1.0"
edition = "2024"

[dependencies]
ruststream = { version = "0.7", features = ["macros", "memory", "json", "asyncapi"] }
serde = { version = "1", features = ["derive"] }

2. 定义消息和处理器¶

处理器是一个 async fn，第一个参数是解码后的载荷。#[subscriber] 宏把它变成可挂载的定义，名字取自函数名。

宏手写

src/orders.rs

use ruststream::runtime::HandlerOutcome;
use ruststream::schemars::JsonSchema;
use ruststream::{Outgoing, subscriber};
use serde::{Deserialize, Serialize};

/// An order placed by a customer.
#[derive(Debug, Deserialize, JsonSchema)]
pub(crate) struct Order {
    pub(crate) id: u64,
    pub(crate) quantity: u32,
}

#[subscriber("orders")]
pub(crate) async fn handle(order: &Order) -> HandlerOutcome {
    println!("order {} x{}", order.id, order.quantity);
    HandlerOutcome::ack()
}

src/orders.rs

use std::future::{Future, ready};

use ruststream::prelude::*;
use ruststream::schemars::JsonSchema;
use serde::Deserialize;

/// An order placed by a customer.
#[derive(Debug, Deserialize, JsonSchema)]
pub(crate) struct Order {
    pub(crate) id: u64,
    pub(crate) quantity: u32,
}

/// The handler: `#[subscriber("orders")]` generates this struct and this impl. The subject it
/// carried is named where the handler is mounted instead.
pub(crate) struct Receive;

impl Handle<Order> for Receive {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        println!("order {} x{}", order.id, order.quantity);
        ready(Ok(()))
    }
}

处理器返回 HandlerOutcome：要么是 ack，要么是 nack。 nack 丢弃消息，或者把它重新入队。处理器也可以返回 () 或 Result<(), E>，其中 Ok 表示 ack，Err 表示丢弃。

JsonSchema derive 把载荷的 schema 写进第 6 步的 AsyncAPI 文档。文档里这条消息的描述取自类型的文档注释。这不需要额外的依赖：asyncapi feature 已经重导出了 schemars。

3. 接入应用¶

宏手写

src/main.rs

mod orders;

use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, RustStream};

use crate::orders::handle;

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders-service", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(handle);
    })
}

src/main.rs

mod orders;

use std::error::Error;

use ruststream::memory::prelude::*;

use crate::orders::Receive;

fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders-service", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(subscriber("orders", Receive).build());
    })
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    app().run().await?;
    Ok(())
}

编解码器的默认值

include 用默认编解码器解码，因此不需要编解码器参数。默认编解码器由 json、cbor、 msgpack 中第一个启用的 feature 选出。要让该 Broker 下的所有处理器换用另一个编解码器，可以用 with_broker_codec(broker, codec, |b| ...) 设定一次。完整的选取规则参见编解码器。

运行它：

cargo run -- run

4. 回复消息¶

要发布一条回复，就返回回复值，并在订阅者上写 publish。目的地由回复类型上的 Outgoing derive 声明：

宏手写

src/orders.rs

/// The service's answer to an order.
#[derive(Debug, Serialize, JsonSchema, Outgoing)]
#[outgoing(name = "confirmations")]
pub(crate) struct Confirmation {
    pub(crate) id: u64,
    pub(crate) accepted: bool,
}

#[subscriber("orders", publish)]
pub(crate) async fn confirm(order: &Order) -> Confirmation {
    Confirmation {
        id: order.id,
        accepted: order.quantity > 0,
    }
}

src/orders.rs

use serde::Serialize;

/// The service's answer to an order, published on `confirmations`.
#[derive(Debug, Serialize, JsonSchema)]
pub(crate) struct Confirmation {
    pub(crate) id: u64,
    pub(crate) accepted: bool,
}

/// What `#[derive(Outgoing)]` with `#[outgoing(name = "confirmations")]` writes: the destination
/// is fixed on the type, so nothing else names it.
impl OutgoingDestination for Confirmation {
    type Form = FixedName;

    const DESTINATION: &'static str = "confirmations";
}

impl MessageHeaders for Confirmation {
    type Contract = NoHeaders;
}

/// The reply form of the same trait: the second parameter of `Handle` is the reply type, so the
/// body returns a `Confirmation`, and the mount site supplies the publisher it leaves through.
pub(crate) struct Confirm;

impl Handle<Order, Confirmation> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        ready(Ok(Confirmation {
            id: order.id,
            accepted: order.quantity > 0,
        }))
    }
}

用同一个 include 把 confirm 挂在 handle 旁边。回复由 Broker 的默认发布策略发出，并用默认编解码器编码。

宏手写

src/main.rs

use crate::orders::{confirm, handle};

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders-service", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(handle);
        b.include(confirm);
    })
}

src/main.rs

use crate::orders::{Confirm, Receive};

fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders-service", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(subscriber("orders", Receive).build());
        // The definition names the reply's destination; with no `.out_reply(..)` at the mount
        // site the reply leaves through the broker's default publisher.
        b.include(
            subscriber("orders", Confirm)
                .reply()
                .to("confirmations")
                .build(),
        );
    })
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    app().run().await?;
    Ok(())
}

在处理器内部发布以及其余的发布方式，参见发布与回复。

5. 用路由器组织代码¶

处理器多起来以后，把它们放进各自的模块，再汇总到 Router 里：

宏手写

src/routes.rs

use ruststream::memory::prelude::*;

use crate::orders;

// The reply wiring is a publish policy: pure declaration, so the router needs no broker at all.
pub(crate) fn orders() -> impl RouterDef<MemoryBroker> {
    Router::new()
        .include(orders::handle)
        .include(orders::confirm)
        .out_reply(Publish)
        .build()
}

src/routes.rs

use ruststream::memory::prelude::*;

use crate::orders::{Confirm, Receive};

// Each handler is bound to its subject where it is mounted; the definition says what it replies
// with, the reply type says where it goes, and the mount chain names who publishes it. The
// publisher wiring is still a publish policy - pure declaration, so the router needs no broker
// at all.
pub(crate) fn orders() -> impl RouterDef<MemoryBroker> {
    Router::new()
        .include(subscriber("orders", Receive).build())
        .include(subscriber("orders", Confirm).reply().build())
        .out_reply(Publish)
        .build()
}

带回复的处理器用链式调用挂到路由器上：.out_reply(..) 指定回复的发布策略，.build() 提交这次注册。不写 .out_reply(..) 时，.build() 采用 Broker 的默认发布策略，也就是第 4 步里 include 用的那一个。路由器的其余用法参见路由。

宏手写

src/main.rs

mod orders;
mod routes;

use ruststream::memory::MemoryBroker;
use ruststream::runtime::{AppInfo, RustStream};

#[ruststream::app]
fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders-service", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        let router = routes::orders();
        b.include_router(router);
    })
}

src/main.rs

mod orders;
mod routes;

use std::error::Error;

use ruststream::memory::prelude::*;

fn app() -> RustStream {
    RustStream::new(AppInfo::new("orders-service", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        let router = routes::orders();
        b.include_router(router);
    })
}

// What `#[ruststream::app]` wraps around the builder: the runtime entry point. Its CLI
// (`run`, `asyncapi gen`) is what a hand-written `main` gives up.
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    app().run().await?;
    Ok(())
}

6. 查看 AsyncAPI 文档¶

cargo run -- asyncapi gen

每个订阅者都会给文档添加一个通道和一个 receive 操作。handle 和 confirm 共用 orders 这个通道，但各自有一个操作：两者的订阅是分开的。回复在 confirmations 上添加一个 send 操作。

载荷的 schema 放在文档的 components.messages 下。输出参数（-o、--yaml）和文档本身参见 AsyncAPI。

7. 换成真正的 Broker¶

上面写的一切都不绑定在内存 Broker 上。Broker 在 with_broker 处选定，更换只是一行改动。把对应的 Broker crate 加进依赖，在那里构造它，例如用 NatsBroker::new("nats://localhost:4222") 代替 MemoryBroker::new()。处理器、路由器和编解码器保持不变。可用的 Broker 和每一种的替换写法，参见 Broker。

完整的服务是一个可编译的示例

本页的每一段代码都来自仓库里的 examples/tutorial， CI 每次改动都会构建它。first_app.rs 和 reply_app.rs 是第 3 步和第 4 步结束时的服务， main.rs 是最终版本。你也可以用 cargo run --example tutorial --features macros,memory,json,asyncapi -- run 自己运行一遍。

下一步¶

中间件：围绕处理器的横切逻辑。
生命周期：共享状态与启动/关闭钩子。
测试：在进程内测试你刚写好的处理器。
指标：Prometheus 计数器与直方图。