Очереди задач

Это руководство делит одну очередь RabbitMQ между воркерами сервиса и сохраняет порядок там, где он важен.

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

Конкурирующие потребители

Каждый экземпляр сервиса подписывается на одну и ту же очередь. RabbitMQ отдаёт каждое сообщение одному из них по очереди и никогда двум сразу. Ещё один запущенный экземпляр - это ещё один потребитель, и очередь разбирается быстрее. Сообщение, которое экземпляр взял и не подтвердил, возвращается в очередь, когда закрывается его соединение, и его берёт другой экземпляр.

Для этого сервису ничего не нужно: имя очереди в #[subscriber] одинаково во всех экземплярах.

Предвыборка

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672")
        .prefetch(nonzero!(32))
        .declare_topology(true);
    RustStream::new(AppInfo::new("invoices", "0.1.0")).with_broker(broker, |b| {
        b.include(render_invoice);
        b.include(archive.batch(nonzero!(64)));
        b.include(post_to_ledger);
    })
}

Вручную

src/main.rs

fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672")
        .prefetch(nonzero!(32))
        .declare_topology(true);
    RustStream::new(AppInfo::new("invoices", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber(RabbitQueue::new("invoices"), RenderInvoice)
                .workers_by_key(nonzero!(8usize))
                .build(),
        );
        b.include(
            subscriber(
                RabbitQueue::new("invoices.archive")
                    .prefetch(nonzero!(64))
                    .batch_wait(Duration::from_millis(200)),
                Archive,
            )
            .batch(nonzero!(64usize))
            .build(),
        );
        b.include(
            subscriber(
                RabbitQueue::new("invoices.ledger")
                    .argument("x-single-active-consumer", AMQPValue::Boolean(true)),
                PostToLedger,
            )
            .build(),
        );
    })
}

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

prefetch ограничивает число неподтверждённых доставок, которые держит подписка, здесь 32. Пока у потребителя столько неподтверждённых сообщений, RabbitMQ приостанавливает доставку ему. Поэтому медленный экземпляр перестаёт брать работу, и следующее сообщение уходит экземпляру, у которого есть место. Без предвыборки RabbitMQ отдаёт очередь потребителям так быстро, как позволяет сеть, и у медленного экземпляра копятся сообщения, которые могли бы взять другие. Число имеет тип NonZeroU16: отсутствие ограничения выражается тем, что значение не задано. Значение брокера действует для всех подписок, а дескриптор очереди переопределяет его для своей очереди.

Окно предвыборки - свойство сервера, поэтому тест видит его только на работающем RabbitMQ.

Партиции воркеров и порядок по ключу

Макросы

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Invoice {
    id: u64,
    customer: String,
    total_cents: u64,
}

/// Renders one invoice; the invoices of one customer arrive in order on one lane.
#[subscriber(RabbitQueue::new("invoices"), workers(8, by_key))]
async fn render_invoice(invoice: &Invoice, Ctx(redelivered): Ctx<Redelivered>) -> HandlerOutcome {
    tracing::info!(
        invoice.id,
        invoice.customer,
        redelivered,
        "invoice rendered"
    );
    HandlerOutcome::ack()
}

Вручную

src/main.rs

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

/// Renders one invoice; the invoices of one customer arrive in order on one lane.
struct RenderInvoice;

impl Handle<Invoice, (), (), AmqpContext> for RenderInvoice {
    fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        ctx: &mut Context<'_, AmqpContext>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        let redelivered = ctx.context(Redelivered);
        tracing::info!(
            invoice.id,
            invoice.customer,
            redelivered,
            "invoice rendered"
        );
        ready(Ok(()))
    }
}

workers(8, by_key) обрабатывает до восьми счетов одного экземпляра одновременно, в восьми партициях воркеров. Биллинговый сервис, который публикует счёт, записывает клиента в заголовок amqp-partition-key (PARTITION_KEY_HEADER). Счета с одинаковым ключом попадают в одну партицию воркеров и обрабатываются друг за другом в том порядке, в каком их получил экземпляр. Счета разных клиентов обрабатываются параллельно. Счёт без этого заголовка уходит в следующую по очереди партицию воркеров. RabbitMQ заголовок не читает: партиции воркеров - соглашение самого сервиса.

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

Пакеты

Макросы

src/main.rs

/// Writes rendered invoices to the archive in batches of up to 64.
#[subscriber(RabbitQueue::new("invoices.archive")
    .prefetch(nonzero!(64))
    .batch_wait(Duration::from_millis(200)))]
async fn archive(invoices: &[Invoice]) -> HandlerOutcome {
    tracing::info!(count = invoices.len(), "invoices archived");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Writes rendered invoices to the archive in batches of up to 64.
struct Archive;

impl Handle<[Invoice]> for Archive {
    fn handle(
        &self,
        invoices: &[Invoice],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        tracing::info!(count = invoices.len(), "invoices archived");
        ready(Ok(()))
    }
}

Параметр-срез запрашивает пакет, а точка монтирования задаёт его размер через .batch(nonzero!(64)). AMQP доставляет по одному сообщению, поэтому пакет собирается в сервисе. Пакет закрывается, когда в нём 64 счёта или когда с первого счёта прошло время batch_wait; по умолчанию это 50 мс. В пакет попадает только то, что RabbitMQ уже отправил, поэтому очередь архива поднимает собственную предвыборку до 64. С 32 от брокера ни один пакет не дошёл бы до 64. Исход обработчика подтверждает каждый счёт пакета.

Один потребитель за раз

Макросы

src/main.rs

/// Posts invoices to the ledger, which needs them in queue order across every instance.
#[subscriber(RabbitQueue::new("invoices.ledger")
    .argument("x-single-active-consumer", AMQPValue::Boolean(true)))]
async fn post_to_ledger(invoice: &Invoice) -> HandlerOutcome {
    tracing::info!(invoice.id, invoice.total_cents, "invoice posted");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Posts invoices to the ledger, which needs them in queue order across every instance.
struct PostToLedger;

impl Handle<Invoice> for PostToLedger {
    fn handle(
        &self,
        invoice: &Invoice,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(invoice.id, invoice.total_cents, "invoice posted");
        ready(Ok(()))
    }
}

Бухгалтерской книге нужны все счета в порядке очереди, по всем экземплярам. Это даёт единственный активный потребитель RabbitMQ: очередь доставляет сообщения одному потребителю, а когда он уходит, его место занимает другой. У дескриптора нет для этого метода, поэтому argument передаёт x-single-active-consumer в объявление очереди как есть. Остальные экземпляры остаются подписанными и простаивают, пока не займут его место.

Аргумент - часть объявления очереди. При declare_topology(true) сервис объявляет очередь с ним; очередь, которую создаёт эксплуатация, получает его из её объявления (Топология).

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

Тесты в файле примера публикуют счета с ключом через with_headers и проверяют, что счета одного клиента приходят в обработчик по порядку. RabbitMQ внутри процесса, как и сервер, делит очередь между конкурирующими потребителями и собирает пакеты. Окно предвыборки и единственный активный потребитель - свойства сервера, поэтому их проверяет тест на сервере. Обвязку разбирает руководство Тестирование сервиса на RabbitMQ.