在生产环境中运行

本指南在 AWS 上运行一个 SQS 服务：它的凭证和区域从哪里来，它使用的超时，以及它所需的 IAM 权限。

来自环境的凭证和区域

SqsBroker::new() 不从代码中获取任何东西。 服务启动时，AWS SDK 的默认链查找区域和凭证：环境变量、共享的 ~/.aws 文件、SSO，以及容器或实例的角色。 在 ECS、EKS、Lambda 或 EC2 上以 IAM 角色运行的服务完全不需要配置。

SDK 在签名第一个请求时读取凭证。 这个请求就是第一个订阅在启动时发出的 GetQueueUrl，所以凭证错误的服务会在处理任何消息之前停止。

你自己的配置

宏

src/main.rs

/// The AWS configuration of the service: a pinned behaviour version, its region, the
/// credentials of the `orders-service` profile, read when the first request is signed, and a
/// request timeout above the 20 seconds a long poll waits.
fn aws_config() -> SdkConfig {
    let credentials = ProfileFileCredentialsProvider::builder()
        .profile_name("orders-service")
        .build();
    SdkConfig::builder()
        .behavior_version(BehaviorVersion::v2026_01_12())
        .region(Region::new("eu-west-1"))
        .credentials_provider(SharedCredentialsProvider::new(credentials))
        .timeout_config(
            TimeoutConfig::builder()
                .operation_attempt_timeout(Duration::from_secs(25))
                .build(),
        )
        .build()
}

手写

src/main.rs

/// The AWS configuration of the service: a pinned behaviour version, its region, the
/// credentials of the `orders-service` profile, read when the first request is signed, and a
/// request timeout above the 20 seconds a long poll waits.
fn aws_config() -> SdkConfig {
    let credentials = ProfileFileCredentialsProvider::builder()
        .profile_name("orders-service")
        .build();
    SdkConfig::builder()
        .behavior_version(BehaviorVersion::v2026_01_12())
        .region(Region::new("eu-west-1"))
        .credentials_provider(SharedCredentialsProvider::new(credentials))
        .timeout_config(
            TimeoutConfig::builder()
                .operation_attempt_timeout(Duration::from_secs(25))
                .build(),
        )
        .build()
}

SqsBroker::from_config 接收服务构建的 SdkConfig。 aws_config 用一个 async 加载器从环境构建配置，而应用函数是同步的。 所以示例用不做 I/O 的 SdkConfig::builder() 构建它。 profile 提供者在第一次请求时读取 orders-service profile；带有 role_arn 的 profile 会承担该角色。 固定的行为版本让 SDK 更新时，服务所用的 SDK 默认值不随之改变。 配置还要携带下文所说的请求超时。 Broker 保留配置的全部设置，只有在 Broker 自身上指定的端点和区域会取代配置中的值。 示例还需要两个依赖：

Cargo.toml

[dependencies]
aws-config = "1"
aws-sdk-sqs = "1"

服务

宏

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
    amount_cents: u64,
}

// The queue belongs to the infrastructure, so the descriptor does not create it. It names its
// visibility, so the subscription does not read the queue's own.
#[subscriber(SqsQueue::new("orders").visibility(Duration::from_secs(120)))]
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::from_config(aws_config()),
        |b| {
            b.include(take_order);
        },
    )
}

手写

src/main.rs

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

impl OutgoingDestination for Order {
    type Form = FixedName;

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

impl MessageHeaders for Order {
    type Contract = NoHeaders;
}

impl MessageInfo for Order {
    const NAME: &'static str = "Order";
}

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::from_config(aws_config()),
        |b| {
            // The queue belongs to the infrastructure, so the descriptor does not create it. It
            // names its visibility, so the subscription does not read the queue's own.
            b.include(
                subscriber("orders", TakeOrder)
                    .map_source(|_| SqsQueue::new("orders").visibility(Duration::from_secs(120)))
                    .build(),
            );
        },
    )
}

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

orders 队列属于基础设施：由 Terraform、CloudFormation 或控制台创建，连同它的重新驱动策略和加密设置。 描述符不用 create_if_missing，所以缺失的队列会让服务在启动时停止，而不是创建一个空队列。 描述符指定了可见性，所以订阅不读取队列自己的超时。

endpoint 和 region 让 Broker 指向另一个端点、VPC 端点或本地环境。 对于用 from_config 构建的 Broker，它们会取代配置自身的 endpoint_url 和 region。 SQS 返回的队列 URL 会被改写到该端点上。

超时与传输

SqsBroker::new() 给每个 SQS 请求每次尝试 25 秒，高于长轮询最多等待的 20 秒。 你自己的配置用 timeout_config 设置它，示例就是这样做的；低于 20 秒的尝试超时会让每次长轮询都中途结束。 SDK 按自己的重试策略重试失败的请求。 请求通过 SDK 默认的 HTTPS 客户端发送。 专用线程上的处理器（threads(n)）通过自己的 SDK 客户端发送，这些客户端由同一份配置构建，共用它的凭证。

IAM 权限

上面的服务需要在它的队列上拥有这些权限：

policy.json

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "sqs:GetQueueUrl",
        "sqs:ReceiveMessage",
        "sqs:DeleteMessage",
        "sqs:ChangeMessageVisibility"
      ],
      "Resource": "arn:aws:sqs:eu-west-1:123456789012:orders"
    }
  ]
}

每项功能都会增加自己的权限：

功能	权限
没有指定可见性的订阅	队列上的 sqs:GetQueueAttributes
max_attempts 和 dead_letter	队列上的 sqs:SetQueueAttributes，死信队列上的 sqs:GetQueueAttributes
create_if_missing	sqs:CreateQueue
向队列发布	队列上的 sqs:GetQueueUrl 和 sqs:SendMessage
向主题发布	主题上的 sns:CreateTopic 和 sns:Publish
subscribe_queue_to_topic	主题上的 sns:CreateTopic 和 sns:Subscribe，队列上的 sqs:GetQueueUrl、sqs:GetQueueAttributes 和 sqs:SetQueueAttributes

AsyncAPI 文档展示什么

启用 asyncapi feature 时，服务的 AsyncAPI 文档指定一个服务器：Broker 端点所在的主机；没有端点时，是其区域的区域主机，例如 sqs.eu-west-1.amazonaws.com。 文档从不包含凭证或队列的 ARN，因此可以公开发布。

示例文件中的测试在进程内启动同一个应用；进程内 SQS 从配置中获取区域，不签名任何东西。 测试 SQS 服务介绍测试工具。