Модуль 4. NATS + JetStream — скорость и элегантность

Цель модуля: освоить двухслойную систему: NATS Core как быструю push-шину (pub/sub + нативный request-reply) и JetStream как персистентный слой (стримы, консьюмеры, KV, Object Store). Дойти до exactly-once через dedup window, понять кластеризацию и leaf nodes — и собрать микросервис «Realtime-уведомления + персистентная очередь».

Перед началом: Модули 1–3 этого курса; таблица «почему семантики разные» — в курсе RabbitMQ (раздел 1.1). Инструменты: docker-compose (nats:2.10-alpine, флаг -js), github.com/nats-io/nats.go. Трудоёмкость: ~10–13 ч (теория 4 ч · практика 7 ч · квиз 1 ч).

📖 4.1. NATS Core: pub/sub, request-reply, wildcards

Модель без очередей

NATS Core — это subject-текстовая шина с push-семантикой: сервер не хранит сообщения. Публикация в subject доставляется всем активным подписчикам; нет подписчиков — сообщение исчезает. Это фича, а не баг: за счёт отказа от персистентности NATS Core держит миллионы сообщений в секунду на одном ядре.

text
[P] --publish orders.created--> NATS
                                  ├──► [sub orders.>]  получает
                                  └──► [sub payments.*] не получает

Правила wildcard'ов: * — ровно один токен, > — один и более токенов (только в конце). orders.*.eu ловит orders.created.eu, но не orders.eu.created; orders.> ловит orders.eu.created.

go
package main

import (
	"log"

	"github.com/nats-io/nats.go"
)

func main() {
	nc, err := nats.Connect(nats.DefaultURL,
		nats.RetryOnFailedConnect(true),
		nats.MaxReconnects(-1),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer nc.Drain() // Drain аккуратно: отписка + досылка буфера (см. shutdown)

	if _, err := nc.Subscribe("orders.>", func(m *nats.Msg) {
		log.Printf("got %s: %s", m.Subject, m.Data)
	}); err != nil {
		log.Fatal(err)
	}

	if err := nc.Publish("orders.created.eu", []byte(`{"id":"o-1"}`)); err != nil {
		log.Fatal(err)
	}
}

Request-reply — нативный, не надстройка

В отличие от Kafka, request-reply в NATS — первоклассная операция: Request создаёт inbox-subject, публикует запрос с reply-to и ждёт ответ с таймаутом. Несколько респондеров на один subject — бесплатный load-balancing (плюс queue groups).

go
// сервер
nc.Subscribe("svc.echo", func(m *nats.Msg) {
	_ = m.Respond(m.Data) // эхо
})

// клиент
resp, err := nc.Request("svc.echo", []byte("ping"), 2*time.Second)
if errors.Is(err, nats.ErrTimeout) {
	// нет ответа за таймаут — респондер мёртв или перегружен
}

Если интереса на subject нет вообще, Request вернёт ErrNoResponders мгновенно — брокер знает карту подписок. Сравните с rabbitmq-RPC из 2.1, где «нет ответа» неотличимо от «медленный респондер»: в NATS это одна семантическая дыра меньше.

Queue groups: nc.QueueSubscribe("jobs", "workers", handler) — сообщение достанется одному из группы. Это work-queue на Core без персистентности: подписчики оффлайн — задачи пропали. Нужна надёжность — JetStream (4.2).

📖 4.2. JetStream: стримы, консьюмеры, политики — подробно

Два уровня

JetStream добавляет поверх Core персистентность: Stream хранит сообщения (по subject'ам), Consumer — именованная позиция чтения с политиками ack. Модель ближе к «Kafka с флексибельными консьюмерами», чем к очередям RabbitMQ.

go
js, err := nc.JetStream()

_, err = js.AddStream(&nats.StreamConfig{
	Name:      "ORDERS",
	Subjects:  []string{"orders.>"},
	Storage:   nats.FileStorage,     // FileStorage vs MemoryStorage
	Retention: nats.LimitsPolicy,    // см. таблицу ниже
	MaxAge:    24 * time.Hour,
	Replicas:  3,                    // Raft-репликация, как quorum queues в 2.5
})

ack, err := js.Publish("orders.created.eu", body)
// ack.Sequence — подтверждение записи стримом (аналог publisher confirms)

Retention-политики: когда сообщения покидают стрим

WorkQueuePolicy требует: один консьюмер на subject (или разделение subject'ов) — это именно очередь, а не подписка.

Консьюмер: пять ручек, которые надо понимать

go
sub, err := js.PullSubscribe("orders.created.eu", "payment",
	nats.BindStream("ORDERS"),
	nats.DeliverAll(),            // DeliverPolicy
	nats.AckExplicit(),           // AckPolicy
	nats.MaxDeliver(5),           // до 5 попыток доставки
	nats.MaxAckPending(1000),     // flow control
)
msgs, err := sub.Fetch(10, nats.MaxWait(5*time.Second))
for _, m := range msgs {
	if err := handle(m.Data); err != nil {
		m.Nak() // отрицательный ack: доставить снова (с учётом MaxDeliver)
		continue
	}
	m.Ack()
}

Ответы на вопросы, которые будут на практике:

• Упал до ack при AckExplicit + DeliverAll → после AckWait сообщение доставится снова (at-least-once). Никакой магии, как и в RabbitMQ/Kafka.
• MaxDeliver vs TTL+DLX RabbitMQ: семантика другая. В RabbitMQ «неудобное» сообщение уезжает в явное место (DLQ), которое вы проектируете. В JetStream по умолчанию после исчерпания MaxDeliver сообщение просто перестаёт доставляться — оно остаётся в стриме, и узнавать о нём нужно через system-адвайзори ($JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES) или мониторинг num_redelivered. DLX-паттерн в JetStream строится руками: отдельный consumer, слушающий advisory и публикующий «в больницу».
• AckPolicy.All — ack сообщения подтверждает и все предыдущие доставленные: удобно для потоковой обработки, где вы всё равно идёте последовательно.
• ReplayPolicy.Original — воспроизводит исходные интервалы между сообщениями: для replay-сценариев («посмотри, что происходило в проде в час пик»), тогда как Instant — по максимуму скорости.

Push vs Pull

• Pull (как выше) — консьюмер сам запрашивает пачки. Ваш выбор для worker-пулов: N горутин/подов сами тянут столько, сколько могут (нативный flow control через Fetch + MaxAckPending).
• Push — сервер доставляет сам (js.Subscribe): ниже латентность, удобно для broadcast-подписок, но контроль потока хуже; для медленных обработчиков берите pull.

📖 4.3. KV и Object Store: конфиг-сервис в 30 строк

JetStream-KV — это компактед-стрим по ключам (аналог compacted topic из 3.5, но с клиентским API). Object Store — чанкованные большие объекты.

go
kv, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "config"})
if err != nil {
	log.Fatal(err)
}

// пишем конфиг
if _, err := kv.Put("payments.timeout", []byte("30s")); err != nil {
	log.Fatal(err)
}

// читаем с ревизией
entry, err := kv.Get("payments.timeout")
log.Printf("timeout=%s rev=%d", entry.Value(), entry.Revision())

// история значений ключа (как в git)
hist, _ := kv.History("payments.timeout")

// watch — живые обновления конфига без рестарта
w, _ := kv.Watch("config.>")
for u := range w.Updates() {
	if u == nil { // nil = начальный снапшот выдан полностью
		continue
	}
	log.Printf("config changed: %s = %s (rev %d)", u.Key(), u.Value(), u.Revision())
}

Thirty-строк достаточно для конфиг-сервиса с историей, watch и оптимистичной блокировкой (kv.Update(key, value, lastRevision) — CAS). Когда уместно: service discovery, feature flags, состояние лидера. Когда нет: secrets (безопасность — в курсе OrderFlow) и объёмы, где нужен настоящий сторедж.

📖 4.4. Exactly-once в JetStream: deduplication window

JetStream даёт серверную дедупликацию публикации: если сообщение несёт заголовок Nats-Msg-Id и попадает в стрим, пока действует DuplicateWindow, повтор отбрасывается, а продюсеру возвращается ack оригинала.

go
_, err = js.AddStream(&nats.StreamConfig{
	Name:           "ORDERS",
	Subjects:       []string{"orders.>"},
	DuplicateWindow: 2 * time.Minute,
})

_, err = js.PublishMsg(&nats.Msg{
	Subject: "orders.created.eu",
	Headers: nats.Header{"Nats-Msg-Id": []byte(orderID + ":created")},
	Data:    body,
})

Сравнение с Kafka (3.4):

Практический вывод: Nats-Msg-Id = idempotency-key из Модуля 1, поднятый на уровень брокера. Клиент может ретраить публикацию сколько угодно — стрим не задвоит. Но после истечения окна дубликаты пройдут: окно должно покрывать худший сценарий ретраев, а для «вечной» дедупликации по-прежнему нужна идемпотентность приложения.

📖 4.5. Кластеризация, суперкластер, leaf nodes — и микросервисная архитектура

Кластер и над ним

• Cluster: N серверов JetStream с Raft-репликацией стримов (Replicas: 3 — переживает 1 отказ, как quorum-очередь в 2.5).
• Supercluster: несколько кластеров, соединённых gateway —subjects видны глобально, NATS маршрутизирует между датацентрами.
• Leaf nodes: лёгкие узлы на краю (филиал, IoT, edge): локальные приложения общаются с leaf, leaf мостится в суперкластер. При обрыве связи leaf живёт автономно, при восстановлении — дозирует трафик. Это «geo-распределение без боли» — то, чего в чистом виде нет ни у RabbitMQ, ни у Kafka.

Микросервисный пример: архитектура на NATS

text
NATS cluster (JetStream)
   ┌──────────────────────────┬───────────────────────────┐
   │                          │                           │
orders-api                 payment-svc                notify-svc
   │ publish orders.>        │ pull consumer             │ push consumer
   │ (Nats-Msg-Id)           │ ORDERS/payments           │ ORDERS/notify (push, AckExplicit)
   │                         │                           │
   └── request svc.users.<id> ──────────► user-svc (req/reply, кэш в KV)
                                             │
                                    KV bucket "users" (кэш-проекция)

Ключевые решения и почему они идиоматичны именно для NATS:

1. API → стрим через JetStream с Nats-Msg-Id — публикация надёжна и дедуплицируется (4.4).
2. payment-svc — pull consumer с MaxAckPending — честный backpressure для воркеров.
3. notify-svc — push consumer — минимальная латентность для пуша в WebSocket/SSE.
4. user-svc — request-reply для синхронных допрашиваний + KV как проекция-кэш: svc.users.<id> сначала смотрят в KV, промах — запрос.
5. Leaf node в регионе: notify-svc живёт на leaf, события доходят через мост, локальные подписчики не ходят в центральный ДЦ.

NATS vs RabbitMQ: производительность (честная таблица)

Типовые ориентиры для сообщений ~100–1000 байт на одном сервере (современное железо; ваши цифры обязан мерить капстоун (курс OrderFlow)):

Оговорки, без которых таблица — враньё: (1) NATS Core быстр, потому что ничего не хранит — сравнивать с персистентным RabbitMQ напрямую некорректно; (2) JetStream с FileStorage близок к RabbitMQ по надёжности и заметно быстрее на fanout; (3) RabbitMQ выигрывает на сложной маршрутизации (exchange-модель) и богатых подтверждениях per-message; (4) при спорных цифрах — только свой бенчмарк на своей нагрузке (9.5).

📖 4.6. Лабораторная: «Realtime-уведомления + персистентная очередь»

Микросервис из двух половин: персистентная обработка заказов (JetStream) и realtime-уведомления (NATS Core), плюс конфиг из KV.

go
package main

import (
	"context"
	"log"
	"time"

	"github.com/nats-io/nats.go"
)

func main() {
	nc, _ := nats.Connect(nats.DefaultURL)
	defer nc.Drain()
	js, _ := nc.JetStream()

	// 1. Персистентная очередь заказов
	_, err := js.AddStream(&nats.StreamConfig{
		Name:            "ORDERS",
		Subjects:        []string{"orders.>"},
		Storage:         nats.FileStorage,
		Retention:       nats.WorkQueuePolicy,
		DuplicateWindow: 5 * time.Minute,
	})
	if err != nil && err != nats.ErrStreamNameAlreadyInUse {
		log.Fatal(err)
	}

	// 2. Реалтайм-бродкаст статуса (Core, без хранения)
	statusSub, _ := nc.Subscribe("orders.status.>", func(m *nats.Msg) {
		log.Printf("UI push: %s %s", m.Subject, m.Data) // здесь был бы WebSocket
	})

	// 3. Pull-консьюмер для обработки
	sub, _ := js.PullSubscribe("orders.created", "processor",
		nats.BindStream("ORDERS"),
		nats.AckExplicit(), nats.MaxDeliver(3), nats.MaxAckPending(100),
	)

	ctx := context.Background()
	for {
		msgs, err := sub.Fetch(10, nats.MaxWait(2*time.Second))
		if err != nil {
			continue // таймаут фетча — норма, идём дальше
		}
		for _, m := range msgs {
			_ = statusSub
			if err := process(ctx, m); err != nil {
				m.Nak()
				continue
			}
			m.Ack()
			_ = nc.Publish("orders.status.done", m.Data) // всем интересующимся
		}
	}
}

Задания к лабораторной:

1. Убедитесь тестом: при WorkQueuePolicy сообщение удаляется из стрима после ack; при LimitsPolicy с тем же кодом — остаётся (и новый консьюмер с DeliverAll получит его снова).
2. Уроните процесс после process, но до m.Ack() — сообщение должно прийти снова (зафиксируйте, что AckWait сработал).
3. Задублируйте публикацию с тем же Nats-Msg-Id в окне дедупликации — стрим должен принять одно.
4. Соберите мини-конфиг в KV и меняйте его на лету — воркер подхватывает без рестарта.

💻 Практика

1. lesson4_1: реализуйте rate-limited RPC поверх nc.Request и покажите ErrNoResponders при мёртвом респондере.
2. lesson4_2: матрица тестов DeliverPolicy × AckPolicy: для каждой пары — что видит новый консьюмер и что происходит при краше.
3. lesson4_3: feature-flag сервис на KV с CAS-обновлением.
4. Промежуточный проект: лабораторная 4.6 + advisory на исчерпание MaxDeliver (подпишитесь на $JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES и сложите «пойманных» в DLQ-стрим — ваш JetStream-аналог dead letter).

🧩 Quiz "Проверь себя"

✅ Итоги модуля

После прохождения ты должен уметь:

• Объяснить двухслойность NATS: Core (быстро, без хранения) vs JetStream (персистентность, ack) — и когда какой слой уместен.
• Настроить консьюмера JetStream из пяти ручек (DeliverPolicy, AckPolicy, AckWait, MaxDeliver, MaxAckPending) и предсказать поведение при сбое для любой комбинации.
• Объяснить разницу push/pull консьюмеров и выбрать под задачу.
• Собрать серверную дедупликацию через Nats-Msg-Id и указать её отличие от Kafka-идемпотентности.
• Отличить Limits/Interest/WorkQueue retention и последствия каждой.
• Спроектировать микросервисную топологию на NATS: стримы, pull/push консьюмеры, request-reply, KV-проекция, leaf nodes.
• Честно сравнить NATS и RabbitMQ по производительности — с оговорками.

Шпаргалка модуля

🔗 Мостик: дальше в треке

Ядро собрано: язык (гарантии, идемпотентность), паттерны и ландшафт из пяти брокеров — RabbitMQ, Kafka, NATS, Redpanda, облачные очереди. Дальше — курсы по отдельным брокерам: Kafka (партиции, транзакции, Schema Registry), RabbitMQ (AMQP, quorum queues, DLX) и Redis как брокер (Pub/Sub и Streams). А затем — капстоун трека: OrderFlow, где все они соберутся в один пайплайн с трейсингом и нагрузочными тестами.