快速开始

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

安装

Cargo.toml

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

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

ruststream 是框架本身：macros 提供属性宏，json 提供默认编解码器，logging 把服务的日志打印到终端。 ruststream-sqs-sns 是 SQS Broker，基于官方客户端 aws-sdk-sqs 构建。 它的 testing feature 加入测试所用的进程内 SQS。

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

Feature	提供的内容
sns	通过 aws-sdk-sns，用 SnsPublish 发布到 SNS 主题（用 SNS 扇出）
asyncapi	生成的 AsyncAPI 文档中的 sqs 和 sns 绑定
testing	TestApp 背后的进程内 SQS

本地 SQS

LocalStack 在一个容器里运行 SQS 和 SNS：

compose.yaml

services:
  localstack:
    image: localstack/localstack:4.14.0
    ports:
      - "4566:4566"
    environment:
      SERVICES: sqs,sns
      SQS_ENDPOINT_STRATEGY: path

docker compose up -d

它在 http://localhost:4566 上应答，启动时没有任何队列。

第一个服务

宏

src/main.rs

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

// The local stack starts empty, so the descriptor creates the queue on subscribe.
#[subscriber(SqsQueue::new("orders").create_if_missing())]
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(
        SqsBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
        |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(
        SqsBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
        |b| {
            // The local stack starts empty, so the descriptor creates the queue on subscribe.
            b.include(
                subscriber("orders", TakeOrder)
                    .map_source(|_| SqsQueue::new("orders").create_if_missing())
                    .build(),
            );
        },
    )
}

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

take_order 从 orders 队列读取 JSON 订单。 带金额的订单被 ack，没有金额的订单被丢弃。 SqsQueue 指定队列，create_if_missing 在订阅打开时创建它。 生产环境的服务不用这一步，它的队列来自基础设施。 SqsBroker::new 只记录配置。 endpoint、region 和 test_credentials 让它指向本地环境。 SDK 客户端在服务启动时才构建。 #[ruststream::app] 把 app 变成 main；在手写形式中，run_main 做同样的事。

运行服务：

cargo run -- run

在另一个终端里，用 AWS CLI 发送一个订单：

export AWS_ENDPOINT_URL=http://localhost:4566 AWS_REGION=us-east-1 \
  AWS_ACCESS_KEY_ID=test AWS_SECRET_ACCESS_KEY=test
aws sqs send-message \
  --queue-url "$(aws sqs get-queue-url --queue-name orders --output text)" \
  --message-body '{"id":1,"amount_cents":1250}'

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

第一个测试

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

cargo test

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

从 aws-sdk-sqs 迁移

基于 aws-sdk-sqs 的服务要自己编写接收循环。 下表列出每一部分在 RustStream 中的位置：

aws-sdk-sqs	RustStream
aws_config::load_from_env 和 Client::new	SqsBroker::new，或者 SqsBroker::from_config 加上你自己的 SdkConfig（在生产环境中运行）
get_queue_url	SqsQueue::new 中的队列名称，启动时解析一次
一个 receive_message().wait_time_seconds(20) 循环	一个 #[subscriber] 处理器（消费队列）
反序列化 message.body()	处理器的参数类型，由编解码器解码
处理完成后调用 delete_message	处理器的结果 HandlerOutcome::ack（确认与投递保证）
change_message_visibility	HandlerOutcome::retry_after（重试与死信队列）
在控制台中设置的 RedrivePolicy	挂载处的 max_attempts 和 dead_letter
send_message	回复或 Out 参数，经由 SqsPublish 发送
message_group_id 和 message_deduplication_id	SqsPublishSteps 的发布步骤 group_id 和 deduplication_id（FIFO 队列）