Топология

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

Дескриптор очереди описывает то, что сервис ожидает найти на сервере. Ничего не создаётся, пока брокер не включит объявление через declare_topology(true). Пример - платёжный сервис с реплицируемой очередью платежей, очередью для неудавшихся платежей и собственной очередью курсов валют у каждого экземпляра.

Классические и кворумные очереди

Макросы

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Payment {
    id: u64,
    card: String,
    amount_cents: u64,
}

/// Payments live in a replicated quorum queue; what it rejects goes to `payments.dead`.
#[subscriber(RabbitQuorumQueue::new("payments").dead_letter_exchange("payments.dead"))]
async fn charge(payment: &Payment) -> HandlerOutcome {
    if payment.card.is_empty() {
        tracing::warn!(payment.id, "payment without a card rejected");
        return HandlerOutcome::drop();
    }
    tracing::info!(payment.id, payment.amount_cents, "payment charged");
    HandlerOutcome::ack()
}

/// Every payment the `payments` queue dead-letters.
#[subscriber(RabbitQueue::new("payments.failed")
    .bind(RabbitExchange::fanout("payments.dead"), ""))]
async fn record_failure(payment: &Payment) -> HandlerOutcome {
    tracing::error!(payment.id, "payment failed");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

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

struct Charge;

impl Handle<Payment> for Charge {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if payment.card.is_empty() {
            tracing::warn!(payment.id, "payment without a card rejected");
            return ready(Err(HandlerOutcome::drop()));
        }
        tracing::info!(payment.id, payment.amount_cents, "payment charged");
        ready(Ok(()))
    }
}

struct RecordFailure;

impl Handle<Payment> for RecordFailure {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::error!(payment.id, "payment failed");
        ready(Ok(()))
    }
}

/// Payments live in a replicated quorum queue; what it rejects goes to `payments.dead`.
fn payments_queue() -> RabbitQuorumQueue {
    RabbitQuorumQueue::new("payments").dead_letter_exchange("payments.dead")
}

/// Every payment the `payments` queue dead-letters.
fn failed_queue() -> RabbitQueue {
    RabbitQueue::new("payments.failed").bind(RabbitExchange::fanout("payments.dead"), "")
}

В RabbitMQ два типа очередей, и тип задаётся при создании очереди. Классическая очередь хранится на одном узле. Кворумная очередь реплицируется по кластеру алгоритмом Raft и переживает потерю узла. У каждого типа свой дескриптор: RabbitQueue и RabbitQuorumQueue. Берите кворумную очередь для данных, которые нельзя потерять, например платежей, а классическую - для остального. Руководство RabbitMQ по кворумным очередям сравнивает оба типа.

Кворумная очередь всегда сохраняется на диск, доступна всем соединениям и не удаляется автоматически. Поэтому у RabbitQuorumQueue нет durable, exclusive и auto_delete, и кворумная очередь, противоречащая своему типу, не компилируется. Кроме того, кворумная очередь считает доставки каждого сообщения, на чём строится руководство по переобработке.

dead_letter_exchange задаёт очереди payments маршрут для отклонённых сообщений. Платёж, который charge отбрасывает, уходит в обменник недоставленных сообщений payments.dead, а оттуда в привязанную к нему очередь payments.failed. dead_letter_routing_key заменяет ключ маршрутизации сообщения на этом маршруте, если обменник маршрутизирует по ключу.

Очередь у каждого экземпляра

Макросы

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct FxRate {
    currency: String,
    micros_per_euro: u64,
}

/// The name of this instance's own queue.
fn rates_queue_name() -> String {
    format!("payments.rates.{}", process::id())
}

/// A queue of this instance's own: it lives as long as the connection, and a rate older than a
/// minute expires unread.
#[subscriber(RabbitQueue::new(rates_queue_name())
    .durable(false)
    .exclusive(true)
    .auto_delete(true)
    .argument("x-message-ttl", AMQPValue::LongLongInt(60_000))
    .bind(RabbitExchange::fanout("fx-rates"), ""))]
async fn update_rate(rate: &FxRate) -> HandlerOutcome {
    tracing::info!(rate.currency, rate.micros_per_euro, "exchange rate updated");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct FxRate {
    currency: String,
    micros_per_euro: u64,
}

/// The name of this instance's own queue.
fn rates_queue_name() -> String {
    format!("payments.rates.{}", process::id())
}

/// A queue of this instance's own: it lives as long as the connection, and a rate older than a
/// minute expires unread.
fn rates_queue() -> RabbitQueue {
    RabbitQueue::new(rates_queue_name())
        .durable(false)
        .exclusive(true)
        .auto_delete(true)
        .argument("x-message-ttl", AMQPValue::LongLongInt(60_000))
        .bind(RabbitExchange::fanout("fx-rates"), "")
}

struct UpdateRate;

impl Handle<FxRate> for UpdateRate {
    fn handle(
        &self,
        rate: &FxRate,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(rate.currency, rate.micros_per_euro, "exchange rate updated");
        ready(Ok(()))
    }
}

Каждому экземпляру нужны все курсы валют, поэтому каждый экземпляр читает собственную очередь, привязанную к обменнику fx-rates типа fanout. Имя очереди содержит идентификатор процесса, поэтому два экземпляра никогда не делят её. exclusive(true) разрешает пользоваться очередью только этому соединению, и RabbitMQ удаляет её, когда соединение закрывается. auto_delete(true) удаляет её, когда уходит последний потребитель. durable(false) не сохраняет её на диск: после перезапуска курсы ничего не стоят. RabbitMQ 4 принимает неустойчивую очередь, только если она эксклюзивная, поэтому durable(false) идёт вместе с exclusive(true).

argument передаёт аргумент объявления, для которого у дескриптора нет метода, здесь x-message-ttl: курс, который никто не прочитал за минуту, устаревает. Так же задаются x-max-length, x-max-priority, x-overflow и x-single-active-consumer. Значение имеет тип AMQPValue, а arguments принимает целую таблицу.

Объявляет сервис

Макросы

src/main.rs

fn service(broker: LapinBroker) -> RustStream {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        b.include(charge);
        b.include(record_failure);
        b.include(update_rate);
    })
}

#[ruststream::app]
fn app() -> impl App {
    // Development declares the topology; production runs with `DECLARE_TOPOLOGY=false` against
    // the queues operations created.
    let declare = env::var("DECLARE_TOPOLOGY").map_or(true, |value| value != "false");
    service(LapinBroker::new("amqp://localhost:5672").declare_topology(declare))
}

Вручную

src/main.rs

fn service(broker: LapinBroker) -> RustStream {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber(payments_queue(), Charge).build());
        b.include(subscriber(failed_queue(), RecordFailure).build());
        b.include(subscriber(rates_queue(), UpdateRate).build());
    })
}

fn app() -> impl App {
    // Development declares the topology; production runs with `DECLARE_TOPOLOGY=false` against
    // the queues operations created.
    let declare = env::var("DECLARE_TOPOLOGY").map_or(true, |value| value != "false");
    service(LapinBroker::new("amqp://localhost:5672").declare_topology(declare))
}

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

При declare_topology(true) каждая подписка при запуске объявляет то, что описывает её дескриптор. Сначала она объявляет обменники, которые называют её привязки, затем очередь с её аргументами, затем привязки. Обменник по умолчанию и встроенные обменники amq.* есть на любом сервере, и их никто не объявляет. Повторное объявление безвредно, пока оно совпадает с тем, что есть на сервере, поэтому все экземпляры сервиса объявляют одни и те же очереди.

Объявление, которое расходится с существующей очередью, возвращает ошибку. На повторное объявление очереди с другой устойчивостью или другими аргументами RabbitMQ отвечает PRECONDITION_FAILED, и сервис не запускается: AmqpError::Declare называет очередь. Тип и аргументы очереди меняются только удалением и повторным объявлением, при котором её содержимое теряется, поэтому их меняют миграцией, а не развёртыванием. Какие свойства должны совпадать, RabbitMQ объясняет в разделе property equivalence.

Объявляет эксплуатация

DECLARE_TOPOLOGY=false запускает тот же сервис с выключенным объявлением. Тогда дескрипторы ничего не объявляют: они описывают очереди, аргументы и привязки, которые ожидает сервис, а создаёт их эксплуатация. Если очереди нет, подписка при запуске возвращает ошибку AmqpError::Subscribe, поэтому отсутствующая очередь останавливает развёртывание, а не теряет сообщения.

Лимиты и TTL эксплуатация часто задаёт политикой RabbitMQ, а не аргументами объявления. Политика меняет очередь без повторного объявления, а аргумент, заданный политикой, никогда не конфликтует с объявлением сервиса. Тогда дескриптор называет только то, на что опирается сам сервис.

От выключенного объявления зависит и настройка переобработки на кворумной очереди; это описывает руководство по переобработке.

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

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