快速开始

本指南安装 ruststream-lapin，启动一个本地 RabbitMQ，并让第一个服务连上它运行。

安装

Cargo.toml
[dependencies]
ruststream = { version = "0.7", features = ["macros", "json", "asyncapi", "logging"] }
ruststream-lapin = "0.7"
serde = { version = "1", features = ["derive"] }
tracing = "0.1"

[dev-dependencies]
ruststream-lapin = { version = "0.7", features = ["testing"] }
tokio = { version = "1", features = ["macros", "rt"] }

ruststream 是框架本身：macros 提供属性宏，json 提供默认编解码器，asyncapi 生成服务的 AsyncAPI 文档，logging 把服务的日志打印到终端。 ruststream-lapin 是 RabbitMQ Broker，基于 lapin 构建，lapin 是用 Rust 编写的 AMQP 0.9.1 客户端。构建只需要 Cargo。它的 testing feature 加入测试所用的进程内 RabbitMQ。

这个 Broker crate 有以下 feature：

Feature	提供的内容
`asyncapi`	生成的 AsyncAPI 文档中的 AMQP 绑定
`testing`	`TestApp` 背后的进程内 RabbitMQ
`tls-native-tls`	通过平台自带的 TLS 库提供 TLS
`tls-rustls`	通过 rustls 提供 TLS
`tls-rustls-ring`	通过使用 `ring` 加密后端的 rustls 提供 TLS
`plugin-consistent-hash`	RabbitMQ 插件提供的一致性哈希交换机
`plugin-dme`	借助延迟消息交换机插件实现的延迟重试

本地 RabbitMQ

开发时，一个带管理界面的 RabbitMQ 节点就够了：

compose.yaml
services:
  rabbitmq:
    image: rabbitmq:4-management-alpine
    ports:
      - "127.0.0.1:5672:5672"
      - "127.0.0.1:15672:15672"

docker compose up -d

Broker 监听 localhost:5672。管理界面位于 http://localhost:15672，用户名 guest，密码 guest。 RabbitMQ 只接受来自 localhost 的 guest 账户，所以端口只发布在本机地址上。

第一个服务

宏
手写

src/main.rs
#[derive(Debug, PartialEq, Deserialize)]
struct Order {
    id: u64,
    amount_cents: u64,
}

#[subscriber("orders")]
async fn take_order(order: &Order) -> HandlerOutcome {
    if order.amount_cents == 0 {
        tracing::warn!(order.id, "order without an amount dropped");
        return HandlerOutcome::drop();
    }
    tracing::info!(order.id, order.amount_cents, "order taken");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672").declare_topology(true),
        |b| {
            b.include(take_order);
        },
    )
}

src/main.rs
#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Order {
    id: u64,
    amount_cents: u64,
}

struct TakeOrder;

impl Handle<Order> for TakeOrder {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if order.amount_cents == 0 {
            tracing::warn!(order.id, "order without an amount dropped");
            return ready(Err(HandlerOutcome::drop()));
        }
        tracing::info!(order.id, order.amount_cents, "order taken");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        LapinBroker::new("amqp://localhost:5672").declare_topology(true),
        |b| {
            b.include(subscriber("orders", TakeOrder).build());
        },
    )
}

fn main() -> ExitCode {
    run_main(app)
}

take_order 从 orders 队列读取 JSON 订单。带金额的订单被确认，没有金额的订单被丢弃。 LapinBroker::new 接收连接 URI，只把它记录下来。连接在服务启动时才打开。 declare_topology(true) 让服务自己声明它读取的队列，所以服务可以在空的 RabbitMQ 上启动。 #[ruststream::app] 把 app 变成 main；在手写形式中，run_main 做同样的事。

运行服务：

cargo run -- run

在另一个终端里，用镜像自带的 rabbitmqadmin 发布一个订单：

docker compose exec rabbitmq rabbitmqadmin publish message \
  --routing-key orders --payload '{"id":1,"amount_cents":1250}'

消息进入默认交换机，默认交换机把它投递到与路由键同名的队列。服务记录一条带订单 id 的 order taken 日志。在管理界面里，orders 队列的页面也能发布同样的消息。 Ctrl+C 停止服务。

服务还能描述自己：

cargo run -- asyncapi gen

这条命令打印服务的 AsyncAPI 文档：服务器、orders 通道和 Order 消息。

cargo generate 可以生成同类服务的脚手架，其中带有路由器和回复：

cargo generate --git https://github.com/powersemmi/ruststream-lapin templates/amqp-queue --name my-service

templates/amqp-topic 模板则从一个 topic 交换机开始。

第一个测试

示例文件里还有两个测试。每个测试都用 TestApp::start 启动同一个 app，发布一个订单，然后检查 take_order 如何结算它。在测试中，LapinBroker 连接的是进程内 RabbitMQ，所以测试不需要服务器：

cargo test

测试 RabbitMQ 服务逐一讲解这些测试，以及进程内 RabbitMQ 模拟了什么。

从 lapin 迁移

基于 lapin 的服务要自己打开信道，自己编写消费者循环。下表列出每一部分在 RustStream 中的位置：

lapin	RustStream
`Connection::connect` 和 `create_channel`	`LapinBroker::new`；运行时在启动时建立连接
`channel.basic_qos`	Broker 上或单个队列上的 `prefetch`
`queue_declare`、`exchange_declare`、`queue_bind`	队列描述符和 `declare_topology(true)`（拓扑）
`basic_consume` 加一个 `while let Some(delivery)` 循环	一个 `#[subscriber]` 处理器
反序列化 `delivery.data`	处理器的参数类型，由编解码器解码
`delivery.ack` 和 `delivery.reject`	处理器的结果（确认、重试与死信）
`basic_publish`	回复或 `Out` 参数，经由发布策略发送
`confirm_select` 并等待确认	`ConfirmsPublish`（可靠发布）
回调队列和关联表	`LapinRequest`（请求与回复）

安装​

本地 RabbitMQ​

第一个服务​

第一个测试​

从 lapin 迁移​

安装

本地 RabbitMQ

第一个服务

第一个测试

从 lapin 迁移