redb.Route.Redis

Коннектор Redis для redb.Route: все структуры данных как шаги маршрута, Pub/Sub, Streams с группами потребителей, список как очередь и Claim Check на Redis.

В интеграционном проекте Redis редко бывает чем‑то одним. Это кэш ответов внешних систем, счётчики и блокировки, шина событий через Pub/Sub, журнал сообщений через Streams, очередь задач на списке. Обычно каждая из этих ролей живёт в своём куске кода со своим подключением, своими повторами и своим логированием.

В redb.Route все они собраны в одном коннекторе redb.Route.Redis. Любая операция Redis становится шагом маршрута, а подписка, стрим или список становятся его входом. Подключение, телеметрия и плавная остановка при этом общие с остальными коннекторами. Под капотом StackExchange.Redis 2.11.8.

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

dotnet add package redb.Route.Redis

services.AddRedbRoute(route =>
{
    route.Services.AddRedbRouteRedis();
    route.AddRouteBuilder<OrderRoutes>();
});

Адрес эндпоинта устроен так: redis:ОПЕРАЦИЯ:ресурс?параметры. Первый сегмент пути это операция Redis (регистр не важен), всё после первого двоеточия это ключ, канал или имя стрима. Двоеточия внутри ключа сохраняются, поэтому redis:SET:session:42?ttl=300 записывает ключ session:42 со сроком жизни 300 секунд.

То же самое собирает fluent‑построитель, и дальше в статье он используется везде:

.To(Redis.Set("session:42").Ttl(300))   // redis:SET:session:42?ttl=300

Для ключей, Pub/Sub, стримов и списков в Redis есть готовые фабрики (Set, Get, Incr, Publish, Subscribe, XAdd, XRead, LPush и остальные). Любую другую операцию задаёт Redis.Command(операция, ключ): например, Redis.Command("HSET", "customer:42").Field("email").

Продюсер поддерживает операции над всеми основными структурами Redis:

Откуда операция берёт данные, определено одинаково для всех:

• значение или элемент это тело сообщения. В списках, хэшах, множествах, sorted sets и стримах тело byte[] пишется как есть, без перевода в строку;

• скалярные параметры задаются в адресе: field для хэшей, score, minScore и maxScore для sorted sets, start и stop для диапазонов, longitude, latitude, member1, member2 и geoUnit для гео, offset и bit для битовых карт;

• наборы значений передаются заголовками: словарь полей для HMSET (redbRedis.HashFields), имена полей для HMGET (redbRedis.FieldNames), исходные ключи для PFMERGE (redbRedis.SourceKeys), центр и радиус для GEORADIUS, аргументы для COMMAND.

Ключ, поле, канал и имя стрима могут содержать шаблоны ${...}: они вычисляются на каждом сообщении.

Результат операции становится телом для следующего шага. GET отдаёт строку или null, INCR новое значение счётчика, LRANGE и SMEMBERS массив строк, HGETALL словарь, GEORADIUS имена найденных точек:

From("direct://order-created")
    .To(Redis.Incr("stats:orders:today"))
    .Log("заказов сегодня: ${body}");

SETNX записывает значение, только если ключа ещё нет, и возвращает "OK" или null. Вместе с Ttl(...) это готовая основа для блокировки на время обработки или для отсечения повторов.

From("direct://publish-event")
    .To(Redis.Publish("orders.events"));

From(Redis.Subscribe("orders.events"))
    .To("direct://notify");

После публикации число получателей лежит в теле и в заголовке redbRedis.Publish.Recipients. У подписчика тело это текст сообщения, а в заголовках канал (redbRedis.Channel), тип (PubSub) и время получения.

Подписка по шаблону делается через PSUBSCRIBE: From(Redis.PSubscribe("orders.*")) получает сообщения всех каналов, подходящих под шаблон, а в заголовке redbRedis.Channel лежит канал, в который сообщение пришло. Параметр usePattern превращает в подписку по шаблону и обычный SUBSCRIBE.

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

Запись в стрим:

From("direct://order-created")
    .To(Redis.XAdd("orders").StreamMaxLength(100_000));

Тело попадает в поле data вместе с полем timestamp в миллисекундах. Если нужен свой набор полей, передайте словарь в заголовке redbRedis.StreamFields. StreamMaxLength обрезает стрим по длине, по умолчанию приблизительно (MAXLEN ~); точную обрезку включает StreamApproximate(false). Идентификатор новой записи возвращается в теле и в заголовке redbRedis.Stream.MessageId.

Чтение группой потребителей:

From(Redis.XRead("orders")
        .ConsumerGroup("billing")
        .ConsumerName("node-1")
        .StreamReadCount(50)
        .StreamClaimMinIdle(60_000))
    .To("direct://bill");

Что происходит под капотом:

• группа создаётся сама при старте, вместе со стримом, если его ещё нет. По умолчанию она начинает с конца стрима и получает записи, добавленные после её создания; StreamStartPosition("0") начинает с первой записи. Если группа уже существует, консюмер подключается к ней;

• записи читаются пачками по streamReadCount. Когда новых записей нет, консюмер ждёт streamBlockTimeMs (по умолчанию 1000 мс) и опрашивает снова;

• тело это словарь полей записи, каждое поле дополнительно лежит в заголовке redbRedis.Stream.<поле>, а идентификатор записи в redbRedis.MessageId;

• имя потребителя по умолчанию равно имени машины.

Несколько узлов с одной группой и разными именами потребителей делят стрим между собой: каждую запись получает один из них. Так один стрим обрабатывается кластером без дублей.

Консюмер группы подтверждает запись (XACK) после того, как вся единица работы завершилась успешно, включая транзакцию маршрута. Запись, обработка которой упала, не подтверждается и остаётся в списке ожидающих группы. Так получается доставка «хотя бы один раз».

Вернуть такие записи в работу помогает StreamClaimMinIdle(мс). На каждом опросе консюмер забирает командой XAUTOCLAIM ожидающие записи группы, простоявшие не меньше заданного времени, и обрабатывает их заново: и свои упавшие, и оставленные потребителем с упавшего узла. Время простоя стоит брать больше самого долгого маршрута, иначе запись заберут, пока её ещё обрабатывают.

Для потоков, где потеря записи допустима, есть StreamNoAck(): чтение идёт с NOACK, запись считается доставленной в момент чтения и в список ожидающих не попадает (не более одного раза). С StreamClaimMinIdle этот режим не сочетается, эндпоинт с обеими опциями не создаётся.

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

From(Redis.XRead("orders").StreamStartPosition("0"))
    .To("direct://rebuild-projection");

Без StreamStartPosition потребитель начинает с записей, добавленных после его запуска, "0" читает с первой записи, а идентификатор записи продолжает после неё. Позиция > означает «ещё не выданное группе» и имеет смысл только для группы: без группы консюмер с ней не создаётся. Упавшую запись без группы ничто не удерживает, и чтение идёт дальше.

From("direct://enqueue-job")
    .To(Redis.LPush("jobs"));

From(Redis.Command("BRPOP", "jobs")
        .ProcessingList("jobs:processing")
        .PollDelay(500))
    .To("direct://run-job");

LPUSH в голову и чтение с хвоста дают очередь в порядке поступления. Консюмер забирает элементы по одному опросом, а когда список пуст, ждёт pollDelayMs. BLPOP и BRPOP здесь имена операций, а не блокирующие команды: блокирующее чтение заняло бы общее соединение, через которое работает весь процесс.

С ProcessingList очередь надёжная. Элемент переносится командой LMOVE в список «в работе» и удаляется оттуда после успешного маршрута. Упавший элемент атомарно возвращается в очередь, туда, откуда его взяли, и обрабатывается снова, а то, что осталось в списке «в работе» после прошлого запуска, при старте возвращается в очередь. На один список «в работе» приходится один консюмер. LMOVE есть в Redis начиная с версии 6.2.

Без ProcessingList элемент снимается со списка до обработки и при сбое маршрута теряется. Такой режим подходит для задач, которые не жалко потерять, например для прогрева кэша.

Самый короткий путь это параметры в адресе: connectionString (по умолчанию localhost:6379), database и password. Пароль помечен как секрет и в логах скрыт.

Для продакшена удобнее именованная фабрика подключений в реестре контекста, тогда секреты не попадают в адрес маршрута:

context.AddToRegistry("prod", new RedisConnectionFactory
{
    ConnectionString = "redis-1:6379,redis-2:6379",
    User = "route",
    Password = secrets.RedisPassword,
    Ssl = true,
    SslProtocols = "Tls12, Tls13",
});

From("direct://rates")
    .To(Redis.Set("rates:usd").ConnectionFactory("prod").Ttl(300));

Фабрика знает ACL‑пользователя Redis 6+, TLS с выбором протоколов, Sentinel (ServiceName), таймауты подключения и операций, keep‑alive, политику переподключения (экспоненциальную или линейную) и префикс каналов. Ошибки конфигурации не маскируются: имя фабрики, которого нет в реестре, и опечатка в SslProtocols дают явную ошибку при подключении вместо тихого перехода на параметры по умолчанию. То же с параметрами адреса: опцию, которой эндпоинт не знает, или значение не того типа он не принимает и называет такой параметр по имени, а не теряет его молча.

Каждый эндпоинт держит одно подключение StackExchange.Redis и переподключается сам. Потеря и восстановление связи пишутся в лог.

Шаблон Claim Check из каталога Enterprise Integration Patterns убирает тяжёлое тело из сообщения на время маршрута и возвращает его, когда оно снова понадобится. Для него есть хранилище на Redis, RedisClaimCheckRepository:

var claims = new RedisClaimCheckRepository(
    new RedisConnectionFactory { ConnectionString = "redis.internal:6379" },
    defaultTtl: TimeSpan.FromHours(1));

From("direct://inbound-invoice")
    .ClaimCheck(claims, ClaimCheckOperation.Push)   // тело в Redis, в сообщении ключ
    .To("direct://route-by-headers")
    .ClaimCheck(claims, ClaimCheckOperation.Pop)    // тело вернулось, запись удалена
    .To("direct://archive");

Хранилище пишет данные как есть, в двоичном виде, и ставит на запись родной срок жизни Redis. Ключи получают префикс redb:claimcheck:, его можно заменить. Чтение с удалением атомарно: оно идёт одним Lua‑скриптом, поэтому работает и на версиях Redis до 6.2, где ещё нет GETDEL.

Каждый вызов продюсера открывает span OpenTelemetry с именем redis <ОПЕРАЦИЯ>, тегом db.system=redis, ресурсом и операцией. Вызовы Redis видны в той же трассе, что и остальные шаги маршрута.

Консюмеры останавливаются плавно, как и в других коннекторах redb.Route: перестают брать новые сообщения и дорабатывают те, что уже в обработке.

dotnet add package redb.Route
dotnet add package redb.Route.Redis

Исходники: github.com/redbase‑app/redb‑route. Пакет на NuGet.

Если было полезно, ⭐ на GitHub поможет другим это найти.

Другие мои статьи — redb.ru/articles, ещё — на Хабре.