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

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

确认与投递保证

本指南用处理器的结果结算 JetStream 投递，为慢处理器设定 ack_wait，处理投递了两次的消息，并说明 Core NATS 不保证什么。

处理器用它返回的值结算自己的投递，运行时在处理器结束后发送结算。 在 JetStream 消费者上，每种结果都是一个服务器据以行动的确认：

结果发给服务器服务器做什么
ack+ACK这个消费者不再保留这条消息
retry-NAK立即再次投递它
retry_after(delay)带延迟的 -NAK扣留它，延迟过后再次投递
drop+TERM不再投递它
没有结算什么也不发ack_wait 过后再次投递它

重新投递保留消息的流序号，服务器统计它的投递次数。

示例是一个支付服务。 它通过银行扣收已授权的付款，并把每次扣款通知给看板。

结算一笔付款​

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

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

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "payments.captured")]
struct Captured {
    payment: u64,
    amount_cents: u64,
}

#[subscriber(
    JetStreamSubject::new("payments.authorized", "PAYMENTS")
        .durable("capture")
        .ack_wait(NonZeroDuration::from_secs(nonzero!(60))),
    reply
)]
async fn capture(
    payment: &Payment,
    Ctx(delivered): Ctx<Delivered>,
    State(bank): State<Arc<Bank>>,
) -> Result<Captured, HandlerOutcome> {
    if delivered.is_some_and(|count| count > 1) {
        tracing::info!(payment.id, ?delivered, "payment delivered again");
    }
    match bank.capture(payment).await {
        Ok(()) => Ok(Captured {
            payment: payment.id,
            amount_cents: payment.amount_cents,
        }),
        // Past the rate limit: the server holds the payment and delivers it after the delay.
        Err(Refusal::Busy) => Err(HandlerOutcome::retry_after(BACKOFF)),
        // No answer: the server delivers the payment again at once.
        Err(Refusal::Unavailable) => Err(HandlerOutcome::retry()),
        // The server never delivers a terminated payment again.
        Err(Refusal::Declined) => {
            tracing::warn!(payment.id, "card declined, payment terminated");
            Err(HandlerOutcome::drop())
        }
    }
}

银行有四种答复，每种需要不同的结果。 扣款成功时返回 Captured，运行时把它作为回复发布，然后确认这笔付款。 Busy 表示银行的速率限额已经用完，所以付款在服务器上等待 30 秒。 Unavailable 表示银行没有应答，所以付款立即回来。 Declined 表示这张卡无法扣款，再试一次也是同样的答复，所以这笔付款被终止。 不设上限的 retry 会无休止地回来；重试与死信加上上限和死信 subject。

投递了两次的消息​

JetStream 至少投递一次。 在 retry 之后、ack_wait 耗尽之后，以及进程在结算前停止之后，付款都会回来。 所以处理器会两次看到某些付款，它的工作必须能接受重复。 这里银行把付款 id 当作幂等键，重复的扣款只扣一次。 Ctx<Delivered> 是服务器统计的投递次数，第一次投递时为 1；capture 会记录它再次看到的付款。

慢处理器与 ack_wait​

服务器等待结算的时间是 ack_wait，描述符未另行指定时为 30 秒。 超时后它再次投递消息，而第一个处理器可能仍在运行。 capture 设为 60 秒，高于银行答复的最长时间。 把 ack_wait 设得高于处理器的最坏耗时，两次投递就永远不会重叠。 耗尽的 ack_wait 也计入尝试上限的投递次数。

Core NATS 上的看板​

src/main.rs
#[subscriber("payments.captured")]
async fn refresh(captured: &Captured) -> HandlerOutcome {
    tracing::info!(captured.payment, "dashboard refreshed");
    HandlerOutcome::ack()
}

refresh 从一个 Core NATS subject 读取扣款。 Core NATS 至多投递一次，并且没有确认。 结算 Core 投递会报告 AckError::Unsupported，运行时把它记录为警告。 进程丢失的 Core 消息就此消失，没有任何东西能把它找回来。 看板可以丢失一次更新，因为下一次扣款会再次刷新它。 付款不能丢失，所以付款经由流传递。

应用​

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| {
            b.include(capture);
            b.include(refresh);
        })
}

银行客户端是应用的状态，在启动时打开一次。 批处理器用一个结果结算整批；返回 Vec<HandlerOutcome> 时，则逐条结算每条消息。

示例文件中的测试在暂停的时钟上于进程内运行应用，所以 Busy 的 30 秒瞬间就过去了。