读取流

本指南逐条和按批读取一个流，分配每个分片的读取额度，并展示一次投递带有什么。

示例是一个分析服务。它在付款到达时逐笔记账，并按批统计页面浏览。

流是如何读取的

订阅为它持有的每个分片运行一个读取者。读取者用分片的迭代器调用 GetRecords，按顺序交出记录，然后立即再次请求。如果一次读取什么也没返回，读取者先暂停 poll_interval（默认一秒），再发起请求。读取者和处理器之间最多有 64 次投递在等待；读取者发现队列已满时就等待，所以慢处理器会让读取变慢。

Kinesis 允许每个分片每秒五次 GetRecords 调用和每秒 2 MB。流的所有消费者共用这份额度。读取过于频繁的服务会被限流：Kinesis 返回 ProvisionedThroughputExceededException。AWS SDK 先自行重试被限流的调用。重试用完后，读取者等待一个 poll_interval，再用同一个迭代器请求。处理器看到的只是一次停顿，既没有错误，也没有重新投递。

逐条读取

宏

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "payments")]
struct Payment {
    id: u64,
    currency: String,
}

/// Books the payments in euros and drops the rest, which another service books.
#[subscriber(
    KinesisStream::new("payments")
        .poll_interval(Duration::from_millis(250))
        .create_if_missing(1)
)]
async fn book(payment: &Payment, Ctx(at): Ctx<Position>) -> HandlerOutcome {
    if payment.currency != "EUR" {
        return HandlerOutcome::drop();
    }
    tracing::info!(payment.id, position = ?at, "payment booked");
    HandlerOutcome::ack()
}

手写

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
struct Payment {
    id: u64,
    currency: String,
}

impl OutgoingDestination for Payment {
    type Form = FixedName;

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

impl MessageHeaders for Payment {
    type Contract = NoHeaders;
}

/// Books the payments in euros and drops the rest, which another service books.
struct Book;

impl Handle<Payment, (), (), KinesisContext> for Book {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        ctx: &mut Context<'_, KinesisContext>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if payment.currency != "EUR" {
            return ready(Err(HandlerOutcome::drop()));
        }
        let at = ctx.context(Position);
        tracing::info!(payment.id, position = ?at, "payment booked");
        ready(Ok(()))
    }
}

book 读取 payments 流，逐笔结算每次付款。逐条读取的订阅每次读取最多请求 1000 条记录。poll_interval(Duration::from_millis(250)) 缩短空读取之后的暂停，所以新付款最多等待四分之一秒。每秒四次空读取，给其他消费者留下分片额度中每秒一次读取。这一设置也是一个挂载点步骤：在 b.include(..) 之后写 .poll_interval(..)。

Ctx<Position> 读取记录在流中的位置：它的分片和序列号，形式为 KinesisPosition。seek 到这个位置，会从这条记录开始重新读取该分片（重放与 seek）。每次投递还带有三个消息头：

消息头	值
kinesis-shard-id（SHARD_HEADER）	读出记录的分片
kinesis-sequence-number（SEQUENCE_HEADER）	记录在该分片中的序列号
partition-key（PARTITION_KEY_HEADER）	发布记录时使用的分区键

处理器用 ctx.headers() 读取它们。记录的其他消息头由它的生产者写入。

批

宏

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "page-views")]
struct PageView {
    page: String,
}

/// Counts page views a batch at a time and settles the whole batch with one outcome.
#[subscriber(KinesisStream::new("page-views").create_if_missing(1))]
async fn count(views: &[PageView]) -> HandlerOutcome {
    tracing::info!(views = views.len(), "page views counted");
    HandlerOutcome::ack()
}

手写

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
struct PageView {
    page: String,
}

impl OutgoingDestination for PageView {
    type Form = FixedName;

    const DESTINATION: &'static str = "page-views";
}

impl MessageHeaders for PageView {
    type Contract = NoHeaders;
}

/// Counts page views a batch at a time and settles the whole batch with one outcome.
struct Count;

impl Handle<[PageView]> for Count {
    fn handle(
        &self,
        views: &[PageView],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        tracing::info!(views = views.len(), "page views counted");
        ready(Ok(()))
    }
}

接收切片的处理器是批处理器。应用用 .batch(nonzero!(500)) 挂载 count，500 就成为它的读取者每次 GetRecords 调用的上限。因此一次读取取回的记录绝不会超过一批，而 Kinesis 把一次读取限制在 10 000 条记录以内。未满的批在它的第一条记录到达 50 毫秒后发出，所以流量很小的流也能很快投递它的记录。一批可以混有本实例持有的所有分片的记录。

批大小是延迟与读取额度之间的取舍。批小，读取就小，而每次读取都要占用分片每秒五次调用中的一次。count 用一个结果结算整批。返回 Vec<HandlerOutcome> 的处理器逐条结算每条记录，每条记录各自推进自己分片的检查点（检查点与投递保证）。

应用

宏

src/main.rs

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("analytics", "0.1.0")).with_broker(broker, |b| {
        b.include(book);
        b.include(count.batch(nonzero!(500)));
    })
}

#[ruststream::app]
fn app() -> impl App {
    service(
        KinesisBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
    )
}

手写

src/main.rs

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("analytics", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber(
                KinesisStream::new("payments")
                    .poll_interval(Duration::from_millis(250))
                    .create_if_missing(1),
                Book,
            )
            .build(),
        );
        b.include(
            subscriber(KinesisStream::new("page-views").create_if_missing(1), Count)
                .batch(nonzero!(500))
                .build(),
        );
    })
}

fn main() -> ExitCode {
    run_main(|| {
        service(
            KinesisBroker::new()
                .endpoint("http://localhost:4566")
                .region("us-east-1")
                .test_credentials(),
        )
    })
}

处理器默认一次运行一次投递。属性中的 workers(n)，或手写形式中的 workers 步骤，最多并行运行 n 次投递；分片、分区键与顺序展示了如何在并行时保持每个键的顺序。

本 crate 通过流的共享吞吐量、用 GetRecords 读取。由 Kinesis Producer Library 聚合的记录不会交给处理器：订阅把它报告为 KinesisError::AggregatedRecord，然后继续读取。

示例文件中的测试在进程内运行应用。批测试在暂停的时钟上启动，所以未满的批所等待的 50 毫秒会瞬间过去。