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

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

重试与死信

本指南立即或稍后重试一次投递，限定尝试次数，把一直失败的消息转入死信地址，并标记副本。

AMQP 1.0 没有延迟重新投递，链路也无法声明投递上限或死信地址。 这两件事都由框架来做：挂载点声明上限和死信地址，延迟重试以消息副本的形式发出，消息头里带着重试次数。

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

订阅者​

处理商有三种拒绝退款的方式，每种需要不同的结果。 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(AmqpAddress::queue("refunds"))]
async fn process_refund(refund: &Refund) -> HandlerOutcome {
    match send_refund(refund) {
        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），其值加一。 延迟的副本是至多一次的：在延迟结束前停止的进程会丢失它。 在队列上，副本像其他消息一样交给一个消费者。

上限与死信地址​

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("refunds", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            // Five deliveries of one refund, the first included; then the dead-letter address.
            b.include(process_refund)
                .max_attempts(nonzero!(5u32))
                .dead_letter("refunds.dead")
                .out_retry(Publish)
                .transform(Retried);
            b.include(review);
        },
    )
}

max_attempts 统计一笔退款的每次投递，第一次也算在内。 第五次投递不再重试：dead_letter 把它按到达时的载荷和消息头原样发布到 refunds.dead。 被丢弃的退款会被拒绝，由 Broker 自己的死信策略处理，而不是进入 refunds.dead。

在声明了上限时，框架从投递中读取计数。 消息带有 AMQP header 段时，计数就是 Broker 的 delivery-count。 本 crate 发布的消息不带这个段，所以这时 retry 会立即发布一份副本，由重试次数消息头统计尝试次数。

有上限而没有 dead_letter 时，最后一次投递会被拒绝，由 Broker 的死信策略决定去向。 Broker 也有自己的上限：Artemis 在 max-delivery-attempts 次（默认 10 次）之后停止投递。 上限只应由一方负责：低于 Broker 上限的上限会让框架的计数生效。

读取死信​

src/main.rs
#[subscriber(AmqpAddress::queue("refunds.dead"))]
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()
}

refunds.dead 是一个普通队列，review 像读取其他队列一样读取它。 死信的重试次数消息头记录了它经历的投递次数，这里是 5。 review 用 ctx.headers() 读取这个值，并记录这笔退款，留给人工查看。

标记副本​

src/main.rs
/// Marks every copy with the address its delivery came from.
struct Retried;

impl<C, Options> PublishTransform<ForReply<C>, Options> for Retried {
    type Destination = Reads;

    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        cx: &PublishContext<'_, C>,
    ) {
        out.headers_mut()
            .insert("x-retried-from", cx.name().to_owned());
    }
}

副本在处理器中没有调用点，所以只有发布转换能触及它。 应用用 .out_retry(Publish).transform(Retried) 挂载 process_refund。 .out_retry(Publish) 指明副本的发布者，其后的步骤作用于副本。 Retried 把投递的来源地址写入 x-retried-from。 每份副本都回到其订阅的地址，所以这个位置上的转换不能改变目的地。

主题上的重试​

副本发布到订阅自己的地址。 在主题地址上，这个地址是共用的，所以主题上的每个订阅都会收到副本，而不只是要求重试的那一个。 消息只归一个读者所有时，在队列上重试。

测试延迟​

示例文件中的测试在暂停的时钟下于进程内运行应用，所以 tb.advance(BACKOFF) 让 30 秒立即过去。 测试 AMQP 服务介绍测试套件。