消费队列

本指南读取 SQS 队列：长轮询、慢处理器的可见性超时、并行处理器、批，以及一个处理器读取多个队列。

示例是一个仓库服务。它打包订单、记账付款，并记录订单的变更。

队列及其长轮询

宏
手写

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
    items: u32,
}

/// Packs an order. Packing takes up to a minute, so each receive asks for a minute of
/// visibility, and four orders are packed at once.
#[subscriber(
    SqsQueue::new("orders")
        .visibility(Duration::from_secs(60))
        .create_if_missing(),
    workers(4)
)]
async fn pack(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, order.items, "order packed");
    HandlerOutcome::ack()
}

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Order {
    id: u64,
    items: u32,
}

impl OutgoingDestination for Order {
    type Form = FixedName;

    const DESTINATION: &'static str = "orders";
}

impl MessageHeaders for Order {
    type Contract = NoHeaders;
}

impl MessageInfo for Order {
    const NAME: &'static str = "Order";
}

/// Packs an order.
struct Pack;

impl Handle<Order> for Pack {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, order.items, "order packed");
        ready(Ok(()))
    }
}

订阅用 ReceiveMessage 对它的队列做长轮询。每次调用最长等待 20 秒，这是 SQS 允许的上限。 wait 设置更短的等待时间。队列可以用名称或 URL 指定，名称在启动时用 GetQueueUrl 解析一次。

逐条处理的订阅每次调用仍然请求最多十条消息，因为 SQS 按请求计费。它把这些消息逐条交给处理器。处理器工作时，一个后台任务接收下一批，它最多领先一批。

可见性与慢处理器

收到的消息在队列的可见性超时内对其他消费者不可见。如果到那时还没有删除，SQS 会再次交出它。 visibility 在每次接收时请求自己的超时，范围是一秒到十二小时。 SQS 按整秒计算超时，所以不足一秒的部分向上取整。 pack 请求一分钟。不用这一步时，订阅在打开时读取一次队列自己的超时。

处理器可以运行得比超时更久。投递存活期间，本 crate 在每个超时过半时延长它的可见性，因此慢处理器的消息不会被交给另一个消费者。 SQS 从接收起最多隐藏消息十二小时。处理器运行这么久时，本 crate 把可见性最后一次延长到这十二小时结束，并记录一条警告日志。之后消息会交给另一个消费者。

并行处理器

workers(4) 同时运行最多四个 orders 的投递。四个处理器分享每次接收返回的消息。标准队列没有需要保持的顺序，所以不需要别的设置。 FIFO 队列用 workers(n, by_key) 让一个消息组保持顺序。

批

宏
手写

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

/// Books the payments one receive returned, in one write to the ledger.
#[subscriber(SqsQueue::new("payments"))]
async fn reconcile(payments: &[Payment]) -> HandlerOutcome {
    let total: u64 = payments.iter().map(|payment| payment.amount_cents).sum();
    tracing::info!(count = payments.len(), total, "payments booked");
    HandlerOutcome::ack()
}

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Payment {
    order: u64,
    amount_cents: u64,
}

impl OutgoingDestination for Payment {
    type Form = FixedName;

    const DESTINATION: &'static str = "payments";
}

impl MessageHeaders for Payment {
    type Contract = NoHeaders;
}

impl MessageInfo for Payment {
    const NAME: &'static str = "Payment";
}

/// Books the payments one receive returned, in one write to the ledger.
struct Reconcile;

impl Handle<[Payment]> for Reconcile {
    fn handle(
        &self,
        payments: &[Payment],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        let total: u64 = payments.iter().map(|payment| payment.amount_cents).sum();
        tracing::info!(count = payments.len(), total, "payments booked");
        ready(Ok(()))
    }
}

接收切片的处理器就是批处理器。挂载处指定的大小成为 MaxNumberOfMessages，因此一次接收就是一批。一批包含队列当时拥有的消息，什么也没取回的长轮询不产出批。 ReceiveMessage 最多返回十条消息，所以更大的大小会被降为十，服务为此记录一次日志。批中的每条消息各自结算；确认一页展示了各种结果。

一个处理器，多个队列

宏
手写

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Change {
    order: u64,
    by: String,
}

/// Records a change to an order. The definition fixes the kind of queue and leaves its name
/// to the mount site.
#[subscriber(SqsQueue)]
async fn audit(change: &Change) -> HandlerOutcome {
    tracing::info!(change.order, change.by, "change recorded");
    HandlerOutcome::ack()
}

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Change {
    order: u64,
    by: String,
}

impl OutgoingDestination for Change {
    type Form = CallerName;
}

impl MessageHeaders for Change {
    type Contract = NoHeaders;
}

impl MessageInfo for Change {
    const NAME: &'static str = "Change";
}

/// Records a change to an order, on whichever queue it is mounted.
struct Audit;

impl Handle<Change> for Audit {
    fn handle(
        &self,
        change: &Change,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(change.order, change.by, "change recorded");
        ready(Ok(()))
    }
}

#[subscriber(SqsQueue)] 固定来源的种类，队列名称留空。挂载处用 .name(..) 指定名称，每个队列一次。

应用

宏
手写

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("warehouse", "0.1.0")).with_broker(
        SqsBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
        |b| {
            b.include(pack);
            // Up to ten payments per call: one receive is one batch.
            b.include(
                reconcile
                    .batch(nonzero!(10))
                    .wait(Duration::from_secs(20))
                    .create_if_missing(),
            );
            // One definition, two queues.
            b.include(audit.name("audit-eu").create_if_missing());
            b.include(audit.name("audit-us").create_if_missing());
        },
    )
}

src/main.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("warehouse", "0.1.0")).with_broker(
        SqsBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
        |b| {
            // A minute of visibility per receive, four orders packed at once.
            b.include(
                subscriber("orders", Pack)
                    .map_source(|_| {
                        SqsQueue::new("orders")
                            .visibility(Duration::from_secs(60))
                            .create_if_missing()
                    })
                    .workers(nonzero!(4))
                    .build(),
            );
            // Up to ten payments per call: one receive is one batch.
            b.include(
                subscriber("payments", Reconcile)
                    .batch(nonzero!(10))
                    .map_source(|_| {
                        SqsQueue::new("payments")
                            .wait(Duration::from_secs(20))
                            .create_if_missing()
                    })
                    .build(),
            );
            // One handler, two queues.
            for queue in ["audit-eu", "audit-us"] {
                b.include(
                    subscriber(queue, Audit)
                        .map_source(|_| SqsQueue::new(queue).create_if_missing())
                        .build(),
                );
            }
        },
    )
}

fn main() -> ExitCode {
    run_main(app)
}

队列的步骤也可以写在挂载处，放在框架自己的步骤之后，由 SqsSubscription 提供。队列名称为空、等待时间超过 20 秒，或者可见性超出范围，都会让服务在启动时以 SqsError::InvalidQueue 停止，这发生在调用 AWS 之前。名称可以包含 SQS 不接受的字符：A-Z、a-z、0-9、_ 和 - 以外的每个字符都会变成 -，.fifo 后缀保留不变。

create_if_missing 创建不存在的队列。生产环境的服务通常从基础设施获得队列，不用这一步；这时缺失的队列会让服务在启动时停止。

投递携带什么

每个消息属性都作为投递的消息头到达，发布时的每个消息头都作为属性发出。有两个消息头是本 crate 自己的。 RECEIVE_COUNT_HEADER 携带队列对这条消息的接收计数。 PARTITION_KEY_HEADER 携带 FIFO 队列的消息组。 SQS 的正文是文本，所以二进制载荷以 base64 编码传输，带一个标记属性，到达时已经解码。 Base64 让这样的载荷增大三分之一，而 SQS 会拒绝超过大小上限的消息。

示例文件中的测试在进程内向应用发布订单、付款和变更。测试 SQS 服务介绍测试工具。

队列及其长轮询​

可见性与慢处理器​

并行处理器​

批​

一个处理器，多个队列​

应用​

投递携带什么​

队列及其长轮询

可见性与慢处理器

并行处理器

批

一个处理器，多个队列

应用

投递携带什么