快速开始

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

安装

Cargo.toml

[dependencies]
ruststream = { version = "0.7", features = ["macros", "json", "logging"] }
ruststream-amqp = "0.7"
serde = { version = "1", features = ["derive"] }
tracing = "0.1"

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

ruststream 是框架本身：macros 提供属性宏，json 提供默认编解码器，logging 把服务的日志打印到终端。 ruststream-amqp 是 AMQP 1.0 Broker。 它的 testing feature 加入测试所用的进程内 Broker。

这个 Broker crate 有以下 feature，默认全部关闭：

Feature	提供的内容
rustls	通过 rustls 建立基于 TLS 的 amqps:// 连接
native-tls	通过平台自带的 TLS 建立 amqps:// 连接
transaction	事务性发布
asyncapi	生成的 AsyncAPI 文档中的 AMQP 1.0 细节
testing	TestApp 背后的进程内 Broker

底层客户端 fe2o3-amqp 用 Rust 编写，因此构建不需要任何系统库。

本地 Broker

开发时，一个 ActiveMQ Artemis 就够了：

compose.yaml

services:
  artemis:
    image: apache/activemq-artemis:latest-alpine
    environment:
      ARTEMIS_USER: artemis
      ARTEMIS_PASSWORD: artemis
    ports:
      - "5672:5672"

docker compose up -d

Broker 在 localhost:5672 上监听 AMQP，用户名是 artemis，密码是 artemis。 服务第一次读取某个队列时，Artemis 会创建它，所以无需事先声明任何东西。

第一个服务

宏

src/main.rs

#[derive(Debug, PartialEq, Deserialize)]
struct Order {
    id: u64,
    amount_cents: u64,
}

#[subscriber(AmqpAddress::queue("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(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |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(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            b.include(subscriber(AmqpAddress::queue("orders"), TakeOrder).build());
        },
    )
}

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

take_order 从队列 orders 读取 JSON 订单。 带金额的订单被接受，没有金额的订单被拒绝。 AmqpAddress::queue 指明队列，队列与主题介绍其他种类的地址。 AmqpBroker::new 接收 Broker 的 URL，只把它记录下来。 连接在服务启动时才打开。 服务用 URL 中的用户名和密码登录。 #[ruststream::app] 把 app 变成 main；在手写形式中，run_main 做同样的事。

运行服务：

cargo run -- run

在另一个终端里，用镜像自带的命令行客户端发布一个订单：

docker compose exec artemis /opt/activemq-artemis/bin/artemis producer \
  --protocol amqp --user artemis --password artemis \
  --destination queue://orders --message-count 1 --message '{"id":1,"amount_cents":1250}'

服务记录一条带订单 id 的 order taken 日志。 Ctrl+C 停止服务。

第一个测试

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

cargo test

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

从 fe2o3-amqp 迁移

基于 fe2o3-amqp 的服务要自己打开连接和会话，附加链路，并编写接收循环。 下表列出每一部分在 RustStream 中的位置：

fe2o3-amqp	RustStream
Connection::open	AmqpBroker::new，以及 sasl 和 container_id
Session::begin	无需编写：每个订阅有自己的会话，各个发布者共用一个会话
Receiver::attach 加一个 recv 循环	一个 #[subscriber(AmqpAddress::..)] 处理器
CreditMode::Auto(n)	AmqpAddress::credit
accept、modify、reject	处理器的结果：ack、retry、drop（确认与投递保证）
auto_accept(true)	Settle::AtMostOnce
Sender::attach 和 send	回复或 Out 参数，经由 Publish 策略发送
用于接收回复的动态接收者	request（请求与回复）
Controller 和 Transaction::declare	TransactionalPublish（全有或全无的发布）

通过 AMQP 0.9.1 与 RabbitMQ 通信的服务应改用 ruststream-lapin：两种协议共用端口和 amqp:// 协议名，但它们是不同的协议。