Чтение очередей

Это руководство читает очереди 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 запрашивает минуту. Без этого шага подписка один раз, при открытии, читает собственный тайм-аут очереди.

Обработчик может работать дольше тайм-аута. Пока доставка не завершена, крейт продлевает её видимость в середине каждого тайм-аута, поэтому сообщение медленного обработчика не достаётся другому потребителю. SQS скрывает сообщение не дольше двенадцати часов с момента получения. Если обработчик работает так долго, крейт продлевает видимость последний раз до конца этих двенадцати часов и пишет предупреждение в журнал, а затем сообщение получает другой потребитель.

Параллельные обработчики

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 создаёт очередь, которой нет. Сервис в промышленной эксплуатации обычно получает очереди от своей инфраструктуры и опускает этот шаг; тогда отсутствующая очередь останавливает его при старте.

Что содержит доставка

Каждый атрибут сообщения приходит в доставке заголовком, а каждый заголовок публикации уходит атрибутом. Два заголовка принадлежат самому крейту. RECEIVE_COUNT_HEADER хранит счётчик получений сообщения в очереди. PARTITION_KEY_HEADER хранит группу сообщений очереди FIFO. Тело SQS - это текст, поэтому двоичная полезная нагрузка передаётся в кодировке base64 с атрибутом-меткой и приходит уже декодированной. Base64 увеличивает такую нагрузку на треть, а сообщение больше предельного размера SQS отвергает.

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