Перейти к основному содержимому
Нормативная версия документации - английская

Эта страница переведена с английского языковой моделью. При любом расхождении верен английский оригинал.

Повторы и очереди dead-letter

Это руководство повторяет сообщение позже, ограничивает число попыток и переносит то, что не обрабатывается, в очередь dead-letter.

ZeroMQ доставляет сообщение не более одного раза и ничего не хранит. Сообщение, на котором обработчик не справился, пропадает, если сервис не опубликует его снова. Это делает фреймворк: точка монтирования объявляет предел попыток и адрес для копий, а каждая копия хранит число повторов в заголовке.

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

Не более одного раза​

ack и nack нечего сообщить сокету: в ZeroMQ нет подтверждения. Каждый из них отвечает AckError::Unsupported, и рантайм пишет в журнал предупреждение. Поэтому обычный retry без предела ничего не возвращает, и обработчик второй раз не вызывается. Повтор, который должен произойти, идёт через копию, как показано ниже.

Воркер заданий​

src/main.rs
use std::time::Duration;

use ruststream_zeromq::queue::prelude::*;
use serde::Deserialize;

/// How long a failed job waits before its next delivery.
const BACKOFF: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Deserialize)]
struct Job {
    id: u64,
}

#[subscriber("jobs")]
async fn work(job: &Job) -> HandlerOutcome {
    match run(job) {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(job.id, %err, "the job failed, retrying later");
            HandlerOutcome::retry_after(BACKOFF)
        }
    }
}

Задание, которое не удалось, просит новую попытку через 30 секунд с помощью retry_after. Рантайм выжидает задержку и публикует копию доставки: её полезную нагрузку и заголовки в том виде, в каком они пришли. Копия ждёт в памяти сервиса, поэтому процесс, который остановился до конца задержки, её теряет.

Воркер счетов​

src/main.rs
#[derive(Debug, PartialEq, Deserialize)]
struct Invoice {
    id: u64,
    amount_cents: u64,
}

#[subscriber("invoices")]
async fn send_invoice(invoice: &Invoice) -> HandlerOutcome {
    if invoice.amount_cents == 0 {
        tracing::warn!(invoice.id, "an invoice without an amount, parking it");
        return HandlerOutcome::retry();
    }
    tracing::info!(invoice.id, invoice.amount_cents, "invoice sent");
    HandlerOutcome::ack()
}

/// Keeps the invoices nobody could send, for a person to look at.
#[subscriber("invoices.dead")]
async fn park(invoice: &Invoice) -> HandlerOutcome {
    tracing::error!(invoice.id, "invoice parked");
    HandlerOutcome::ack()
}

Счёт без суммы отправить нельзя, и повторная отправка ничего не изменит. send_invoice возвращает retry, а точка монтирования превращает его в dead-letter. park читает очередь dead-letter и пишет каждое сообщение в журнал.

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

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    let dead = ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5561")).bindable();
    let parked = dead.bind(Publish);
    RustStream::new(AppInfo::new("billing", "0.1.0"))
        .with_broker_labeled(
            "jobs",
            ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")),
            |b| {
                // Five deliveries at most, the first one included; the copies return to this queue.
                b.include(work)
                    .max_attempts(nonzero!(5u32))
                    .out_retry(Publish);
            },
        )
        .with_broker_labeled(
            "invoices",
            ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5556")),
            |b| {
                // A failed invoice leaves for the second queue, under the name it is read by.
                b.include(send_invoice)
                    .dead_letter("invoices.dead")
                    .out_retry(parked);
            },
        )
        .with_broker_labeled("dead", dead, |b| {
            b.include(park);
        })
}

max_attempts считает все доставки одного задания, включая первую. В ZMTP нет счётчика доставок, поэтому фреймворк считает их в своём заголовке x-ruststream-retry-count, и предел соблюдается, куда бы ни уходили копии. Пятая доставка неудавшегося задания не повторяется. Без адреса dead-letter она отбрасывается с предупреждением, в котором названа подписка.

out_retry выбирает издателя, через которого уходят копии. Очередь jobs привязывает этот сервис, поэтому .out_retry(Publish) отправляет каждую копию обратно в неё, и копию берёт тот воркер заданий, который свободен.

Регистрация счетов объявляет dead_letter("invoices.dead") без предела, поэтому каждый неудавшийся счёт сразу уходит туда. Dead-letter - такая же публикация, как любая другая, а привязанная очередь принимает только имя своей подписки. Поэтому dead-letter уходят через parked, политику на брокере dead, и там их читает park. Копии для повтора регистрации уходят через того же издателя, поэтому регистрация выбирает между повторами и dead-letter. Регистрация, которой нужно и то и другое, держит повторы в собственной очереди и обходится без dead-letter.

Воркер, который подключается​

Воркер, который подключается к вентилятору, не может отправить копию обратно: вентилятор только отправляет. Его регистрация называет адрес для копий через .out_retry(policy).to("jobs") над очередью, которую воркер привязывает для себя, и монтирует на этой очереди тот же обработчик. Регистрацию без адреса старт отвергает, а цепочка Router без адреса не компилируется. Такую регистрацию показывает страница out_retry, а использует руководство Конвейеры на PUSH/PULL.

Проверка задержек​

Тесты запускают приложение во внутрипроцессном режиме. Тест заданий начинается с остановленными часами, поэтому tb.advance(BACKOFF) проматывает 30 секунд мгновенно. Он проверяет пять доставок и последнюю копию с числом повторов 4. Тест счетов проверяет, что счёт дошёл до park после одной доставки. Обвязку разбирает руководство Тестирование сервиса на ZeroMQ.