redb.Route — коннектор Amazon SQS + SNS, at-least-once и Publish-Subscribe через SNS→SQS. Уходим от MassTransit / Хабр

series redb ecosystem

Про Kafka и RabbitMQ в этой серии уже было. Теперь — Amazon SQS, и вместе с ним в одном пакете SNS: два транспорта, потому что в мире AWS они ходят парой. Коннектор redb.Route.Sqs — поверх нативного AWS SDK for .NET v4 (AWSSDK.SQS, AWSSDK.SimpleNotificationService), но писать вам придётся не «клиент», а маршруты, где вся очередь задаётся одной строкой-URI:

From("sqs://orders?waitTimeSeconds=20&concurrentConsumers=4")
    .Log("Order in: ${body}")
    .To("direct://process");

Прочитали — и уже знаете, что происходит: длинный поллинг очереди orders по 20 секунд, четыре конкурирующих консьюмера, логируем и отдаём дальше. Ни «создай клиента», ни «ReceiveMessage», ни «DeleteMessage после обработки» — всё это коннектор.

Про «уходим от MassTransit» — та же оговорка, что и в прошлых статьях: дело не в том, что MassTransit чего-то не умеет. SQS/SNS у него есть давно — пакет MassTransit.AmazonSQS, и он крутит ровно ту же связку SNS→SQS. Разница в модели. MassTransit — шина с контрактами сообщений и конфигурацией автобуса. redb.Route — явные маршруты в духе Apache Camel: endpoint — строка-URI, паттерны интеграции — шаги маршрута, транспорт — абстракция под ними. Эта статья — про то, как SQS/SNS выглядит во второй модели.

Отдельно — про то, о чём сейчас спрашивают чаще всего, и здесь важна точность, а не злорадство. MassTransit с v9 перешёл на коммерческую лицензию — линия развивается под компанией Massient, Inc., пакет MassTransit.AmazonSQS 9.0.0 на NuGet уже оттуда. Но без передёргиваний: v8 остаётся под Apache 2.0 и получает security-патчи и критические фиксы как минимум до конца 2026; v9 — не закрытый бинарь, а source-available; а организациям с выручкой меньше $1 млн в год (и НКО с расходами меньше $1 млн) он даётся со 100% скидкой — то есть бесплатно, без коммерческой поддержки. Так что «MassTransit стал платным» — неправда для маленькой команды и правда для крупной, у которой это превращается в решение: платить, оставаться на v8 до конца его поддержки или искать альтернативу. redb.Route — Apache 2.0, и SQS-коннектор из этой статьи в том числе; но берите его не поэтому, а если вам ближе модель маршрутов. Лицензия — повод посмотреть по сторонам, а не аргумент в архитектурном споре.

Весь код — на английском, весь текст — по-русски. Примеры не выдуманы: они гоняются против LocalStack (docker контейнер, эндпоинт http://localhost:4566), ровно как интеграционные тесты коннектора.

Цикл про redb и redb.Route. Это продолжение серии, свежие статьи — сверху:
коннектор RabbitMQ — один URI на endpoint, RPC, конкурирующие консьюмеры и dead-letter (предыдущая в серии)
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.

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

redb.Route.Sqs регистрирует две схемы, и роли у них разные:

sqs:// — очередь. Работает и консьюмером (в From(...)), и продюсером (в .To(...)).
sns:// — топик. Только publisher. У SNS нет pull-консьюмера — доставка это push (в SQS / HTTP / email / SMS). Попытка сделать From("sns://...") честно падает с NotSupportedException: чтобы «читать» топик, подписываете на него SQS-очередь и читаете её через sqs://.

Почему они в одном пакете: у AWS это связка. SQS — надёжная очередь «точка-точка» (одно сообщение — одному потребителю). SNS — топик «издатель-подписчик» (одно сообщение — всем подписчикам). Классическая AWS-архитектура «fan-out» — это SNS-топик, на который подписаны несколько SQS-очередей. Поэтому и коннектор один: sqs:// + sns://.

Регистрация в DI:

services.AddRedbRoute(route =>
{
    route.Services.AddRedbRouteSqs();   // регистрирует и sqs://, и sns://
    route.AddRouteBuilder<MyRoutes>();
});

Анатомия URI

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

sqs://queue-name?region=us-east-1&waitTimeSeconds=20&concurrentConsumers=4
sns://topic-name?region=us-east-1&autoCreateTopic=true

Схема, имя очереди/топика в пути, дальше query-параметры (имена — в таблицах ниже). Флентом — то же самое: Sqs.Queue("orders").WaitTimeSeconds(20)... и Sns.Topic("events").Region("us-east-1")... собирают ту же строку с URL-энкодингом.

Три вещи, специфичные именно для AWS:

Креды не надо дублировать. Двадцать очередей на один аккаунт — заведите AwsConnectionFactory в реестре и ссылайтесь через connectionFactory=имя. Одна фабрика строит и SQS-, и SNS-клиент (креды, регион и service URL у них общие):

context.AddToRegistry("prod-aws", new AwsConnectionFactory
{
    Region = "eu-west-1",
    UseDefaultCredentialsProvider = true,   // IAM role на инстансе / SSO / env
});

From("sqs://orders?connectionFactory=prod-aws&concurrentConsumers=4");
To("sns://events?connectionFactory=prod-aws");

LocalStack / ElasticMQ — через serviceUrl=. Явный URL перебивает региональный эндпоинт (регион остаётся только для подписи запроса). Это и есть режим для локальной разработки и тестов:

sqs://orders?serviceUrl=http://localhost:4566&region=us-east-1&accessKey=test&secretKey=test

Очередь/топик создаются на месте. autoCreateQueue=true (или autoCreateTopic=true) создаёт ресурс на старте, если его нет. Имя с суффиксом .fifo создаётся как FIFO автоматически (с ContentBasedDeduplication=true).

Все параметры

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

Подключение и креды (общие для sqs:// и sns://)

Параметр	Дефолт	Зачем
`region`	`us-east-1`	AWS-регион
`serviceUrl`	(пусто)	Явный эндпоинт (LocalStack/ElasticMQ); перебивает регион
`accessKey` / `secretKey`	(пусто)	Статические ключи
`sessionToken`	(пусто)	Временные (STS) креды
`profileName`	(пусто)	Именованный профиль из shared credentials
`useDefaultCredentialsProvider`	`false`	Дефолтная цепочка AWS (env / IAM role / SSO)
`connectionFactory`	(пусто)	Имя фабрики в реестре; перекрывает креды из URI
`retryCount`	`3`	Ретраи SDK на транзиентных ошибках
`connectionTimeout`	`30000`	Таймаут HTTP-клиента, мс
`proxyHost` / `proxyPort`	—	HTTP-прокси

Приоритет резолва кредов: дефолтная цепочка → профиль → session-token → статические ключи. Validate() требует, чтобы был хотя бы один способ: accessKey+secretKey, profileName, useDefaultCredentialsProvider=true или зарегистрированный connectionFactory — иначе исключение на старте.

SQS — консьюмер (From("sqs://..."))

Параметр	Дефолт	Зачем
`waitTimeSeconds`	`20`	Long-poll (0–20). Меньше пустых receive, меньше запросов
`maxNumberOfMessages`	`10`	Сколько тянуть за один receive (1–10)
`visibilityTimeout`	`0` (дефолт очереди)	На сколько сообщение прячется на время обработки
`concurrentConsumers`	`1`	Число конкурирующих receive-циклов (см. рецепт про многопоточность)
`extendMessageVisibility`	`false`	Продлевать visibility, пока обработчик работает (heartbeat); нужен `visibilityTimeout > 0`
`deleteAfterRead`	`true`	Удалять сообщение после успешной обработки (это и есть ack)
`resetVisibilityOnFailure`	`false`	При ошибке сбросить visibility в 0 → немедленная переотдача
`transacted`	`false`	Отложить delete в транзакцию маршрута (`.Transacted()`)
`attributeNames`	`All`	Какие системные атрибуты запрашивать
`messageAttributeNames`	`All`	Какие message-атрибуты запрашивать
`delay`	`0`	Пауза (мс) после пустого receive перед следующим поллом
`initialDelay`	`0`	Задержка (мс) перед первым поллом

SQS — продюсер (.To("sqs://..."))

Параметр	Дефолт	Зачем
`delaySeconds`	`0`	Задержка доставки (0–900). FIFO-очереди per-message delay не принимают
`messageGroupId`	—	FIFO group id. Константа или `${...}`-выражение (на каждое сообщение)
`messageDeduplicationId`	—	FIFO dedup id. Опустить, если у очереди content-based dedup
`enableBatch`	`false`	Тело-`IEnumerable` отправить одним `SendMessageBatch`
`batchMaxMessages`	`10`	Размер батча (жёсткий лимит SQS — 10)
`autoCreateQueue`	`false`	Создать очередь на старте (FIFO, если имя оканчивается на `.fifo`)
`queueUrl`	(пусто)	Явный URL очереди вместо резолва по имени

SNS — publisher (.To("sns://..."))

Параметр	Дефолт	Зачем
`autoCreateTopic`	`false`	Создать топик на старте (FIFO по `.fifo`)
`topicArn`	(пусто)	Явный ARN топика вместо резолва по имени
`subject`	(пусто)	Тема (для email-доставки). Поддерживает `${...}`
`messageStructure`	(пусто)	`json` — payload по протоколам
`messageGroupId` / `messageDeduplicationId`	—	Для FIFO-топиков
`subscribeSnsToSqs` + `subscribeQueueArn`	`false`	На старте подписать SQS-очередь (по ARN) на этот топик
`rawMessageDelivery`	`false`	На той подписке — отдавать голый payload (см. рецепт про Pub-Sub)

Заголовки и трейсинг сквозь брокер

Консьюмер sqs:// кладёт в заголовки метаданные сообщения (префикс redbSqs.), а входящие message-атрибуты — под префиксом redbSqs.attr.:

Заголовок	Что значит
`redbSqs.queue`	Имя очереди
`redbSqs.messageId`	Идентификатор сообщения
`redbSqs.receiptHandle`	Receipt handle (нужен для delete / смены visibility)
`redbSqs.approximateReceiveCount`	Сколько раз сообщение уже отдавалось (полезно для «сколько раз ретраили»)
`redbSqs.messageGroupId` / `redbSqs.sequenceNumber`	FIFO group + порядковый номер
`redbSqs.sentTimestamp`	Когда отправили (epoch-millis)
`redbSqs.attr.<имя>`	Входящий пользовательский message-атрибут

Обратно — маппинг заголовок → message-атрибут на продюсере устроен так: пользовательские заголовки уезжают атрибутами как есть; входящие redbSqs.attr.<имя> пробрасываются под именем <имя> (префикс снимается — так sqs→sqs и sqs→sns мосты сохраняют атрибуты); внутренние redbSqs.* / redbSns.* — отбрасываются, чтобы метаданные одного хопа не утекли в следующий.

Трейсинг — из коробки: перед отправкой продюсер инжектит W3C traceparent/tracestate в message-атрибуты через стандартный DistributedContextPropagator, а консьюмер на другом конце их поднимает и продолжает трейс. Через прямой SQS (sqs:// → sqs://) цепочка непрерывна без единой строчки настройки. (Про нюанс с SNS-конвертом — ниже, в рецепте про Pub-Sub.)

Честно про доставку: at-least-once, visibility, transacted

Место, где маркетинг любит сказать «exactly-once». Открываем код.

SQS-консьюмер — это at-least-once. Сообщение удаляется (DeleteMessage) только после успешного прохода по маршруту. Логика settle такая:

прошло успешно + deleteAfterRead=true → DeleteMessage (ack);
упало + resetVisibilityOnFailure=true → ChangeMessageVisibility(0) → сообщение сразу возвращается в очередь (быстрый ретрай);
упало без этого флага → сообщение просто «протухает» по visibility timeout и отдаётся снова.

visibilityTimeout — на сколько сообщение спрятано на время обработки. Если обработчик долгий, есть extendMessageVisibility=true: фоновый heartbeat продлевает visibility каждые visibilityTimeout / 2, чтобы длинная работа не спровоцировала переотдачу.

transacted=true откладывает ack в транзакцию маршрута. Тогда delete не происходит сразу — регистрируется отложенное действие SqsAckAction, у которого Commit удаляет сообщение, а Rollback ставит visibility в 0 (немедленная переотдача). Коммит/откат едут вместе с redb-работой на границе .Transacted():

From("sqs://orders?transacted=true&visibilityTimeout=60")
    .Transacted()
        .ProcessWithRedb((redb, ex) => redb.SaveAsync(Parse(ex)))   // запись в БД
    .EndTransaction();     // БД закоммитилась → сообщение удалилось; упала → вернулось

Чего тут нет — кросс-очередной атомарности «прочитал из A, записал в B» на уровне брокера (у SQS её и не существует). Краш ровно между «работа сделана» и «delete» на рестарте даст повтор — at-least-once, а не exactly-once. Поэтому обработчик должен быть идемпотентным; redbSqs.approximateReceiveCount в помощь. Честная формулировка — «at-least-once + транзакционный ack на уровне роута».

Порядок сохраняется только у FIFO-очереди (.fifo) и только при concurrentConsumers=1.

Рецепт 1: продюсер, консьюмер, FIFO, батч

Минимум:

// Consumer — long-poll, delete after success
From(Sqs.Queue("orders").WaitTimeSeconds(20))
    .Process(HandleOrder);

// Producer — send the body to the queue
To(Sqs.Queue("orders").Region("eu-west-1"));

FIFO — имя оканчивается на .fifo, group id обязателен (константа или выражение на каждое сообщение):

To(Sqs.Queue("orders.fifo").MessageGroupId("${header.customerId}"));

Батч — тело-IEnumerable уходит одним SendMessageBatch (чанками по batchMaxMessages, лимит SQS — 10):

From("timer://tick?period=5000")
    .SetBody(_ => new[] { "a", "b", "c", "d", "e" })
    .To(Sqs.Queue("orders").EnableBatch());   // один вызов вместо пяти

Рецепт 2: конкурирующие консьюмеры (многопоточность)

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

From(Sqs.Queue("orders").ConcurrentConsumers(8).MaxNumberOfMessages(1))
    .Process(HandleOrder);   // до 8 сообщений обрабатываются параллельно

Это SQS-нативная модель конкуренции: коннектор поднимает N независимых receive-циклов (по задаче на цикл), каждый тянет и обрабатывает своё сообщение. maxNumberOfMessages=1 при этом заставляет каждый из N воркеров держать ровно одно сообщение — пул насыщается по-честному, а не «один цикл забрал батч из 10». Дефолт — 1 (строго серийно). Ставите N > 1 — порядок на очереди больше не гарантируется, обработчик должен быть потокобезопасным.

Насколько это реально параллелит — видно на демо SqsRpcDemo (о нём ниже): 12 запросов по ~300 мс работы каждый, пул из 4 воркеров — вся пачка отрабатывает за ~1.2 с вместо ~3.6 с серийно, а замеренный пик одновременности упирается ровно в 4.

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

Рецепт 3: RPC (request/reply) — и почему SQS его не дарит

Вот тут — честный контраст с прошлой статьёй. У RabbitMQ RPC включался одним флагом replyTo=true на клиенте и нулём настроек на сервере: брокер сам умеет reply-очереди. У SQS такого нет — нет ни reply-to на уровне протокола, ни ожидания ответа. RPC поверх SQS строится руками, по классическому паттерну корреляции: временная/выделенная reply-очередь, correlationId, ожидание коррелированного ответа. Это ровно то, что показывает демо SqsRpcDemo.

Клиент к каждому запросу цепляет два message-атрибута — correlationId и replyTo (имя очереди для ответа) — и шлёт в очередь запросов. Воркер считает результат и отправляет его в очередь из replyTo, а correlationId возвращается сам (входящий redbSqs.attr.correlationId продюсер пробрасывает обратно атрибутом). Воркер — обычный консьюмер + динамический .ToD(...):

// Worker: N competing consumers on the request queue, reply to the queue named in replyTo
From(Sqs.Queue("rpc-requests").ConcurrentConsumers(4).MaxNumberOfMessages(1))
    .Process(async (ex, ct) =>
    {
        var n = int.Parse(ex.In.Body!.ToString()!);
        await Task.Delay(300, ct);              // simulated work
        ex.In.Body = (n * (long)n).ToString();  // the reply body
    })
    // dynamic destination — the reply queue comes from the request's replyTo attribute
    .ToD(ex => Sqs.Queue(ReplyTo(ex)).Build());

На стороне клиента отдельный маршрут читает reply-очередь и сопоставляет correlationId ожидающему вызову:

From(Sqs.Queue("rpc-replies").MaxNumberOfMessages(10))
    .Process((ex, ct) =>
    {
        var id = Attr(ex, "correlationId");
        if (id is not null && Pending.TryRemove(id, out var tcs))
            tcs.TrySetResult(ex.In.Body?.ToString() ?? "");
        return Task.CompletedTask;
    });

Мораль: RPC на SQS — рабочий, но это сборка из примитивов (атрибуты + .ToD + корреляция), а не флаг. Где нужен «RPC из коробки» — берут брокер с нативным reply (RabbitMQ). SQS честно про своё: он про надёжные очереди и переотдачу, а request/reply вы достраиваете сами. Полный воспроизводимый пример — в redb.Route/demos/SqsRpcDemo (гоняется против LocalStack).

Рецепт 4 (EIP): Publish-Subscribe через SNS→SQS fan-out

Вот главный паттерн, ради которого SQS ходит в паре с SNS. Publish-Subscribe Channel из книги Хопе и Вульфа: издатель публикует одно сообщение, а получает его каждый подписчик — независимо, в свою очередь, со своими ретраями. У AWS это делается не в коде, а топологией: SNS-топик, на который подписаны несколько SQS-очередей.

publish "order" ──▶ SNS topic ──┬──▶ SQS "orders-billing"  ──▶ billing route
                                 └──▶ SQS "orders-shipping" ──▶ shipping route

Издатель просто публикует в топик, а получают все подписанные очереди — каждая своим независимым консьюмером:

// Publisher — publish an order event; SNS fans it out to every subscribed queue:
From("timer://orders?period=5000")
    .SetBody(_ => BuildOrder())
    .To(Sns.Topic("orders-events"));

// Two independent subscribers — each its own SQS queue, each its own consumer:
From(Sqs.Queue("orders-billing")).Process(Charge);
From(Sqs.Queue("orders-shipping")).Process(Ship);

Сами подписки коннектор умеет завести из URI — subscribeSnsToSqs + ARN очереди; подписка выполняется на старте sns://-продюсера. Каждый sns://-endpoint подписывает одну очередь (ARN один), поэтому fan-out на две — это две подписки на один топик:

// Subscribe each queue to the topic with raw delivery (bare payload, no envelope):
Sns.Topic("orders-events").SubscribeSnsToSqs(billingArn).RawMessageDelivery();
Sns.Topic("orders-events").SubscribeSnsToSqs(shippingArn).RawMessageDelivery();

Точную обвязку fan-out на две очереди (создание очередей, чтение ARN, подписка, публикация) — см. redb.Route/demos/SqsPubSubDemo.

Контраст с прямой очередью: у sqs:// одно сообщение достаётся одному потребителю (competing consumers, деление работы). Здесь — широковещание: одно событие → у каждой подписанной очереди своя копия. Добавить нового подписчика (аналитика, аудит) — это ещё одна очередь на тот же топик; издателя трогать не нужно. Полностью развязанные стороны.

⚠️ Важный нюанс: конверт vs raw

По умолчанию SNS не кладёт в очередь ваш payload как есть — он оборачивает его в JSON-конверт уведомления:

{ "Type": "Notification", "MessageId": "...", "TopicArn": "...",
  "Message": "<ваш payload здесь, строкой>", "MessageAttributes": { ... } }

То есть подписчик-очередь получает конверт, а не ваше сообщение, и ваши SNS-атрибуты лежат внутри него, а не как SQS-атрибуты. Это стандартное поведение AWS SNS (флаг подписки Raw Message Delivery = OFF), полезное ровно тогда, когда одна очередь слушает несколько топиков и по TopicArn из конверта надо понять источник. Для обычного fan-out это только мешает — приходится разворачивать конверт, а сквозной трейс SNS→SQS рвётся (traceparent уезжает внутрь конверта, а SQS-консьюмер ищет его в SQS-атрибутах).

Лечится флагом подписки Raw Message Delivery = true: очередь получает голый payload, а SNS-атрибуты становятся SQS-атрибутами (и трейс снова сквозной). В коннекторе это .RawMessageDelivery() — он ставит RawMessageDelivery=true на той самой авто-подписке subscribeSnsToSqs. Демо SqsPubSubDemo печатает первое доставленное тело: с .RawMessageDelivery() это {"orderId":1,"amount":100}, а не конверт.

Про rawMessageDelivery. Опция выйдет в следующем релизе NuGet пакета redb.Route.Sqs. Прямо сейчас её уже можно взять из исходников — коннектор redb.Route.Sqs на GitHub (папка коннектора, SnsEndpointOptions / SnsProducer / флент RawMessageDelivery). Без него SNS→SQS тоже работает, просто в тело приходит JSON-конверт, который надо развернуть (envelope.Message).

Рецепт 5 (родня): Multicast — тот же веер, но внутри маршрута

Publish-Subscribe размножает сообщение на брокере (SNS отдаёт копию каждой подписанной очереди). Есть родственный EIP, который размножает внутри маршрута, — Multicast: ваш шаг отправляет копию exchange в несколько получателей сам.

From("sqs://orders")
    .Multicast()
        .To("sqs://orders-billing")
        .To("sqs://orders-shipping")
    .End();

Выглядит похоже, но разница принципиальная — где происходит размножение и кто про кого знает:

	SNS→SQS (Pub-Sub)	Multicast
Где копируется	на брокере (SNS)	в маршруте (ваш процесс)
Кто знает список получателей	SNS (подписки)	маршрут (список `.To`)
Добавить получателя	подписать ещё очередь, издателя не трогать	править код маршрута
Развязка сторон	полная	получатели зашиты в маршрут
Параллельность / агрегация ответов	нет (fire-and-forget по подпискам)	есть: `.Parallel().MaxParallelism(N)`, склейка веток

Правило простое: нужно развязать издателя и подписчиков (кто угодно подпишется потом, издатель не в курсе) — SNS→SQS. Нужно тут же разослать по фиксированному списку и, возможно, собрать ответы — Multicast (или его брат Scatter-Gather из статьи про Kafka). Один — топология брокера, другой — шаг маршрута; выбираются по тому, где должна жить логика веера.

Итог

SQS/SNS в redb.Route — это маршруты, где вся очередь и весь топик заданы одним URI: From("sqs://…") / .To("sqs://…") / .To("sns://…"), а вся возня с AWS SDK, receive/delete, visibility, подписками и корреляцией спрятана за параметрами строки. Long-poll, FIFO, батч, конкурирующие консьюмеры одним параметром, at-least-once с транзакционным ack, SNS→SQS fan-out с выбором «конверт или голый payload» — набор полный, а на выходе короткая и понятная строка.

Честно про размен: exactly-once тут нет (его нет и у самого SQS) — есть at-least-once + идемпотентность; RPC не «из коробки», а собирается из атрибутов и .ToD; SNS-подписка по умолчанию заворачивает payload в конверт. Всё это — свойства AWS, а не недоделки коннектора, и коннектор про них честен.

Два воспроизводимых примера — redb.Route/demos/SqsRpcDemo (RPC + многопоточность) и redb.Route/demos/SqsPubSubDemo (SNS→SQS fan-out + raw delivery); оба поднимаются на LocalStack (http://localhost:4566, test/test) и гоняются как self-test. Ловится что-то в вашем сценарии — пишите в комментариях.

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

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