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

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

FIFO 队列

本指南按消息组向 SQS FIFO 队列发送，用重复数据删除 id 丢弃重复的发送，并同时处理多个组，每个组保持顺序。

FIFO 队列在一个消息组内保持消息的顺序，只有前一条消息被删除后，才交出该组的下一条消息。 每次发送还带有一个重复数据删除 id，队列在五分钟内丢弃带相同 id 的第二次发送（AWS：重复数据删除）。 名称以 .fifo 结尾的队列就是 FIFO 队列。

示例是一个发货服务。 它把每个订单变成一次发货，同一客户的发货按客户下单的顺序发出。

在组中发送​

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

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "shipments.fifo")]
struct Shipment {
    order: u64,
    customer: String,
}

#[derive(OutSlot)]
#[publishes(Shipment)]
struct Shipments;

#[subscriber(SqsQueue::new("orders").create_if_missing())]
async fn dispatch(
    order: &Order,
    Out(shipments): Out<impl Publisher<Options = SqsPublishOptions>, Shipments>,
) -> HandlerOutcome {
    let shipment = Shipment {
        order: order.id,
        customer: order.customer.clone(),
    };
    // A customer's shipments form one group, so they are handled in the order they were sent.
    // The order id is the deduplication id, so sending the same order again ships nothing more.
    let sent = shipments
        .message(&shipment)
        .group_id(&order.customer)
        .deduplication_id(order.id.to_string())
        .publish()
        .await;
    if sent.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

dispatch 通过 Shipments 槽把一次发货发布到 shipments.fifo。 这个槽以 SqsPublishOptions 为约束，所以发布可以使用 SqsPublishSteps 的步骤。 使用这些步骤的处理器要导入本 crate 的 prelude。

group_id 把这次发货放进客户的组。 发布没有指定组时，组取自消息的 PARTITION_KEY_HEADER，其次取自策略用 Publish::default().group_id(..) 固定的组，最后是 "default"。 这个消息头是每个 RustStream Broker 都读取的写法，所以设置它的生产者不需要任何 SQS 专用步骤。

deduplication_id 是订单的 id。 orders 把一个订单投递两次时，第二次发货就是一次被队列丢弃的发送。 没有指定重复数据删除 id 的发布会得到一个在本进程内唯一的 id，因此有意发送的两个相同载荷都会到达。 策略不携带重复数据删除 id：一个固定的 id 会让第一次之后的每次发送都被丢弃。

在标准队列上设置组或重复数据删除 id 会被拒绝，返回 SqsError::NotFifo。

同时处理多个组​

src/main.rs
/// Ships one shipment. Four groups are handled at once, each group in its own order.
#[subscriber(SqsQueue::new("shipments.fifo").create_if_missing(), workers(4, by_key))]
async fn ship(shipment: &Shipment, ctx: &mut Context<'_>) -> HandlerOutcome {
    let group = ctx
        .headers()
        .get_str(PARTITION_KEY_HEADER)
        .unwrap_or_default();
    tracing::info!(shipment.order, group, "shipment sent");
    HandlerOutcome::ack()
}

来自 FIFO 队列的投递在 PARTITION_KEY_HEADER 中携带它的组。 workers(4, by_key) 运行四个工作分区，并把一个组的每条消息都送到同一个工作分区。 所以四个客户同时发货，而同一客户的发货一个接一个地发出。 一次接收可能返回一个组的多条消息，普通的 workers(4) 会让它们并行运行。

应用​

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(
        SqsBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
        |b| {
            b.include(dispatch)
                .out(Shipments, Publish::default())
                .build();
            b.include(ship);
        },
    )
}

create_if_missing 把 shipments.fifo 创建为启用基于内容的重复数据删除的 FIFO 队列。 FIFO 队列对应 FIFO 死信队列；重试与死信队列介绍次数上限。 SQS 限制单个组的吞吐量，所以服务把负载分散到许多组上。

示例文件中的测试检查一次发货所在的组，把一个订单发送两次，并同时发送两个客户的订单。 测试 SQS 服务介绍测试工具。