Redis Cluster

Это руководство запускает сервис на Redis Cluster и даёт ключам имена так, чтобы ключи одной команды находились в одном хеш-слоте.

Пример - серверная часть магазина. Она выполняет задания из очереди заданий, выставляет счета по заказам из потока и слушает события в шардированном канале.

Подключение

Макросы

src/main.rs

fn service(broker: RedisBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(broker, |b| {
        b.include(run_job);
        b.include(invoice.pipeline().atomic());
        b.include(on_event);
    })
}

/// Two seeds: the client reaches the cluster through whichever answers first.
#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(RedisBroker::cluster([
        "redis-1.internal:6379",
        "redis-2.internal:6379",
    ]))
}

Вручную

src/main.rs

fn service(broker: RedisBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber(RedisList::new("{jobs}").reliable(), RunJob).build());
        let invoice = subscriber(RedisStream::new("{orders}").group("billing"), Invoice);
        b.include(invoice.pipeline().atomic().build());
        let events = RedisPubSub::new("events").mode(PubSubMode::Sharded);
        b.include(subscriber(events, OnEvent).build());
    })
}

/// Two seeds: the client reaches the cluster through whichever answers first.
fn app() -> impl App<State = ()> {
    service(RedisBroker::cluster([
        "redis-1.internal:6379",
        "redis-2.internal:6379",
    ]))
}

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

RedisBroker::cluster принимает начальные узлы. Достаточно одного доступного узла: клиент читает с него устройство кластера и подключается к каждому ведущему узлу. Как и любой конструктор брокера, этот не выполняет ввода-вывода. Соединение открывается при запуске сервиса.

Обработчики одинаковы на одиночном сервере и в кластере. Меняется то, как сервис называет свои ключи.

Хеш-теги

Кластер распределяет ключи по 16384 хеш-слотам, и каждый ведущий узел владеет их диапазоном. Команда, блок MULTI или скрипт, которые затрагивают несколько ключей, выполняются, только если все ключи находятся в одном слоте. Слот определяет хеш-тег: если в ключе есть часть в фигурных скобках, хешируется только она. {jobs} и {jobs}.processing находятся в одном слоте, как и {orders} и {orders}:invoices.

Имя в фигурных скобках не может быть постоянным местом назначения типа, потому что #[outgoing(name = "..")] читает фигурные скобки как шаблон имени. Производитель задаёт такой ключ там, где публикует: .message(&job).to("{jobs}").

Очередь заданий

Макросы

src/main.rs

/// The hash tag `{jobs}` keeps the queue and its processing list `{jobs}.processing` on one slot.
#[subscriber(RedisList::new("{jobs}").reliable())]
async fn run_job(job: &Job) -> HandlerOutcome {
    tracing::info!(job.id, "job done");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// The hash tag `{jobs}` keeps the queue and its processing list `{jobs}.processing` on one slot.
struct RunJob;

impl Handle<Job> for RunJob {
    fn handle(
        &self,
        job: &Job,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(job.id, "job done");
        ready(Ok(()))
    }
}

Надёжный список переносит каждую запись в список обработки одной командой LMOVE, поэтому оба ключа должны находиться в одном слоте. Список обработки называется как очередь с добавленным .processing и сохраняет хеш-тег очереди. Надёжный список, два ключа которого попадают в разные слоты, не запускается, а ошибка называет хеш-тег, который нужно использовать.

Побочные эффекты в слоте потока

Макросы

src/main.rs

/// Invoices an order; the invoice and the acknowledgement commit as one block on the slot of
/// `{orders}`.
#[subscriber(RedisStream::new("{orders}").group("billing"))]
async fn invoice(order: &Order, Ctx(pipeline): Ctx<keys::Pipeline>) -> HandlerOutcome {
    if pipeline
        .lpush("{orders}:invoices", order.id.to_string())
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Invoices an order; the invoice and the acknowledgement commit as one block on the slot of
/// `{orders}`.
struct Invoice;

impl Handle<Order, (), (), PipelineContext> for Invoice {
    async fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, PipelineContext>,
    ) -> Result<(), HandlerOutcome> {
        ctx.context(keys::Pipeline)
            .lpush("{orders}:invoices", order.id.to_string())
            .await
            .map_err(|_| HandlerOutcome::retry())
    }
}

Поток - это один ключ, поэтому подписка на поток работает под любым именем. Атомарное окно помещает команды обработчика и XACK в один блок MULTI, и блок выполняется в слоте потока. invoice пишет в {orders}:invoices, в слот потока {orders}. Команда для ключа в другом слоте заставляет Redis отклонить весь блок: ничего не записывается, и запись потока остаётся неподтверждённой.

Транзакция издателя потока - тоже блок MULTI, и её места назначения не привязаны к одному слоту. Поэтому на кластере сервис с политикой TransactionalPublish не запускается.

Шардированный Pub/Sub

Макросы

src/main.rs

/// Sharded Pub/Sub: the channel lives on the node that owns its slot.
#[subscriber(RedisPubSub::new("events").mode(PubSubMode::Sharded))]
async fn on_event(event: &Event) -> HandlerOutcome {
    tracing::info!(event.kind, "event received");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Sharded Pub/Sub: the channel lives on the node that owns its slot.
struct OnEvent;

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

Классический Pub/Sub в кластере отправляет каждое сообщение на каждый узел. PubSubMode::Sharded отправляет его только на узел, которому принадлежит слот канала, командами SPUBLISH и SSUBSCRIBE, поэтому трафик масштабируется вместе с кластером. Этому режиму нужен Redis 7 или новее.

Два режима не видят друг друга: шардированное сообщение никогда не доходит до классического подписчика, и наоборот. Издатель канала использует режим подписки. Подписка по шаблону бывает только классической.

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

Тесты запускают промышленное приложение через TestApp::start с брокером кластера. Redis внутри процесса ведёт себя как кластер: он отклоняет команду или блок, которые охватывают несколько хеш-слотов. Поэтому тесты проверяют, что каждая форма работает с хеш-тегами сервиса, что надёжный список без тега не запускается и что блок на два слота ничего не фиксирует. То же тело теста работает с настоящим кластером, когда REDIS_CLUSTER_TEST_URL называет его узел.

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