在生产环境中运行

本指南在 AWS 上运行一个 Kinesis 服务：它的凭据和区域从哪里来，它期望找到哪些流，以及它需要的 IAM 权限。

来自环境的凭据和区域

KinesisBroker::new() 不从代码中接收任何东西。服务启动时，AWS SDK 的默认提供链找到区域和凭据：环境变量、共享的 ~/.aws 文件、SSO，以及容器或实例的角色。在 ECS、EKS 或 EC2 上以 IAM 角色运行的服务完全不需要配置。Broker 用 SDK 最新的行为版本构建它的配置。

SDK 在为第一个请求签名时读取凭据。这个请求就是启动时对第一个订阅的流所做的检查，所以凭据错误的服务在处理任何记录之前就会停止。

你自己的配置

宏

src/main.rs

/// The AWS configuration of the service: a pinned behaviour version, its region, and the
/// credentials of the `orders-service` profile, read when the first request is signed.
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))
        .build()
}

手写

src/main.rs

/// The AWS configuration of the service: a pinned behaviour version, its region, and the
/// credentials of the `orders-service` profile, read when the first request is signed.
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))
        .build()
}

KinesisBroker::from_config 接收服务构建的 SdkConfig。aws_config 用一个 async 加载器从环境构建配置，而应用函数是同步的。所以示例用不做任何 I/O 的 SdkConfig::builder() 构建配置。配置文件提供者在第一个请求时读取 orders-service 配置文件；带有 role_arn 的配置文件会代入该角色。固定的行为版本让 SDK 更新时，它的默认值不会在服务不知情时改变。Broker 按原样使用这份配置：endpoint、region 和 test_credentials 只作用于 KinesisBroker::new()，你自己的配置在构建器上用 endpoint_url、region 和 credentials_provider 指定它们。DynamoLeaseStore 接收同一份配置（多个实例）。示例还需要两个依赖：

Cargo.toml

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

服务

宏

src/main.rs

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

// The stream belongs to the infrastructure, so the descriptor does not create it.
#[subscriber(KinesisStream::new("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(
        KinesisBroker::from_config(aws_config()),
        |b| {
            b.include(take_order);
        },
    )
}

手写

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
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;
}

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(
        KinesisBroker::from_config(aws_config()),
        |b| {
            // The stream belongs to the infrastructure, so the descriptor does not create it.
            b.include(subscriber(KinesisStream::new("orders"), TakeOrder).build());
        },
    )
}

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

orders 流属于基础设施：由 Terraform、CloudFormation 或控制台创建，连同它的容量模式、保留期和加密设置。流用名称指定；账户和区域来自配置。描述符省去了 create_if_missing，所以缺失的流会让服务在启动时停止，并返回指出该流的错误，而不是创建一个空流。create_if_missing(n) 创建一个有 n 个分片的预置模式流，适合本地的 LocalStack；生产流采用预置模式还是按需模式，由运维人员决定。

请求经由 SDK 默认的 HTTPS 客户端发送，SDK 按自己的重试策略重试失败的请求。运行在专用线程上的处理器（threads(n)）通过它自己的 SDK 客户端发送，该客户端由同一份配置构建，并共享其凭据。

IAM 权限

上面的服务需要对它的流拥有以下权限：

policy.json

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "kinesis:DescribeStreamSummary",
        "kinesis:ListShards",
        "kinesis:GetShardIterator",
        "kinesis:GetRecords"
      ],
      "Resource": "arn:aws:kinesis:eu-west-1:123456789012:stream/orders"
    }
  ]
}

每项功能各自增加所需的权限：

功能	权限
发布、回复或重试副本	对目标流的 kinesis:PutRecord
create_if_missing	kinesis:CreateStream 和 kinesis:DescribeStream
DynamoLeaseStore	对租约表的 dynamodb:GetItem 和 dynamodb:UpdateItem

AsyncAPI 文档显示什么

启用 asyncapi feature 后，服务的 AsyncAPI 文档指定一个服务器：Broker 的 endpoint 的主机；region 指定了区域时为 kinesis.<region>.amazonaws.com；否则为 kinesis.amazonaws.com。文档从不包含凭据，所以可以公开发布。

示例文件中的测试在进程内启动同一个应用；进程内 Kinesis 不签名任何请求，也不读取任何配置文件。测试 Kinesis 服务讲解了测试工具。