Пакетная обработка

Это руководство передаёт обработчику много сообщений сразу, для работы, которая дешевле на пакет, чем на сообщение, например одной записи в хранилище данных.

Pub/Sub доставляет сообщения по одному и завершает каждое отдельно. Сервис сам собирает доставки одной подписки в пакет, вызывает обработчик один раз на пакет, а затем завершает каждое его сообщение.

Пример загружает клики по страницам в аналитическое хранилище.

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

Макросы

src/main.rs

/// Loads a batch of clicks into the warehouse in one write. A click from a user the warehouse
/// does not know yet comes back on its own; a click without a page is dropped.
#[subscriber(
    GooglePubSub::new("clicks-loader")
        .create_with_topic("clicks")
        .batch_wait(Duration::from_millis(500))
)]
async fn load(clicks: &[Click]) -> Vec<HandlerOutcome> {
    tracing::info!(count = clicks.len(), "clicks loaded");
    clicks.iter().map(settle).collect()
}

fn settle(click: &Click) -> HandlerOutcome {
    if click.page.is_empty() {
        HandlerOutcome::drop()
    } else if click.user == UNKNOWN_USER {
        HandlerOutcome::retry()
    } else {
        HandlerOutcome::ack()
    }
}

Вручную

src/main.rs

/// Loads a batch of clicks into the warehouse in one write. A click from a user the warehouse
/// does not know yet comes back on its own; a click without a page is dropped.
struct Load;

impl Handle<[Click]> for Load {
    fn handle(
        &self,
        clicks: &[Click],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        tracing::info!(count = clicks.len(), "clicks loaded");
        // `Err` carries one outcome per click, in the order of the batch.
        ready(Err(clicks.iter().map(settle).collect()))
    }
}

fn settle(click: &Click) -> HandlerOutcome {
    if click.page.is_empty() {
        HandlerOutcome::drop()
    } else if click.user == UNKNOWN_USER {
        HandlerOutcome::retry()
    } else {
        HandlerOutcome::ack()
    }
}

Параметр-срез &[Click] делает load пакетным обработчиком. batch_wait на дескрипторе - сколько неполный пакет ждёт новых сообщений, здесь полсекунды. Пакет передаётся, как только заполнится, или по истечении ожидания с тем, что в нём есть.

load возвращает по исходу на каждый клик в порядке пакета. Клик пользователя, которого хранилище ещё не знает, переобрабатывается: Pub/Sub доставляет этот клик снова, и он приходит отдельно или в одном из следующих пакетов. Клик без страницы отбрасывается, а остальные подтверждаются. Один HandlerOutcome завершает весь пакет одинаково. В ручной форме Ok(()) подтверждает пакет, а Err содержит исходы для каждого клика.

Точка монтирования

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("clicks", "0.1.0")).with_broker(
        PubSubBroker::new("my-project").emulator("localhost:8085"),
        |b| {
            // At most a hundred clicks per call; a click refused five times goes to `clicks-dead`.
            b.include(load.batch(nonzero!(100)))
                .max_attempts(nonzero!(5u32))
                .dead_letter("clicks-dead");
        },
    )
}

Вручную

src/main.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("clicks", "0.1.0")).with_broker(
        PubSubBroker::new("my-project").emulator("localhost:8085"),
        |b| {
            // At most a hundred clicks per call; a click refused five times goes to `clicks-dead`.
            b.include(
                subscriber(
                    GooglePubSub::new("clicks-loader")
                        .create_with_topic("clicks")
                        .batch_wait(Duration::from_millis(500)),
                    Load,
                )
                .batch(nonzero!(100))
                .build(),
            )
            .max_attempts(nonzero!(5u32))
            .dead_letter("clicks-dead");
        },
    )
}

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

batch(nonzero!(100)) задаёт размер пакета, и без него пакетный обработчик не компилируется. Каждое сообщение пакета остаётся в аренде, пока пакет не завершён, поэтому размер пакета не превышает max_outstanding подписки, по умолчанию 1000. Объявление dead-letter считает каждый клик отдельно: после пяти доставок клик уходит в clicks-dead, в скольких бы пакетах он ни побывал. Тему dead-letter описывает руководство Подтверждение и переобработка.

Тестирование

Тесты публикуют клики вместе через tokio::join!, чтобы они попали в один пакет, и читают пакеты через assert_batch_sizes и batches. Переобработанный клик виден как пакет из одного сообщения: сначала пакет из двух, затем клик незнакомого пользователя отдельно ещё четыре раза, после чего он оказывается в clicks-dead. Работу с обвязкой описывает руководство Тестирование сервиса на Pub/Sub.