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

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

重试与死信主题

本指南立即或稍后重试一条 Kafka 记录，限定尝试次数，并把一直失败的记录转入死信主题，适用于单个主题或多个主题。

Kafka 既不扣留记录，也不统计投递次数。 这两件事都由框架来做：挂载点声明上限和死信主题，重试以记录副本的形式传递，消息头里带着重试次数。

示例是一个退款服务，它通过发卡处理商发出退款。

订阅者​

处理商有三种拒绝退款的方式，每种需要不同的结果。 Busy 表示它的速率限额已经用完：立即重试仍会被拒绝，所以 retry_after 等待 30 秒。 Unavailable 表示它没有应答：retry 立即再试一次。 Declined 表示卡账户已经关闭，再试一次也是同样的答复，所以这笔退款被丢弃。

src/main.rs
/// How long a busy processor asks its callers to wait.
const BACKOFF: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Refund {
    id: u64,
    card: String,
    amount_cents: u64,
}

#[subscriber(KafkaTopic::new("refunds").commit(Commit::Tracked))]
async fn process_refund(
    refund: &Refund,
    State(processor): State<Arc<Processor>>,
) -> HandlerOutcome {
    settle(&processor, refund).await
}

async fn settle(processor: &Processor, refund: &Refund) -> HandlerOutcome {
    match processor.refund(refund).await {
        Ok(()) => HandlerOutcome::ack(),
        // Past the rate limit: a delivery right now would be refused again.
        Err(Refusal::Busy) => HandlerOutcome::retry_after(BACKOFF),
        // No answer: try again at once, up to the attempt cap.
        Err(Refusal::Unavailable) => HandlerOutcome::retry(),
        Err(Refusal::Declined) => {
            tracing::warn!(refund.id, "card declined, refund dropped");
            HandlerOutcome::drop()
        }
    }
}

retry_after 结算这条记录，并在延迟过去后把一份副本发布回 refunds。 副本带着重试次数消息头（RETRY_COUNT_HEADER），其值加一。 延迟的副本是至多一次的：在延迟结束前停止的进程会丢失它。 副本是位于主题末尾的一条新记录，所以排在这笔退款后面的记录不必等它。

在声明了上限时，retry 也会变成一份副本，并立即发布。 没有上限时，它在 Commit::Tracked 下把记录交还给订阅，提交与投递保证展示了这一点。

上限与死信主题​

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("refunds", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(
            KafkaBroker::new(["localhost:9092"]).default_group("refunds"),
            |b| {
                // Five deliveries of one refund, the first included; then the dead-letter topic.
                b.include(process_refund)
                    .max_attempts(nonzero!(5u32))
                    .dead_letter("refunds.dlq");
                b.include(review);
                // A set of topics names where a copy goes: back to the topic it came from.
                b.include(process_regional_refund)
                    .max_attempts(nonzero!(5u32))
                    .dead_letter("refunds.dlq")
                    .out_retry(Publish::default())
                    .transform(ToSourceTopic);
            },
        )
}

max_attempts 统计一笔退款的每次投递，第一次也算在内。 第五次投递不再重试：dead_letter 把它发布到 refunds.dlq，载荷和消息头保持到达时的样子。 被丢弃的退款完成结算，不会去任何地方；死信主题只存放用完了尝试次数的退款。 重试主题和死信主题是服务的基础设施：像其他主题一样创建它们，框架只负责向它们发布。

读取死信主题​

死信主题是一个普通主题，review 像读取其他主题一样读取它。

src/main.rs
#[subscriber(
    KafkaTopic::new("refunds.dlq")
        .group("refunds-review")
        .start(StartOffset::Earliest)
)]
async fn review(refund: &Refund, ctx: &mut Context<'_>) -> HandlerOutcome {
    let deliveries = ctx.headers().get_str(RETRY_COUNT_HEADER).unwrap_or("1");
    tracing::error!(refund.id, deliveries, "refund needs a person to look at it");
    HandlerOutcome::ack()
}

转入死信的记录，其重试次数消息头保存它经历的投递次数，这里是 5。 review 用 ctx.headers() 读取它，并把这笔退款记入日志，留给人来查看。

多个主题上的重试​

记录的副本发布到订阅读取的主题。 KafkaTopic 读取一个主题，所以不需要额外的设置。 KafkaTopics 和 KafkaPartitions 读取的是一个单次发布到达不了的集合，所以它们的注册要指定副本的去向，否则服务拒绝启动。

src/main.rs
#[subscriber(
    KafkaTopics::new(["refunds.eu", "refunds.us"])
        .group("regional-refunds")
        .commit(Commit::Tracked)
)]
async fn process_regional_refund(
    refund: &Refund,
    Ctx(topic): Ctx<Topic>,
    State(processor): State<Arc<Processor>>,
) -> HandlerOutcome {
    tracing::debug!(refund.id, %topic, "regional refund");
    settle(&processor, refund).await
}

应用用 .out_retry(Publish::default()).transform(ToSourceTopic) 挂载 process_regional_refund。 ToSourceTopic 把每份副本送回其投递所来自的主题，所以来自 refunds.us 的重试退款会回到 refunds.us。 它从 Kafka 上下文读取主题，所以处理器要接收一个上下文键，比如 Ctx<Topic>；在手写形式中，它的上下文类型是 KafkaContext。 .out_retry(Publish::default()).to("refunds.retry") 则把所有副本发到你自己的一个主题；把这个主题列入订阅，副本才会回来。 同样的规则也适用于从不重试的订阅：每个基于一组主题的注册都要指定重试目的地。

测试延迟​

示例文件中的测试在暂停的时钟上于进程内运行应用，所以 tb.advance(BACKOFF) 让 30 秒瞬间过去。 测试 Kafka 服务讲解了测试工具。