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

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

Рассылка через Pub/Sub

Это руководство рассылает сообщения всем работающим экземплярам сервиса через Redis Pub/Sub.

Пример - витрина интернет-магазина. Каждый экземпляр обновляет кеш цен, когда меняется цена, следит за событиями заказов через шаблон и объявляет об отправленных заказах.

Сообщение получает каждый подписчик​

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

#[subscriber(RedisPubSub::new("prices.changed").buffer(nonzero!(1_000)))]
async fn refresh_price(change: &PriceChanged) -> HandlerOutcome {
    tracing::info!(change.sku, change.price_cents, "price cache refreshed");
    HandlerOutcome::ack()
}

#[subscriber(RedisPubSub::new("prices.changed"))]
async fn reindex_price(change: &PriceChanged) -> HandlerOutcome {
    tracing::info!(change.sku, "search index updated");
    HandlerOutcome::ack()
}

RedisPubSub подписывается на один канал. Сообщение, опубликованное в канал, получает каждая подписка, подключённая в этот момент: его получают и refresh_price, и reindex_price в каждом работающем экземпляре. Сообщения не делятся между подписками и не подтверждаются. Исход, который возвращает обработчик, ничего не завершает.

Redis не хранит копию. Сообщение, опубликованное, пока ни одна подписка не подключена, пропадает, как и сообщение для экземпляра, который в это время перезапускается. Подписка открывается только после того, как Redis направил ей её канал, поэтому сообщение, опубликованное после запуска сервиса, не теряется.

Медленный подписчик​

Redis отправляет каждое сообщение сразу после публикации, готов обработчик или нет. Подписка держит то, что обработчик ещё не забрал, в своём буфере: 32 сообщения, если buffer не задаёт другой размер. Когда буфер заполнен, каждое новое сообщение вытесняет самое старое, а подписка пишет в журнал предупреждение с каналом и числом потерянных сообщений. refresh_price задаёт буфер на тысячу сообщений, поэтому всплеск изменений цен ждёт обработчика, а не обрезается. Буфер выделяется при открытии подписки, около 600 байт на сообщение, сколько бы ни весила полезная нагрузка.

Шаблоны​

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
struct OrderEvent {
    order_id: u64,
}

#[subscriber(RedisPubSubPattern::new("orders.events.*"))]
async fn track_order(event: &OrderEvent, Ctx(channel): Ctx<keys::Channel>) -> HandlerOutcome {
    tracing::info!(event.order_id, channel, "order event");
    HandlerOutcome::ack()
}

RedisPubSubPattern подписывается на маску командой PSUBSCRIBE, здесь на все каналы с префиксом orders.events.. Обработчик узнаёт канал, по которому пришло сообщение, из контекста Pub/Sub через Ctx<keys::Channel>. keys::FromPattern сообщает, пришло ли сообщение через шаблон.

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

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "shipments")]
struct Shipment {
    order_id: u64,
}

#[subscriber(
    RedisStream::new("shipments").group("notifier"),
    reply("orders.events.shipped")
)]
async fn announce_shipment(shipment: &Shipment) -> OrderEvent {
    OrderEvent {
        order_id: shipment.order_id,
    }
}

announce_shipment читает отправки из потока и отвечает сообщением OrderEvent в канал orders.events.shipped.

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("storefront", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(refresh_price);
            b.include(reindex_price);
            // A glob is no channel a publish can reach, so the registration names one for copies.
            b.include(track_order)
                .out_retry(pubsub::Publish::default())
                .to("orders.events.retry");
            b.include(announce_shipment)
                .out_reply(pubsub::Publish::default());
        },
    )
}

Ответ уходит через pubsub::Publish, который отправляет PUBLISH. Имя, которое не читает ни одна подписка этого сервиса, издатель брокера по умолчанию считает ключом потока и отправляет в него XADD. Подписка по шаблону не читает orders.events.shipped по имени, поэтому регистрация указывает политику Pub/Sub. Подписка по шаблону в том же сервисе получает это объявление.

Шаблон не называет канала, в который можно опубликовать копию. Поэтому регистрация track_order задаёт канал через out_retry(..).to("orders.events.retry"), и без этого сервис не запускается. Этот канал подходит под шаблон, поэтому переобработанное событие возвращается к обработчику в каждом работающем экземпляре.

Шардированные каналы в кластере​

В кластере классический PUBLISH уходит на каждый узел. PubSubMode::Sharded вместо него использует SSUBSCRIBE и SPUBLISH: канал находится на узле, которому принадлежит его хеш-слот, поэтому трафик распределяется по кластеру. Этому режиму нужен Redis 7 или новее, и шаблонов в нём нет. Два режима не видят друг друга, поэтому подписка и политика публикации указывают один и тот же режим: RedisPubSub::new(..).mode(PubSubMode::Sharded) и pubsub::Publish::new().mode(PubSubMode::Sharded).

Когда соединение обрывается​

Брокер переподключается после обрыва соединения, перезапуска Redis или переключения при отказе, и каждая подписка Pub/Sub подписывается заново. Сообщения, опубликованные за время отсутствия, не доставляются. Шардированная подписка не восстанавливается после того, как перераспределение слотов переносит её слот на другой узел.

Когда вместо Pub/Sub нужен поток​

Pub/Sub подходит для того, что полезно только сейчас: обновить кеш, показать живую панель мониторинга, уведомить каждый экземпляр. Если каждый читатель должен получить каждое сообщение, включая опубликованные, пока он не работал, используйте поток и по одной группе потребителей на читателя. Каждая группа читает каждую запись, а запись остаётся в потоке, пока её не обрежут.

Тесты в файле примера публикуют изменение цены и отправку и проверяют, что каждая подписка получила опубликованное.