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

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

重试与死信

本指南立即或稍后重试一次投递，限定尝试次数，并把一直失败的消息存入死信流，适用于流消费者、Core subject 和模式。

NATS 没有死信机制。 JetStream 可以限定一条消息的投递次数，但不会把最后一次投递发往任何地方。 因此由挂载点声明上限和死信 subject，框架把它们应用到两种模型上。

示例是一个支付服务，它通过银行结清付款并发出退款。 它把死信保存在一个专用的流中：

nats stream add PAYMENTS --subjects 'payments.settle' --defaults
nats stream add DEAD --subjects 'dead.*' --defaults

在流上重试​

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

/// The outcome a refusal of the bank calls for.
fn outcome(refusal: Refusal) -> HandlerOutcome {
    match refusal {
        // Past the rate limit: a call right now would be refused again.
        Refusal::Busy => HandlerOutcome::retry_after(BACKOFF),
        // No answer: try again at once, up to the attempt cap.
        Refusal::Unavailable => HandlerOutcome::retry(),
        Refusal::Declined => HandlerOutcome::drop(),
    }
}

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

#[subscriber(JetStreamSubject::new("payments.settle", "PAYMENTS").durable("settler"))]
async fn settle(payment: &Payment, State(bank): State<Arc<Bank>>) -> HandlerOutcome {
    match bank.charge(&payment.card).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(refusal) => {
            tracing::warn!(payment.id, ?refusal, "payment refused by the bank");
            outcome(refusal)
        }
    }
}

outcome 把银行的拒绝变成处理器的结果。 银行繁忙时，retry_after 在 30 秒后再问一次。 在 JetStream 消费者上，延迟放在否定确认里：服务器扣留这笔付款，然后在同一个消费者上以相同的流序号再次投递它。 不发布任何东西，在此期间停止的进程也不会丢失任何东西。 银行没有应答时，retry 立即再问一次；卡被拒时，付款被丢弃。

上限与死信​

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("payments", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(NatsBroker::new("nats://localhost:4222"), |b| {
            // Five deliveries of one payment, the first included; then the dead-letter subject.
            b.include(settle)
                .max_attempts(nonzero!(5u32))
                .dead_letter("dead.payments");
            b.include(review);
            // A pattern is no subject to publish a copy to, so the mount site names one.
            b.include(send_refund)
                .max_attempts(nonzero!(3u32))
                .dead_letter("dead.refunds")
                .out_retry(Publish)
                .to("refunds.retry");
        })
}

max_attempts 统计一笔付款的每次投递，第一次也算在内。 在 JetStream 消费者上，这个次数就是服务器自己统计的次数，所以耗尽的 ack_wait 也计算在内。 第五次投递不再重试：dead_letter 把它发布到 dead.payments，载荷和消息头保持到达时的样子。 被丢弃的付款完成结算，不会去任何地方，所以死信中只有用完了尝试次数的付款。 设了上限却没有死信 subject 的消费者，会终止它的最后一次投递。

dead.payments 是一次普通发布，而 Core NATS 不保存无人读取的消息。 DEAD 流保存它，所以死信会一直等到有人查看。 应用读取的任何 subject 都不与死信 subject 匹配，否则用完尝试次数的投递会回到放弃了它的处理器。

读取死信​

src/main.rs
#[subscriber(JetStreamSubject::new("dead.payments", "DEAD").durable("review"))]
async fn review(payment: &Payment) -> HandlerOutcome {
    tracing::error!(payment.id, "payment needs a person to look at it");
    HandlerOutcome::ack()
}

review 像读取其他流一样从 DEAD 流读取失败的付款，并把每一笔记入日志，留给人来查看。

在 Core subject 上重试​

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

#[subscriber(CoreWildcard::new("refunds.*"))]
async fn send_refund(refund: &Refund, State(bank): State<Arc<Bank>>) -> HandlerOutcome {
    match bank.charge(&refund.card).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(refusal) => {
            tracing::warn!(refund.id, ?refusal, "refund refused by the bank");
            outcome(refusal)
        }
    }
}

Core NATS 无法扣留消息。 在 Core 订阅上，一次重试就是框架发布的投递副本：retry 立即发布它，retry_after 在延迟结束时发布它。 副本在 x-ruststream-retry-count 消息头中带着投递次数，上限读取的就是它。 延迟的副本是至多一次的：在延迟结束前停止的进程会丢失它。

CoreSubject 把副本发布到它自己的 subject。 send_refund 读取模式 refunds.*，这个模式没有唯一的 subject 可供发布。 所以应用用 out_retry(Publish).to("refunds.retry") 指定一个模式能匹配的 subject，副本就会回到 send_refund。 基于模式却没有这一设置的注册拒绝启动，不论它的处理器是否重试。 退款的第三次投递进入 dead.refunds，次数记在消息头里。

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