Очереди и темы

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

AMQP 1.0 задаёт формат обмена, а смысл адреса оставляет брокеру. Поэтому подписка сама говорит, что ей нужно. У AmqpAddress есть конструктор для каждого намерения, и связь сообщает брокеру, какое из них выбрано. Всё, что утверждает эта страница, проверено на ActiveMQ Artemis, на котором работают тесты крейта.

Пример - сервис заказов. Он берёт заказы из очереди и объявляет о каждом принятом заказе в теме, на которую подписаны биллинг, журнал аудита и бухгалтерская книга.

Рабочая очередь

Макросы

src/main.rs

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

/// What every service that follows orders hears about a taken order.
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "order-events")]
struct OrderTaken {
    order: u64,
    amount_cents: u64,
}

#[subscriber(AmqpAddress::queue("orders"), reply)]
async fn take_order(order: &Order) -> OrderTaken {
    tracing::info!(order.id, "order taken");
    OrderTaken {
        order: order.id,
        amount_cents: order.amount_cents,
    }
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Order {
    id: u64,
    amount_cents: u64,
}

/// What every service that follows orders hears about a taken order.
#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct OrderTaken {
    order: u64,
    amount_cents: u64,
}

impl OutgoingDestination for OrderTaken {
    type Form = FixedName;

    const DESTINATION: &'static str = "order-events";
}

impl MessageHeaders for OrderTaken {
    type Contract = NoHeaders;
}

impl MessageInfo for OrderTaken {
    const NAME: &'static str = "OrderTaken";
}

struct TakeOrder;

impl Handle<Order, OrderTaken> for TakeOrder {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<OrderTaken, HandlerOutcome>> {
        tracing::info!(order.id, "order taken");
        ready(Ok(OrderTaken {
            order: order.id,
            amount_cents: order.amount_cents,
        }))
    }
}

AmqpAddress::queue запрашивает очередь: связь передаёт возможность терминуса queue. Artemis создаёт адрес anycast orders с одной очередью за ним. Потребители очереди конкурируют, и каждый заказ достаётся одному из них. Чтобы принимать заказы быстрее, запустите больше реплик сервиса: каждая реплика - ещё один потребитель orders.

take_order отвечает на каждый заказ сообщением OrderTaken. OrderTaken называет свой адрес через #[outgoing(name = "order-events")], поэтому аргумент reply макроса без имени отправляет его туда.

Рассылка

Макросы

src/main.rs

#[subscriber(AmqpAddress::topic("order-events"))]
async fn bill(event: &OrderTaken) -> HandlerOutcome {
    tracing::info!(event.order, event.amount_cents, "invoice issued");
    HandlerOutcome::ack()
}

#[subscriber(AmqpAddress::topic("order-events"))]
async fn audit(event: &OrderTaken) -> HandlerOutcome {
    tracing::info!(event.order, "order recorded");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

struct Bill;

impl Handle<OrderTaken> for Bill {
    fn handle(
        &self,
        event: &OrderTaken,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(event.order, event.amount_cents, "invoice issued");
        ready(Ok(()))
    }
}

struct Audit;

impl Handle<OrderTaken> for Audit {
    fn handle(
        &self,
        event: &OrderTaken,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(event.order, "order recorded");
        ready(Ok(()))
    }
}

AmqpAddress::topic запрашивает тему: связь передаёт возможность topic. Artemis делает order-events адресом multicast и даёт каждой подписке собственную очередь. Поэтому bill и audit получают каждый свою копию каждого OrderTaken.

Подписка на тему существует, пока существует её связь. То, что опубликовано в order-events, пока сервис остановлен, для него не сохраняется. Читатель, которому нельзя пропустить сообщение, читает очередь.

Рассылка, которая ждёт читателя

Макросы

src/main.rs

/// Reads the queue `ledger` that the operator created on the `order-events` address.
#[subscriber(AmqpAddress::raw("order-events::ledger"))]
async fn post_to_ledger(event: &OrderTaken) -> HandlerOutcome {
    tracing::info!(event.order, "order posted to the ledger");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Reads the queue `ledger` that the operator created on the `order-events` address.
struct PostToLedger;

impl Handle<OrderTaken> for PostToLedger {
    fn handle(
        &self,
        event: &OrderTaken,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(event.order, "order posted to the ledger");
        ready(Ok(()))
    }
}

Бухгалтерская книга должна увидеть каждый заказ, в том числе принятые, пока она была остановлена. Оператор создаёт долговечную очередь ledger на адресе multicast order-events:

docker compose exec artemis /opt/activemq-artemis/bin/artemis queue create \
  --user artemis --password artemis --silent \
  --name ledger --address order-events --auto-create-address \
  --multicast --durable --preserve-on-no-consumers

Теперь каждое сообщение, опубликованное в order-events, попадает и в ledger, независимо от того, работает ли post_to_ledger. order-events::ledger - полное имя очереди в Artemis: адрес, два двоеточия, очередь. AmqpAddress::raw отправляет адрес как написан, без возможности, поэтому связь присоединяется к этой очереди. Строковая форма #[subscriber("order-events::ledger")] делает то же самое.

Приложение

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            b.include(take_order).out_reply(Publish);
            b.include(bill);
            b.include(audit);
            b.include(post_to_ledger);
        },
    )
}

Вручную

src/main.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            b.include(
                subscriber(AmqpAddress::queue("orders"), TakeOrder)
                    .reply()
                    .build(),
            )
            .out_reply(Publish);
            b.include(subscriber(AmqpAddress::topic("order-events"), Bill).build());
            b.include(subscriber(AmqpAddress::topic("order-events"), Audit).build());
            let ledger = AmqpAddress::raw("order-events::ledger");
            b.include(subscriber(ledger, PostToLedger).build());
        },
    )
}

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

.out_reply(Publish) называет издателя ответов - AmqpPublish под именем из прелюдии. При первой публикации издатель присоединяет связь отправителя к адресу ответа и дальше держит её. Если брокер отказывается принять сообщение, публикация возвращает ошибку AmqpError::PublishNotAccepted.

Заголовки и другие клиенты

Общеизвестные заголовки передаются в секции AMQP properties: content-type, correlation-id, reply-to, message-id. Все остальные заголовки передаются в application-properties. Сообщение этого крейта - обычное сообщение AMQP, поэтому клиент на другом языке читает его без RustStream.

Ключ партиционирования передаётся заголовком PARTITION_KEY_HEADER и становится полем group-id сообщения. Artemis использует group-id для группировки сообщений: все сообщения одной группы достаются одному потребителю очереди. Поэтому заказы с ключом по клиенту сохраняют свой порядок на нескольких репликах.

Адреса в других брокерах

Другие брокеры называют свои узлы по-своему, и raw передаёт их запись как есть. Её описывает собственная документация брокера: RabbitMQ 4.x, Azure Service Bus, ActiveMQ Classic, на котором работает Amazon MQ.

Тестирование маршрутизации

Тесты в файле примера запускают приложение внутри процесса. Один заказ публикует один OrderTaken, обработчик очереди orders вызывается один раз, а order-events - дважды, по разу на каждую подписку на тему. Брокер внутри процесса доставляет адрес raw точно в том виде, в каком он написан, поэтому его тест публикует прямо в order-events::ledger. Руководство Тестирование сервиса на AMQP разбирает обвязку.