Выбор транспорта

Это руководство сравнивает три структуры Redis, которые читает сервис: поток, список и канал Pub/Sub.

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

Потоки: надёжный журнал

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
    amount_cents: u64,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "invoices")]
struct Invoice {
    order_id: u64,
    amount_cents: u64,
}

#[subscriber(RedisStream::new("orders").group("billing"), reply)]
async fn bill(order: &Order) -> Invoice {
    tracing::info!(order.id, "order billed");
    Invoice {
        order_id: order.id,
        amount_cents: order.amount_cents,
    }
}

Вручную

src/main.rs

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

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

named!(Order, "orders");
named!(Invoice, "invoices");

struct Bill;

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

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

Заказы - это деньги, поэтому они идут через поток. bill отвечает на каждый заказ счётом Invoice, который попадает в поток invoices.

Списки: очередь заданий

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "thumbnails")]
struct Thumbnail {
    image: String,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "thumbnails.ready")]
struct ThumbnailReady {
    image: String,
}

#[subscriber(RedisList::new("thumbnails").reliable(), reply)]
async fn render(job: &Thumbnail) -> ThumbnailReady {
    tracing::info!(job.image, "thumbnail rendered");
    ThumbnailReady {
        image: job.image.clone(),
    }
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Thumbnail {
    image: String,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct ThumbnailReady {
    image: String,
}

named!(Thumbnail, "thumbnails");
named!(ThumbnailReady, "thumbnails.ready");

struct Render;

impl Handle<Thumbnail, ThumbnailReady> for Render {
    fn handle(
        &self,
        job: &Thumbnail,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<ThumbnailReady, HandlerOutcome>> {
        tracing::info!(job.image, "thumbnail rendered");
        ready(Ok(ThumbnailReady {
            image: job.image.clone(),
        }))
    }
}

Список Redis - это очередь: производитель добавляет запись, и ровно один потребитель её извлекает. RedisList::new извлекает запись командой BRPOP. Запись покидает Redis до запуска обработчика, и сбой в обработчике её теряет. reliable вместо этого переносит запись в список обработки и удаляет её только после подтверждения обработчиком. У списка нет групп и нет повторного чтения.

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

Pub/Sub: широковещательная рассылка

Макросы

src/main.rs

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

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

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct PriceChange {
    sku: String,
    price_cents: u64,
}

named!(PriceChange, "prices");

struct RefreshPrice;

impl Handle<PriceChange> for RefreshPrice {
    fn handle(
        &self,
        change: &PriceChange,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(change.sku, change.price_cents, "price cache refreshed");
        ready(Ok(()))
    }
}

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

Изменение цены важно только экземплярам, которые работают сейчас и обновляют свой кеш, поэтому оно идёт через канал.

Сравнение

	Поток	Список	Pub/Sub
Хранится, пока никто не читает	пока не обрезан	пока не извлечён	нет
Кто получает сообщение	один потребитель в каждой группе	один потребитель	каждый подключённый подписчик
ack	XACK	надёжный: LREM из списка обработки; простой: ничего	ничего
retry	копия, добавленная в поток	надёжный: обратно в список; простой: теряется	теряется
retry_after	копия после задержки	копия после задержки	копия после задержки
Пакет	одно чтение до n записей	собирает клиент	собирает клиент
Повторное чтение	перемотка группы	нет	нет

Страницы о подтверждении описывают каждый исход полностью, а руководство Переобработка, задержки и dead-letter показывает, как работают задержка и предел попыток на каждой структуре. Пакет на потоке - это COUNT одного чтения. Список и канал доставляют по одной записи, поэтому пакет того же размера собирает клиент.

Одно имя занимает одна структура. Поток и список - это ключи определённого типа, и запись другого типа возвращает ошибку WRONGTYPE. Канал вообще не ключ: XADD под именем канала создаёт поток, которого подписчики канала никогда не увидят.

Один сервис, три транспорта

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(bill).out_reply(stream::Publish);
            b.include(render).out_reply(pubsub::Publish::default());
            b.include(refresh_price);
        },
    )
}

Вручную

src/main.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(
                subscriber(RedisStream::new("orders").group("billing"), Bill)
                    .reply()
                    .build(),
            )
            .out_reply(stream::Publish);
            b.include(
                subscriber(RedisList::new("thumbnails").reliable(), Render)
                    .reply()
                    .build(),
            )
            .out_reply(pubsub::Publish::default());
            b.include(subscriber(RedisPubSub::new("prices"), RefreshPrice).build());
        },
    )
}

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

Сервис на одном транспорте импортирует прелюдию этого транспорта, например ruststream_fred::stream::prelude. Этот сервис сочетает транспорты, поэтому импортирует ruststream_fred::prelude. В ней политика публикации каждого транспорта сохраняет имя своего модуля: stream::Publish пишет командой XADD, pubsub::Publish командой PUBLISH. Тип ответа задаёт, куда уходит ответ, а политика в точке монтирования задаёт, как он туда попадает.

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

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