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

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

确认与投递保证

本指南说明处理器的每种结果对 SQS 消息做什么，消息到达两次时处理器如何保持正确，以及批如何结算。

SQS 至少投递一次。 收到的消息留在队列中，保持不可见，直到消费者删除它。 标准队列还可能把一条消息投递两次，顺序也可能与发送顺序不同（AWS：至少一次投递）。

示例是一个付款服务，它把付款记入账本，并支付退款。

每种结果做什么​

每种结果都是对处理器收到的那条消息的一次 SQS 调用：

结果SQS 调用消息会怎样
ackDeleteMessage离开队列
dropDeleteMessage离开队列：SQS 没有拒绝消息又保留它的调用
retryChangeMessageVisibility 设为零立即再次交出
retry_afterChangeMessageVisibility 设为延迟时间延迟结束时再次交出

处理器从未结算的消息不会产生调用，比如进程在中途停止时。 它在可见性超时结束时回来。 所以处理器的工作发生在结果之前，消息只在工作完成之后才删除。

只记账一次的处理器​

src/main.rs
/// How long a payment on an account under review waits before the next try.
const REVIEW: Duration = Duration::from_secs(60);

/// The receive after which a payment under review is given up.
const LAST_RECEIVE: u32 = 3;

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

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "ledger-entries")]
struct Entry {
    payment: u64,
    account: String,
    amount_cents: u64,
}

#[derive(OutSlot)]
#[publishes(Entry)]
struct Entries;

/// The ids of the payments already booked.
#[derive(Debug, Default, Clone)]
struct Ledger {
    booked: Arc<Mutex<HashSet<u64>>>,
}

#[subscriber(SqsQueue::new("payments"))]
async fn book(
    payment: &Payment,
    ctx: &mut Context<'_, (), Ledger>,
    Out(entries): Out<impl Publisher, Entries>,
) -> HandlerOutcome {
    // SQS counts the receives of a message, this one included.
    let receives: u32 = ctx
        .headers()
        .get_str(RECEIVE_COUNT_HEADER)
        .and_then(|count| count.parse().ok())
        .unwrap_or(1);
    if payment.under_review {
        if receives >= LAST_RECEIVE {
            tracing::error!(payment.id, receives, "payment dropped, still under review");
            return HandlerOutcome::drop();
        }
        return HandlerOutcome::retry_after(REVIEW);
    }
    let booked = Arc::clone(&ctx.state().booked);
    if booked
        .lock()
        .unwrap_or_else(PoisonError::into_inner)
        .contains(&payment.id)
    {
        tracing::info!(payment.id, receives, "payment already booked");
        return HandlerOutcome::ack();
    }
    let entry = Entry {
        payment: payment.id,
        account: payment.account.clone(),
        amount_cents: payment.amount_cents,
    };
    if entries.message(&entry).publish().await.is_err() {
        // Nothing is booked yet: the payment comes back and the work runs again.
        return HandlerOutcome::retry();
    }
    booked
        .lock()
        .unwrap_or_else(PoisonError::into_inner)
        .insert(payment.id);
    tracing::info!(payment.id, "payment booked");
    HandlerOutcome::ack()
}

book 为一笔付款写入一条账本记录，然后确认它。 SQS 第二次投递的付款会在已记账的 id 中被找到，于是直接确认，不再写第二条记录。 应用状态中的集合代替了服务保存已记账 id 的存储。这里的 id 是付款自己的 id，而不是 SQS 消息 id，因为发送两次的生产者发出的是两条消息。 记录无法发布时，什么也不记账，retry 把付款交还，再试一次。

处理器读取 RECEIVE_COUNT_HEADER，即队列自己的 ApproximateReceiveCount。 它也计入本次接收，所以第一次投递读到 1。 审核中账户的付款用 retry_after 等待一分钟，到第三次接收时，处理器放弃并丢弃它。 重试与死信队列展示队列本身如何限制接收次数并保留失败的消息。

批逐条消息结算​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "refunds")]
struct Refund {
    payment: u64,
    amount_cents: u64,
}

/// The largest refund paid without a person's approval.
const APPROVAL_LIMIT_CENTS: u64 = 50_000;

/// How long a refund waits for approval before it is looked at again.
const APPROVAL: Duration = Duration::from_secs(300);

/// Settles each refund of a batch on its own: a small refund is paid and deleted, a large one
/// waits on the queue for its approval.
#[subscriber(SqsQueue::new("refunds"))]
async fn settle_refunds(refunds: &[Refund]) -> Vec<HandlerOutcome> {
    refunds
        .iter()
        .map(|refund| {
            if refund.amount_cents > APPROVAL_LIMIT_CENTS {
                tracing::info!(refund.payment, "refund waits for approval");
                HandlerOutcome::retry_after(APPROVAL)
            } else {
                tracing::info!(refund.payment, refund.amount_cents, "refund paid");
                HandlerOutcome::ack()
            }
        })
        .collect()
}

批处理器为每条消息返回一个结果，顺序与切片相同。 每个结果都是对其消息的一次单独调用。 这里小额退款被删除，大额退款留在队列中，五分钟后单独回来。 只返回一个结果的批处理器用它结算批中的每条消息。

应用​

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = Ledger> {
    RustStream::new(AppInfo::new("payments", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Ledger::default()))
        .with_broker(
            SqsBroker::new()
                .endpoint("http://localhost:4566")
                .region("us-east-1")
                .test_credentials(),
            |b| {
                b.include(book).out(Entries, Publish::default()).build();
                b.include(settle_refunds.batch(nonzero!(10)));
            },
        )
}

book 通过 Entries 槽发布它的记录，这个槽在挂载处绑定到队列发布者。 FIFO 队列在五分钟内丢弃带相同重复数据删除 id 的第二次发送，这能防止生产者把一条消息发送两次；FIFO 队列展示了这一点。 标准队列上的处理器仍然要自己记录做过的事。

示例文件中的测试把一笔付款投递两次，在暂停的时钟上让审核中的付款用尽接收次数，并在一批中发送三笔退款。 测试 SQS 服务介绍测试工具。