redb.Route.RabbitMQ

Про Kafka в этой серии уже было. Теперь — RabbitMQ, и с упором на то, как им пользоваться. Коннектор redb.Route.RabbitMQ — поверх официального RabbitMQ.Client 7.x, но писать вам придётся не «клиент», а маршруты, где весь брокер задаётся одной строкой-URI:

From("rabbitmq://orders?exchange=shop&exchangeType=topic&routingKey=order.*&concurrentConsumers=4")
    .Log("Order in: ${body}")
    .To("direct://process");

Прочитали — и уже знаете, что происходит: слушаем очередь orders, привязанную к topic-обменнику shop по маске order.*, в четыре потока, логируем и отдаём дальше. Ни «створи ченнел», ни «забинди», ни «поставь QoS» — всё это коннектор.

Про «уходим от MassTransit» сразу оговорюсь честно: дело не в том, что MassTransit чего-то не умеет — умеет, и RPC (request/response через IRequestClient<T>) в том числе. Разница в модели. MassTransit — это шина с контрактами сообщений, консьюмерами и конфигурацией автобуса: покупаешь всю его картину мира целиком. redb.Route — это явные маршруты в духе Apache Camel: endpoint — строка-URI, паттерны интеграции — шаги маршрута, транспорт — абстракция под ними. Меньше церемоний с контрактами и регистрациями, больше явного «откуда → куда → что по дороге». Что удобнее — вопрос вкуса и задачи; эта статья — про то, как RabbitMQ выглядит во второй модели.

Дальше — как этим пользоваться: анатомия URI, все параметры справочником (что и зачем), заголовки, которые ставит консьюмер, конкурирующие консьюмеры (многопоточность), широковещание по нодам кластера (реальный пример из прода), RPC (у RabbitMQ он получается особенно изящным) и два EIP-паттерна, которые на брокер ложатся идеально — Filter (на компилируемых предикатах) и Dead Letter Channel.

Весь код — на английском, весь текст — по-русски. Примеры не выдуманные — это redb.Route/demos/redb.Route.Demo/Routes, тот же демо-проект, что гоняется в CI.

Цикл про redb и redb.Route. Это продолжение серии, свежие статьи — сверху:

• redb.Route — уходим от MassTransit, идём к Apache Camel: Kafka, Scatter‑Gather и транзакции

• Apache Camel под .NET: HTTP-коннектор без ASP.NET MVC + Content-Based Router

• redb.Route — Apache Camel для .NET, который мы написали потому что выхода другого не было

• redb — типизированное хранилище для .NET поверх Postgres/MSSQL: без миграций, без Include, с полным LINQ

Полный список — в профиле. Исходники: github.com/redbase-app. Про саму БД: redb.ru.

Endpoint — это строка:

rabbitmq://queue-name?host=localhost&exchange=my-exchange&exchangeType=topic&routingKey=order.*

Схема rabbitmq://, хост URI — имя очереди, дальше query-параметры (имена — в таблице ниже). Одна и та же строка работает и консьюмером (в From(...)), и продюсером (в .To(...)) — роль определяется тем, где endpoint стоит в маршруте. Подключение можно не дублировать: заведите фабрику подключения в реестре и сошлитесь через connectionFactory=имя — host/креды/SSL лягут в одно место, соединение переиспользуется.

Важная деталь: значения параметров — это шаблоны. routingKey=order.new — константа, а routingKey=order.${header.type} — динамика: ключ вычисляется на каждое сообщение из его заголовка (тот же механизм, что и ключ партиционирования у Kafka). Так что «строка» здесь не значит «только статика» — ${...} внутри значения превращает его в выражение.

Про флюент. У коннектора есть и билдер Rabbit.Queue("orders")…, но он expression-first: строковые методы принимают выражение, чтобы значение продюсера можно было вычислять на каждое сообщение (.RoutingKey(Header("order.type")) — динамический ключ, как у Kafka). Для константы там оборачивают (.Host(Constant("localhost"))). В этой статье я пишу через URI — для статичной конфигурации это короче и нагляднее; про динамику через выражения — отдельно.

Главная причина сохранить статью в закладки. Имена — как в URI, дефолты — из коробки.

Валидатор на старте не даст выстрелить в ногу: prefetch/concurrentConsumers/timeout должны быть > 0, а autoAck + transacted вместе прямо запрещены.

Коннектор мапит свойства сообщения RabbitMQ на заголовки exchange в обе стороны — и на консьюмере, и на продюсере. Консьюмер, приняв доставку, раскладывает BasicProperties по заголовкам; продюсер на отправке читает их обратно и ставит на исходящее сообщение. Значит consume → produce переносит все AMQP-свойства насквозь, а любое из них можно задать/переопределить, просто выставив заголовок.

Именование в два яруса.

Метаданные доставки — стандартного AMQP-имени нет, поэтому префикс redbRmq.:

Стандартные свойства сообщения — по общеизвестным именам, без префикса (их и так все знают, и round-trip несёт их как есть):

Особенно полезен redbRmq.Redelivered: по нему отдельно обработать «а это уже вторая попытка» — увести повторную доставку в другой маршрут вместо бесконечного requeue.

ReplyTo-заголовок ≠ параметр replyTo. Похожие имена — разные вещи. URI-параметр replyTo=true включает клиентский RPC-режим (см. ниже). А заголовок ReplyTo — это само AMQP-свойство reply-to: выставив его на обычной отправке, вы задаёте адрес reply-очереди сообщения напрямую, и продюсер прокинет заголовок в свойство. Всё это работает потому, что маппинг свойств двусторонний — не только консьюмер их читает, но и продюсер уважает.

Fanout-обменник рассылает копию всем подписчикам, ключ игнорируется. Продюсер публикует в обменник, каждый консьюмер объявляет свою очередь и биндит:

// издатель — публикует в fanout-обменник (копия всем подписчикам)
From("timer://tick?period=5000")
    .SetBody(_ => BuildEvent())
    .To("rabbitmq://events-pub?exchange=events&exchangeType=fanout");

// два независимых подписчика — каждый со своей очередью
From("rabbitmq://audit?exchange=events&exchangeType=fanout&declare=true").To("direct://audit");
From("rabbitmq://email?exchange=events&exchangeType=fanout&declare=true").To("direct://email");

Нужна маршрутизация похитрее — берёте exchangeType=topic и routingKey с масками (order.*, #.error). Меняется только тип обменника и ключ.

Один параметр — concurrentConsumers:

From("rabbitmq://orders?concurrentConsumers=8&prefetchCount=16")   // до 8 сообщений параллельно
    .Process(HandleOrder);

Это единственный рычаг параллелизма консьюмера: он задаёт и AMQP dispatch concurrency у канала, и app-level семафор. Дефолт — 1 (строго серийно, порядок сохранён). Ставите N > 1 — порядок на очереди больше не гарантируется, обработчик должен быть потокобезопасным, а prefetchCount держите ≥ N.

Ремарка для тех, кто щупал коннектор раньше: до версии 3.2.2 concurrentConsumers из-за одной ловушки в RabbitMQ.Client 7.x молча не давал параллелизма (канал был зажат в серийную выдачу). С 3.2.2 это единственный и работающий рычаг — если вы ставили N > 1 и «оно как-то не ускорялось», обновитесь.

Подтверждения — тоже параметром. По умолчанию at-least-once: ack после успешного прохода, nack+requeue при ошибке. Нужна семантика «получил и забыл» — autoAck=true переключает на at-most-once (брокер гасит доставку на выдаче, ошибка не возвращает сообщение).

А если параллелить надо не приём, а обработку внутри маршрута (источник серийный, но работа тяжёлая) — есть ортогональный EIP-шаг .Threads(N). concurrentConsumers масштабирует чтение с очереди, .Threads — обработку; комбинируются свободно.

Реальный паттерн из прода. Сервис работает кластером из нескольких нод, у каждой — свой локальный кеш. Пишем в БД на любой ноде — и хотим, чтобы кеш обновился на всех. Классика поверх RabbitMQ: продюсер аудита публикует доменное событие в durable topic-обменник, а каждая нода вешает на него свою временную очередь и слушает.

Продюсер (там, где идёт запись в БД) — по маршруту на тип события, все в один обменник lt.events, каждый со своим routing key. Заголовки проставляем сразу, чтобы подписчики роутили/фильтровали, не разбирая тело. Подключение — через общую фабрику:

// общий stamp: доменные заголовки для подписчиков — маршрутизация без парсинга тела
Action<IExchange> stamp = e =>
{
    var a = e.In.GetBody<AuditEvent>();
    e.In.Headers["CorrelationId"] = a.CorrelationId;
    e.In.Headers["EntityType"]    = a.EntityType;
    e.In.Headers["EventType"]     = a.EventType.ToString();
};

From("direct://audit.entity.created")
    .RouteId("audit-entity-created")
    .Process(stamp)
    .SetBody(e => JsonSerializer.Serialize(e.In.GetBody<AuditEvent>()))
    .To("rabbitmq://?connectionFactory=RabbitMQ&exchange=lt.events&exchangeType=topic"
        + "&routingKey=dal.events.created&exchangeDurable=true");

// updated / deleted / batch / error — тот же обменник, routingKey=dal.events.{updated|deleted|…}

Консьюмер на каждой ноде — вот тут вся соль. queue= пустой + autoDelete=true: каждая нода получает свою временную очередь (имя выдаёт сервер), привязанную к общему обменнику. Нода отключилась — очередь исчезла. Значит копию каждого события получает каждая нода:

// created и updated — два консюмера в одну общую логику
From("rabbitmq://?connectionFactory=RabbitMQ&exchange=lt.events&exchangeType=topic"
        + "&routingKey=dal.events.created&queue=&autoDelete=true&expires=300000"
        + "&concurrentConsumers=2&prefetchCount=10")
    .RouteId("cache-consumer-created")
    .To("direct://cache-update");

From("rabbitmq://?connectionFactory=RabbitMQ&exchange=lt.events&exchangeType=topic"
        + "&routingKey=dal.events.updated&queue=&autoDelete=true&expires=300000"
        + "&concurrentConsumers=2&prefetchCount=10")
    .RouteId("cache-consumer-updated")
    .To("direct://cache-update");

// один процессор на оба потока — обновляет локальный кеш этой ноды
From("direct://cache-update")
    .RouteId("cache-update")
    .Process(new CacheUpdateProcessor(/* … */));

Ключевое — контраст с Рецептом 2. Там competing consumers: одна общая durable-очередь, работа делится между консюмерами (каждое сообщение достаётся кому-то одному). Здесь наоборот: у каждой ноды своя ephemeral-очередь на общий обменник → широковещание, событие получают все. Один коннектор, противоположная топология — разница в одном параметре (queue=имя&durable=true против queue=&autoDelete=true). Запись в БД на любой ноде → одно событие в обменник → локальный кеш обновился на всех нодах, без опроса и без единой точки отказа.

Вот где особенно приятно. Классический RPC на RabbitMQ — это временная reply-очередь, correlationId, ожидание ответа. Здесь — один параметр на клиенте и ничего на сервере. Вот как это задано в демо (DemoEndpoints.cs):

// сервер — обычный консьюмер, declare создаёт очередь
public const string RabbitConsumer =
    "rabbitmq://demo-rpc-queue?host=localhost&port=5672&username=admin&password=admin&declare=true";

// клиент — replyTo=true включает RPC, timeout=15 — ждать ответ не дольше 15 с
public const string RabbitProducer =
    "rabbitmq://demo-rpc-queue?host=localhost&port=5672&username=admin&password=admin&replyTo=true&timeout=15";

Клиент — обычный .To(...), только он не «выстрелил и забыл», а дождался ответа и поехал дальше уже с ним (MainPipelineRoutes.cs):

.Log("[3-RABBIT] → Sending to RabbitMQ RPC...")
.To(RabbitProducer)                                      // публикуем запрос и ЖДЁМ ответ
.Log("[3-RABBIT] ← stamp.rabbit=${header.stamp.rabbit}") // ответ уже в exchange

Сервер — просто консьюмер, ему для RPC настраивать не надо ничего. Коннектор сам видит reply-to на входящем сообщении, делает маршрут InOut и отправляет то, что лежит в Out, обратно в reply-очередь с тем же correlationId. Весь «сервер» из демо:

From(RabbitConsumer)
    .RouteId("demo-rabbit-worker")
    .Log("[RABBIT-W] ▶ Received: ${body}")
    .SetHeader("stamp.rabbit", e => $"ok:{DateTime.UtcNow:HH:mm:ss.fff}")
    .Log("[RABBIT-W] ◀ Stamped, replying");         // reply уходит сам

Ни слова про reply-очередь и корреляцию — клиент отправил запрос и получил обратно заголовок stamp.rabbit, проставленный сервером. Чистый request/reply поверх брокера, без ручной возни с временными очередями.

Тут и виден контраст с MassTransit. Там request/response тоже есть — но это IRequestClient<TRequest>, типизированные контракты TRequest/TResponse и консьюмер, зарегистрированный в автобусе. Мощно, но это отдельный слой абстракций, в который надо войти. Здесь — replyTo=true в URI, а на сервере вообще ничего: обычный консьюмер, коннектор сам увидел reply-to и ответил. Разные подходы к одной задаче: контракты и шина против одного флага на endpoint.

Message Filter — простой и постоянно нужный EIP: пропустить дальше только те сообщения, что удовлетворяют условию, остальные — молча отбросить. В redb.Route это .Filter(...) … .EndFilter() (или короче .End() — это алиас): блок между ними выполняется только для прошедших фильтр, для остальных маршрут на этом заканчивается. И вот тут разворачивается компилируемый предикатный движок — тот самый, что мы обсуждали в теме про выражения.

Условие задаётся тремя способами. Предикатом по заголовку (или телу):

From("rabbitmq://orders?concurrentConsumers=4")
    .Filter(Header("score").isGreaterThan(50))       // score > 50
        .To("rabbitmq://vip-orders")
    .EndFilter();

Набор предикатов — весь, что ожидаешь: .isEqualTo / .isNotEqualTo / .isGreaterThan / .isLessThan / .isGreaterThanOrEqualTo / .isLessThanOrEqualTo / .isBetween(a, b) / .contains(...). Слева — Header("...") или Body():

.Filter(Header("email").contains("@example"))   .To("direct://internal") .EndFilter();
.Filter(Header("grade").isEqualTo("B+"))         .To("direct://grade-b")  .EndFilter();

Строковым выражением — ${...} резолвится, результат приводится к boolean. Для truthy-проверки — просто значение; для сравнения оберните его в logical(...) (голое ${header.score > 50} внутри шаблона вернёт пустую строку — так писать нельзя):

.Filter("${header.status}")                  // 'active' → truthy → пропустит
    .Log("active only")
.EndFilter();

.Filter("${logical(header.score > 50)}")     // сравнение → "True"/"False"
    .To("rabbitmq://vip")
.EndFilter();

Или лямбдой, когда нужна произвольная логика на C#:

.Filter(ex => ex.In.Body is string s && s.Contains("urgent"))
    .To("rabbitmq://urgent")
.EndFilter();

Фильтровать удобно как раз по тем заголовкам, что консьюмер уже проставил, — например, .Filter(Header("redbRmq.RoutingKey").contains("order")), чтобы в блок попадали только сообщения с нужным ключом маршрутизации.

Под капотом Filter на каждое сообщение вычисляет выражение в object? и приводит к bool — по правилам ConvertToBoolean:

bool   → как есть
string → "true"/"false" парсятся; иначе непустая строка = true, пустая = false
null   → false
прочее (число/объект, не-null) → true

Отсюда — поведение трёх форм:

• Предикат Header("amount").isGreaterThan(1000) — это IPredicate, у которого уже готовый Func<IExchange,bool>. Коэрция не нужна: настоящий boolean прямо от сравнения. Компилируется в делегат и кэшируется — не рефлексия на каждом сообщении, а быстрый вызов.

• Строка-шаблон "${header.status}" — ${...} интерполируется в значение, дальше truthy-правила. status="active" → непустая строка → true (потому в демо и «active is truthy»); а "${header.enabled}" при "false" → bool.TryParse → false. Сравнение прямо в шаблоне ("${header.score > 50}") не считается — вернёт пустую строку (то есть false); для сравнения нужен logical(...), как показано выше.

• Лямбда ex => … — что вернули, то и предикат.

То есть «булево выражение» — это либо честный bool от предиката/сравнения, либо truthy-приведение интерполированного значения. Пустая строка, null и "false" — отсекают; непустая строка, число, true — пропускают.

Обрати внимание на один момент. Строковой и предикатной форм Filter(string) / Filter(IPredicate) нет в интерфейсе IRouteDefinition — там только канонические Filter(Func<IExchange,bool>) и Filter(IExpression). Строковая форма — это extension-метод, тонкий фасад над каноническим:

// весь строковый overload целиком (redb.Route/…/RouteDefinitionCamelDslExtensions.cs):
public static FilterDefinition Filter(this IRouteDefinition self, string simpleTemplate)
    => self.Filter(new StringExpression(simpleTemplate));   // оборачиваем строку в шаблон-выражение

Отсюда важное следствие: DSL расширяемый, и свой шаг добавляется ровно так же — extension-методом с делегированием в канонический. Захотелось собственный «глагол» в маршруте — пишете:

// свой шаг: пропустить только "urgent" в теле
public static FilterDefinition OnlyUrgent(this IRouteDefinition self)
    => self.Filter(ex => ex.In.Body is string s && s.Contains("urgent"));

// и в маршруте он выглядит как родной:
From("rabbitmq://orders")
    .OnlyUrgent()
        .To("rabbitmq://urgent")
    .EndFilter();

Никакой магии и никаких правок в движке — просто метод расширения над IRouteDefinition. Ровно так же в самом фреймворке сделаны и Camel-совместимые формы, и строковые шаблоны, и предикатные overload'ы. То есть если вам не хватает какого-то шага — вы его дописываете сами, не форкая коннектор.

Здесь мы задели только край выражений — Header(...), Body(), ${...}, logical(...). А вообще выражения в redb.Route — это отдельный компилируемый язык: доступ к телу и заголовкам, свойства exchange, индексаторы (items[0]), строковые функции, JSONPath и XPath. Всё это токенизируется, парсится в AST и компилируется в делегат (System.Linq.Expressions, не рефлексия), с кэшем. Тема большая — про сам движок будет своя статья, он того заслуживает.

Из паттернов Хопе и Вульфа на RabbitMQ идеально ложится Dead Letter Channel — потому что у брокера это встроенная фича (x-dead-letter-exchange), а не эмуляция в коде. Идея: сообщение, которое не обработалось (или протухло по TTL, или не влезло в лимит очереди), не теряется и не крутится вечно в requeue, а уезжает в «морг» — отдельный dead-letter обменник, откуда его можно спокойно разобрать.

Настраивается прямо в URI рабочей очереди:

// рабочая очередь: при nack без requeue / протухании / переполнении — сообщение уходит в DLX
From("rabbitmq://orders?declare=true"
        + "&deadLetterExchange=orders.dlx"          // куда
        + "&deadLetterRoutingKey=orders.failed"     // с каким ключом
        + "&messageTtl=30000"                       // не обработали за 30 с — в DLX
        + "&overflow=reject-publish-dlx&maxLength=100000"   // и переполнение туда же
        + "&concurrentConsumers=4")
    .Process(HandleOrder);

// «морг»: отдельный маршрут слушает DLX — логирует и складывает в redb для разбора
From("rabbitmq://orders.dead?declare=true&exchange=orders.dlx&exchangeType=direct&routingKey=orders.failed")
    .Log("[DLQ] dead-lettered: ${body}")
    .ProcessWithRedb((redb, ex) => redb.SaveAsync(BuildDeadLetter(ex)));

Никакого самописного retry-цикла и «поспал-повторил» — надёжность отдаётся брокеру, а у вас остаётся чистый маршрут-разгребатель, который пишет проблемные сообщения в базу. И очередь не разбухает под наплывом: лишнее сразу уезжает в DLX, а не копится в памяти брокера.

RabbitMQ в redb.Route — это маршруты, где весь брокер задан одним URI: From("rabbitmq://…") / .To("rabbitmq://…"), а вся возня с каналами, QoS, биндами, reply-очередями и dead-letter спрятана за параметрами строки. Обменники всех типов, durable/quorum-очереди, приоритеты, TTL, DLX, авто-recovery для кластера, TLS/mTLS — набор полный, а на выходе — короткая и понятная строка. Конкурирующие консьюмеры включаются одним параметром, RPC — одним флагом на клиенте и нулём настроек на сервере, Filter отсекает лишнее компилируемыми предикатами, а Dead Letter Channel — парой параметров поверх нативного DLX.

Полный демо-проект — в redb.Route/demos/redb.Route.Demo, поднимается на локальном RabbitMQ (admin/admin) и гоняется как self-test. Ловится что-то в вашем сценарии — пишите в комментариях.

Исходники и релизы: github.com/redbase-app. Про БД redb: redb.ru. Прошлые статьи — в профиле.

If this was useful — a ⭐ on GitHub helps others find it.