Очереди FIFO

Это руководство отправляет сообщения в очередь SQS FIFO по группам сообщений, отбрасывает повторные отправки по идентификатору дедупликации и обрабатывает несколько групп одновременно, каждую по порядку.

Очередь FIFO сохраняет порядок сообщений внутри группы сообщений и выдаёт следующее сообщение группы только после того, как удалено предыдущее. Каждая отправка также указывает идентификатор дедупликации, и в течение пяти минут очередь отбрасывает повторную отправку с тем же идентификатором (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()
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Order {
    id: u64,
    customer: String,
}

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";
}

#[derive(Debug, PartialEq, Serialize, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Shipment {
    order: u64,
    customer: String,
}

impl OutgoingDestination for Shipment {
    type Form = FixedName;

    const DESTINATION: &'static str = "shipments.fifo";
}

impl MessageHeaders for Shipment {
    type Contract = NoHeaders;
}

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

struct Shipments;

impl OutSlot for Shipments {
    const NAME: &'static str = "Shipments";
    type Destination = Reads;
}

impl PublishedThrough<Shipments> for Shipment {}

struct Dispatch;

impl<Wire> Handle<Order, (), Outs<(Wire,)>> for Dispatch
where
    Wire: OutEntry<Shipments, Wire: Publisher<Options = SqsPublishOptions>>,
{
    async fn handle(
        &self,
        order: &Order,
        outs: &Outs<(Wire,)>,
        _ctx: &mut Context<'_>,
    ) -> Result<(), 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 = outs
            .get(Shipments)
            .message(&shipment)
            .group_id(&order.customer)
            .deduplication_id(order.id.to_string())
            .publish()
            .await;
        if sent.is_err() {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

dispatch публикует отправление в shipments.fifo через слот Shipments. Слот ограничен SqsPublishOptions, поэтому публикация принимает шаги SqsPublishSteps. Обработчик, который вызывает эти шаги, импортирует прелюдию этого крейта.

group_id помещает отправление в группу клиента. Если публикация не называет группу, группа берётся из заголовка PARTITION_KEY_HEADER сообщения, затем из группы, которую зафиксировала политика через Publish::default().group_id(..), и наконец это "default". Этот заголовок читает каждый брокер RustStream, поэтому отправителю, который его задаёт, шаг SQS не нужен.

deduplication_id - это идентификатор заказа. Когда orders доставляет заказ дважды, второе отправление - это отправка, которую очередь отбрасывает. Публикация, которая не называет идентификатор дедупликации, получает уникальный для процесса идентификатор, поэтому две одинаковые полезные нагрузки, отправленные намеренно, приходят обе. У политики нет идентификатора дедупликации: один постоянный идентификатор отбросил бы каждую отправку после первой.

Группу или идентификатор дедупликации на стандартной очереди крейт отвергает с ошибкой 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()
}

Вручную

src/main.rs

/// Ships one shipment.
struct Ship;

impl Handle<Shipment> for Ship {
    fn handle(
        &self,
        shipment: &Shipment,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let group = ctx
            .headers()
            .get_str(PARTITION_KEY_HEADER)
            .unwrap_or_default();
        tracing::info!(shipment.order, group, "shipment sent");
        ready(Ok(()))
    }
}

Доставка из очереди 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);
        },
    )
}

Вручную

src/main.rs

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(
                subscriber("orders", Dispatch)
                    .map_source(|_| SqsQueue::new("orders").create_if_missing())
                    .build(),
            )
            .out(Shipments, Publish::default())
            .build();
            // Four groups at once, each group in its own order.
            b.include(
                subscriber("shipments.fifo", Ship)
                    .map_source(|_| SqsQueue::new("shipments.fifo").create_if_missing())
                    .workers_by_key(nonzero!(4))
                    .build(),
            );
        },
    )
}

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

create_if_missing создаёт shipments.fifo как очередь FIFO с дедупликацией по содержимому. Очереди FIFO нужна DLQ типа FIFO; предел разбирает руководство Переобработка и DLQ. SQS ограничивает пропускную способность одной группы, поэтому сервис распределяет нагрузку по многим группам.

Тесты в файле примера проверяют группу, в которой уходит отправление, отправляют один заказ дважды и отправляют заказы двух клиентов одновременно. Обвязку разбирает руководство Тестирование сервиса на SQS.