跳到主要内容
英文版本为准

本页译自英文，由模型翻译。若与英文原文存在差异，以英文原文为准。

重放与 seek

本指南从最旧的记录或从某个时刻重新读取流，并在处理器内部移动正在运行的订阅。

消费者读过一条记录之后，Kinesis 仍然保留它，直到流的保留期结束。服务可以再读一遍：重建读模型，为新的消费者回填数据，或者跳过它无法使用的记录。订阅在挂载点上取得起始位置，处理器则通过投递的上下文移动自己的订阅。

读模型​

服务在内存中维护两个读模型：每个账户的余额和每张卡的消费额。每条记账都带有账簿的修订号。对过去记账的更正作为新的修订发布，读模型遇到新修订时，会从流的最旧记录开始重建自己。

src/main.rs
/// One posting of the ledger. `revision` is the ledger's revision: a correction of past
/// postings is published as a new revision, and a read model meeting one rebuilds itself.
#[derive(Debug, PartialEq, Deserialize)]
struct Posting {
    id: u64,
    account: String,
    amount_cents: i64,
    revision: u32,
}

/// A read model kept in memory: the balance of every account.
#[derive(Debug, Default)]
struct Projection {
    revision: u32,
    balances: HashMap<String, i64>,
}

impl Projection {
    /// Applies `posting`, or clears the model and answers `false` when the posting belongs to a
    /// newer revision, which the model has to be rebuilt for from the start of the log.
    fn apply(&mut self, posting: &Posting) -> bool {
        if posting.revision > self.revision {
            self.revision = posting.revision;
            self.balances.clear();
            return false;
        }
        *self.balances.entry(posting.account.clone()).or_default() += posting.amount_cents;
        true
    }
}

/// The service's state: one read model per subscription that builds one.
#[derive(Clone, Default)]
struct Projections {
    balances: Arc<Mutex<Projection>>,
    spend: Arc<Mutex<Projection>>,
}

从最旧的记录开始​

余额保存在内存中，所以重启之后要从整个流重新构建。

src/main.rs
#[subscriber(
    KinesisStream::new("ledger").create_if_missing(1),
    start_at(KinesisPosition::horizon())
)]
async fn project_balance(
    posting: &Posting,
    ctx: &mut Context<'_, KinesisContext, Projections>,
    Ctx(seeker): Ctx<SeekHandle>,
) -> HandlerOutcome {
    let applied = ctx
        .state()
        .balances
        .lock()
        .unwrap_or_else(PoisonError::into_inner)
        .apply(posting);
    if applied {
        tracing::info!(posting.id, "balance updated");
        return HandlerOutcome::ack();
    }
    tracing::info!(
        posting.id,
        posting.revision,
        "new revision, rebuilding the balances"
    );
    if seeker.seek(KinesisPosition::horizon()).await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

start_at(KinesisPosition::horizon()) 在每个分片的裁剪起点（trim horizon），即流仍保留的最旧记录处，打开该分片。它在每次启动时生效，作用于订阅开始时打开的分片，并优先于任何已存储的检查点。不用 start_at 时，分片从它的检查点之后继续，没有检查点时在最新处打开。

KinesisPosition 有四种位置：

位置订阅从哪里读取
horizon()每个分片最旧的记录
latest()每个分片的最新处，因此只读取新记录
timestamp(ms)每个分片中在 ms 或之后到达的第一条记录
sequence(shard, seq)某个分片中的某一条记录，以及它之后的记录

流默认保留记录 24 小时，使用延长保留期最长可达 365 天。超过保留期的记录不复存在，裁剪起点也随之前移。

在处理器中 seek​

同一个处理器在收到更新修订的记账时重建余额。Ctx<SeekHandle> 把处理器所在订阅的 KinesisSeeker 交给它。seek 把订阅移到最旧的记录，处理器再读一遍每条记账。触发重建的那条记账也会回来，这一次投影会应用它。

seek 可以使用上面任何一种位置。针对整个流的位置，会移动本实例持有的每个分片。Ctx<Position> 保存投递自身的位置，即一个 sequence 位置：seek 到那里，会从这条记录开始重新读取该分片，不移动其他任何分片。这要求本实例持有该分片；seek 到另一个实例正在读取的分片或已读完的分片，会返回错误。

在批处理器中 seek​

卡的记账数量很大，所以消费额以最多 500 条的批构建。

src/main.rs
#[subscriber(KinesisStream::new("card-postings").create_if_missing(1))]
async fn project_spend(
    postings: &[Posting],
    ctx: &mut Context<'_, KinesisBatchContext, Projections>,
) -> HandlerOutcome {
    let applied = {
        let mut spend = ctx
            .state()
            .spend
            .lock()
            .unwrap_or_else(PoisonError::into_inner);
        postings.iter().all(|posting| spend.apply(posting))
    };
    if applied {
        tracing::info!(postings = postings.len(), "card spend updated");
        return HandlerOutcome::ack();
    }
    tracing::info!("new revision, rebuilding the card spend");
    if ctx
        .context(SeekHandle)
        .seek(KinesisPosition::horizon())
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

批处理器声明 KinesisBatchContext，而不是单次投递的上下文。它带有同样的 SeekHandle 键，但没有位置，因为一批跨越许多记录。需要某一条记录位置的处理器主体，读取它的 kinesis-shard-id 和 kinesis-sequence-number 消息头。

从某个时刻开始​

付款审查在每次启动时读取最近一天的付款。

src/main.rs
/// A payout as the payments service publishes it.
#[derive(Debug, PartialEq, Deserialize)]
struct Payout {
    id: u64,
    amount_cents: i64,
}

#[subscriber(KinesisStream::new("payouts").create_if_missing(1))]
async fn review_payout(payout: &Payout) -> HandlerOutcome {
    tracing::info!(payout.id, payout.amount_cents, "payout reviewed");
    HandlerOutcome::ack()
}

这个时刻要到服务启动时才知道，所以由挂载点而不是属性指定位置。

src/main.rs
/// Milliseconds since the epoch, the unit of a Kinesis arrival timestamp.
fn epoch_millis(at: SystemTime) -> u64 {
    at.duration_since(UNIX_EPOCH).map_or(0, |since| {
        u64::try_from(since.as_millis()).unwrap_or(u64::MAX)
    })
}

fn service(broker: KinesisBroker, review_from: SystemTime) -> impl App<State = Projections> {
    RustStream::new(AppInfo::new("ledger", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Projections::default()))
        .with_broker(broker, |b| {
            b.include(project_balance);
            b.include(project_spend.batch(nonzero!(500usize)));
            b.include(
                review_payout.start_at(KinesisPosition::timestamp(epoch_millis(review_from))),
            );
        })
}

/// The review reads the payouts of the last day on every start, the stream's default retention.
const REVIEW_WINDOW: Duration = Duration::from_hours(24);

#[ruststream::app]
fn app() -> impl App<State = Projections> {
    let review_from = SystemTime::now() - REVIEW_WINDOW;
    service(
        KinesisBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
        review_from,
    )
}

KinesisPosition::timestamp 接收自纪元起的毫秒数。Kinesis 把它与自己给每条记录打上的到达时间戳相比较。这个时间戳是近似值，所以在同一时刻前后到达的记录无法精确区分。

seek 与检查点​

seek 会清除它移动的每个分片为检查点所做的跟踪。在 seek 之前读取的投递，结算后不产生任何效果，所以触发重建的那条记账的 ack 不推进任何检查点。因此，之后的检查点绝不会越过一条被 seek 重放、但还没有处理器处理过的记录。seek 只移动本实例：其他实例继续读取它们持有的分片，接管某个分片的实例从它的检查点继续。

示例文件中的测试在进程内运行应用。进程内 Kinesis 保存每个流的完整日志，所以 seek 在那里同样会重新读取它。