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

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

Обменники и маршрутизация

Это руководство направляет сообщения через обменники RabbitMQ: по ключу маршрутизации, во все привязанные очереди и по заголовкам.

Издатель никогда не пишет в очередь. Он публикует сообщение в обменник с ключом маршрутизации, а обменник копирует сообщение в каждую очередь, привязка которой подходит. В RustStream привязки очереди задаются в её дескрипторе, а обменник, в который уходит сообщение, - в политике публикации. Ключ маршрутизации - это имя сообщения: #[outgoing(name = "..")] его типа или имя, которое задаёт публикация.

Пример - магазин, который сообщает о заказах через обменник типа topic, об изменениях цен - через обменник типа fanout, а об отгрузках - через обменник типа headers. Брокер объявляет обменники, очереди и привязки, поэтому пример работает на пустом RabbitMQ (Топология).

Обменник по умолчанию​

Все сервисы до сих пор публиковали в обменник по умолчанию. Это обменник типа direct, к которому каждая очередь привязана под собственным именем. Поэтому сообщение с ключом маршрутизации orders попадает в очередь orders, и сервис читает его через #[subscriber("orders")] без всякой привязки. Привязку к обменнику по умолчанию отвергают и сервер, и RabbitMQ внутри процесса.

Собственный обменник типа direct маршрутизирует так же: привязка называет один ключ маршрутизации, и сообщение попадает в очереди, привязанные ровно под его ключом.

Обменники типа topic​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct Order {
    id: u64,
    region: String,
}

/// An order event; its routing key is `order.placed.<region>`.
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct OrderPlaced {
    id: u64,
    region: String,
}

/// Takes an order from the `orders` queue and announces it on the `events` exchange.
#[subscriber("orders")]
async fn place_order(order: &Order, Out(events): Out<impl Publisher>) -> HandlerOutcome {
    let placed = OrderPlaced {
        id: order.id,
        region: order.region.clone(),
    };
    let key = format!("order.placed.{}", order.region);
    if events.message(&placed).to(&key).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

/// Every order event of every region.
#[subscriber(RabbitQueue::new("notifications.orders")
    .bind(RabbitExchange::topic("events"), "order.placed.*"))]
async fn notify_customer(event: &OrderPlaced, Ctx(key): Ctx<RoutingKey>) -> HandlerOutcome {
    tracing::info!(event.id, routing_key = %key, "customer notified");
    HandlerOutcome::ack()
}

/// The events of the EU region only.
#[subscriber(RabbitQueue::new("warehouse.eu")
    .bind(RabbitExchange::topic("events"), "*.*.eu"))]
async fn reserve_stock(event: &OrderPlaced) -> HandlerOutcome {
    tracing::info!(event.id, "stock reserved in the EU warehouse");
    HandlerOutcome::ack()
}

/// Every event on the exchange, whatever its key.
#[subscriber(RabbitQueue::new("audit").bind(RabbitExchange::topic("events"), "#"))]
async fn audit(event: &OrderPlaced, Ctx(key): Ctx<RoutingKey>) -> HandlerOutcome {
    tracing::info!(event.id, routing_key = %key, "event audited");
    HandlerOutcome::ack()
}

place_order публикует каждый заказ как OrderPlaced с ключом маршрутизации order.placed.<region>. Тип не объявляет имени, поэтому имя задаёт публикация через .to(&key). Обменник типа topic сравнивает ключ с шаблоном каждой привязки по словам, разделённым точками. * заменяет ровно одно слово, а # - любое число слов, в том числе ни одного.

Поэтому order.placed.* принимает заказы всех регионов, *.*.eu - только заказы из ЕС, а # - все сообщения обменника. Заказ из США попадает в уведомления и в аудит, а европейский склад его не видит. Обработчик читает ключ, с которым пришло сообщение, через Ctx<RoutingKey>.

Обменники типа fanout​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct PriceChange {
    sku: String,
    price_cents: u64,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "price.changed")]
struct PriceChanged {
    sku: String,
    price_cents: u64,
}

/// Applies a price change from the catalog and announces it to every reader of `prices`.
#[subscriber("catalog.prices", reply)]
async fn change_price(change: &PriceChange) -> PriceChanged {
    tracing::info!(change.sku, change.price_cents, "price changed");
    PriceChanged {
        sku: change.sku.clone(),
        price_cents: change.price_cents,
    }
}

#[subscriber(RabbitQueue::new("storefront.prices").bind(RabbitExchange::fanout("prices"), ""))]
async fn refresh_storefront(change: &PriceChanged) -> HandlerOutcome {
    tracing::info!(change.sku, "storefront price refreshed");
    HandlerOutcome::ack()
}

#[subscriber(RabbitQueue::new("search.prices").bind(RabbitExchange::fanout("prices"), ""))]
async fn reindex(change: &PriceChanged) -> HandlerOutcome {
    tracing::info!(change.sku, "search index updated");
    HandlerOutcome::ack()
}

Обменник типа fanout копирует каждое сообщение во все привязанные к нему очереди и не смотрит на ключ маршрутизации. change_price отвечает сообщением PriceChanged, и витрина и поисковый индекс получают каждый свою копию. Привязки называют пустой ключ, потому что обменник ключ не читает. Новый читатель изменений цен - это ещё одна очередь, привязанная к prices; издатель при этом не меняется.

Обменники типа headers​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
struct PaidOrder {
    id: u64,
    region: String,
    tier: String,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "shipment")]
struct Shipment {
    order_id: u64,
    region: String,
    tier: String,
}

/// Ships a paid order, with its region and tier as headers, which the exchange routes on.
#[subscriber("orders.paid")]
async fn ship_order(order: &PaidOrder, Out(shipments): Out<impl Publisher>) -> HandlerOutcome {
    let shipment = Shipment {
        order_id: order.id,
        region: order.region.clone(),
        tier: order.tier.clone(),
    };
    let mut attributes = HeaderMap::new();
    attributes.insert("region", order.region.clone());
    attributes.insert("tier", order.tier.clone());
    let sent = shipments
        .message(&shipment)
        .with_headers(attributes)
        .publish()
        .await;
    if sent.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

/// The binding arguments: how to match, then the headers matched on.
fn gold_in_eu(x_match: &str) -> FieldTable {
    let mut arguments = FieldTable::default();
    arguments.insert("x-match".into(), AMQPValue::LongString(x_match.into()));
    arguments.insert("region".into(), AMQPValue::LongString("eu".into()));
    arguments.insert("tier".into(), AMQPValue::LongString("gold".into()));
    arguments
}

/// Gold shipments in the EU: both headers have to match.
#[subscriber(RabbitQueue::new("couriers.express")
    .bind_with(RabbitExchange::headers("shipments"), "", gold_in_eu("all")))]
async fn book_express(shipment: &Shipment) -> HandlerOutcome {
    tracing::info!(shipment.order_id, "express courier booked");
    HandlerOutcome::ack()
}

/// Shipments that are gold or in the EU: one header is enough.
#[subscriber(RabbitQueue::new("insurance")
    .bind_with(RabbitExchange::headers("shipments"), "", gold_in_eu("any")))]
async fn insure(shipment: &Shipment) -> HandlerOutcome {
    tracing::info!(shipment.order_id, "shipment insured");
    HandlerOutcome::ack()
}

Обменник типа headers маршрутизирует по заголовкам сообщения, а не по ключу. bind_with задаёт аргументы привязки: заголовки для сравнения и x-match, который говорит, сколько из них должно совпасть. При all сообщению нужны все заголовки, при any достаточно одного. ship_order ставит каждой отгрузке заголовки region и tier через with_headers. Золотая отгрузка в ЕС попадает в обе очереди, серебряная в ЕС - только в страхование, а серебряная в США - никуда.

Простой bind нужен для обменников, которые маршрутизируют по ключу. На обменнике типа headers он создаёт привязку без заголовков для сравнения, а условие all по пустому набору заголовков истинно для любого сообщения. FieldTable и AMQPValue взяты из lapin и реэкспортируются в прелюдии.

Публикация в обменник​

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    let broker = LapinBroker::new("amqp://localhost:5672").declare_topology(true);
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(broker, |b| {
        b.include(place_order)
            .out(DefaultSlot, Publish::default().exchange("events"))
            .build();
        b.include(notify_customer);
        b.include(reserve_stock);
        b.include(audit);

        b.include(change_price)
            .out_reply(Publish::default().exchange("prices"));
        b.include(refresh_storefront);
        b.include(reindex);

        b.include(ship_order)
            .out(DefaultSlot, Publish::default().exchange("shipments"))
            .build();
        b.include(book_express);
        b.include(insure);
    })
}

Обменник называется один раз, в политике в точке монтирования: Publish::default().exchange("events"). Тело обработчика называет только сообщение и его ключ, поэтому тот же обработчик публикует в другой обменник, если так указано в точке монтирования. Для ответа та же политика указывается в .out_reply(..).

RabbitMQ отвергает публикацию в несуществующий обменник и закрывает канал, через который она шла. Здесь подписки объявляют events, prices и shipments при запуске, до того как какой-либо обработчик начнёт публиковать. Если сервис публикует в обменник, из которого не читает, объявить этот обменник должны эксплуатация или другой сервис.

Каждая привязка подписки появляется и в документе AsyncAPI сервиса, в расширении x-ruststream-amqp её канала.

Согласованное хеширование​

Фича plugin-consistent-hash добавляет RabbitExchange::consistent_hash - тип обменника из плагина согласованного хеширования RabbitMQ. Он распределяет один поток сообщений по нескольким очередям по хешу ключа маршрутизации, а каждая очередь привязывается со своим весом в качестве ключа маршрутизации. Сообщения с одинаковым ключом всегда попадают в одну и ту же очередь. Плагин должен быть включён на сервере, и маршрутизация выполняется только там, поэтому его тесты работают на настоящем RabbitMQ.

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

Тесты в файле примера публикуют заказ, изменение цены и три оплаченных заказа и проверяют, в какие очереди попало каждое сообщение. RabbitMQ внутри процесса маршрутизирует через обменники по умолчанию, direct, topic, fanout и headers по тем же привязкам, которые объявляют дескрипторы. Обвязку разбирает руководство Тестирование сервиса на RabbitMQ.