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

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

重放与 seek

本指南从头或从某个时刻重新读取主题，并在处理器内部移动正在运行的订阅。

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

读模型​

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

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 Entry {
    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 `entry`, or clears the model and answers `false` when the entry belongs to a newer
    /// revision, which the model has to be rebuilt for from the start of the log.
    fn apply(&mut self, entry: &Entry) -> bool {
        if entry.revision > self.revision {
            self.revision = entry.revision;
            self.balances.clear();
            return false;
        }
        *self.balances.entry(entry.account.clone()).or_default() += entry.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(
    KafkaTopic::new("ledger").group("balances").commit(Commit::Tracked),
    start_at(KafkaPosition::earliest())
)]
async fn project_balance(
    entry: &Entry,
    ctx: &mut Context<'_, KafkaContext, Projections>,
    Ctx(seeker): Ctx<SeekHandle>,
) -> HandlerOutcome {
    let applied = ctx
        .state()
        .balances
        .lock()
        .unwrap_or_else(PoisonError::into_inner)
        .apply(entry);
    if applied {
        tracing::info!(entry.id, "balance updated");
        return HandlerOutcome::ack();
    }
    tracing::info!(
        entry.id,
        entry.revision,
        "new revision, rebuilding the balances"
    );
    if seeker.seek(KafkaPosition::earliest()).await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

start_at(KafkaPosition::earliest()) 在每次启动时都从日志开头打开订阅，不管组 balances 之前提交过什么。 描述符上的 StartOffset 与此不同：它只在组没有已提交的偏移量时生效。 这个位置落在组分配给该消费者的分区上，位于它们的第一条记录之前。

KafkaPosition 有五种位置：

位置订阅从哪里继续
earliest()Kafka 保留的最旧记录
latest()日志末尾，因此只读取新记录
offset(partition, n)某个分区的偏移量 n
topic_offset(topic, partition, n)某个主题某个分区的偏移量 n，用于读取多个主题的订阅
timestamp(ms)每个分区中时间戳不早于 ms 的第一条记录

起始位置如果没有指定订阅的任何一个分区，启动就会失败。 在 Assignment::CooperativeSticky 下，第一次分配可能只是组份额的一部分，之后增量到达的分区不会被移动。

在处理器中 seek​

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

seek 可以使用上面任何一种位置。 Ctx<Position> 保存投递自身的位置：seek 到那里，会重新读取这条记录及其后的一切。 Ctx<Partition> 和 Ctx<Offset> 为 offset(..) 位置提供坐标。

在批处理器中 seek​

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

src/main.rs
#[subscriber(KafkaTopic::new("card-entries")
        .group("card-spend")
        .commit(Commit::Tracked))]
async fn project_spend(
    entries: &[Entry],
    ctx: &mut Context<'_, KafkaBatchContext, Projections>,
) -> HandlerOutcome {
    let applied = {
        let mut spend = ctx
            .state()
            .spend
            .lock()
            .unwrap_or_else(PoisonError::into_inner);
        entries.iter().all(|entry| spend.apply(entry))
    };
    if applied {
        tracing::info!(entries = entries.len(), "card spend updated");
        return HandlerOutcome::ack();
    }
    tracing::info!("new revision, rebuilding the card spend");
    if ctx
        .context(SeekHandle)
        .seek(KafkaPosition::earliest())
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

批处理器声明 KafkaBatchContext，而不是单次投递的上下文。 它带有同样的 SeekHandle 键，但没有位置，因为一批跨越许多记录。 整批结算之后，订阅才会移动。

从某个时刻开始​

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

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

#[subscriber(KafkaTopic::new("payouts")
        .group("payout-review")
        .commit(Commit::Tracked))]
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 Kafka stamps records with.
fn epoch_millis(at: SystemTime) -> i64 {
    at.duration_since(UNIX_EPOCH).map_or(0, |since| {
        i64::try_from(since.as_millis()).unwrap_or(i64::MAX)
    })
}

fn service(broker: KafkaBroker, 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(KafkaPosition::timestamp(epoch_millis(review_from))));
        })
}

/// The review reads the payouts of the last seven days on every start.
const REVIEW_WINDOW: Duration = Duration::from_hours(7 * 24);

#[ruststream::app]
fn app() -> impl App<State = Projections> {
    let review_from = SystemTime::now() - REVIEW_WINDOW;
    service(KafkaBroker::new(["localhost:9092"]), review_from)
}

KafkaPosition::timestamp 接收自纪元起的毫秒数，这正是 Kafka 给记录打时间戳所用的单位。 Kafka 在每个分区中把这个时刻解析为时间戳不早于它的第一条记录。 没有这种记录的分区从末尾继续。

seek 对组的影响​

seek 移动的是这个消费者，而不是整个组。 它移动该成员持有的分区，不替任何人提交任何东西。 组的其他成员继续在原处读取。 下一次再均衡会抛弃这次 seek：届时分到某个分区的成员，从组已提交的偏移量继续。 start_at 位置在每次启动时都会重新生效。

Commit::Tracked 下的 seek​

三个订阅都以 Commit::Tracked 提交（提交与投递保证）。 seek 会清除它移动的每个分区的跟踪位置。 在 seek 之前拉取的投递，结算后不产生任何效果，所以触发重建的那条记账的 ack 不提交任何东西。 因此，之后的提交绝不会越过一条被 seek 重放、但还没有处理器处理过的记录。