Модуль 13 ⏱ ~27 мин чтения 📐 2 Mermaid диаграмм

7. Когда выбирать NATS

Решающие деревья и архитектурные компромиссы

«Простота — непременное условие надежности. Прежде чем браться за инструмент, требующий целой команды для обслуживания, спросите себя: не принесет ли больше пользы более простое решение с четкими границами?».
— Брайан Керниган

Стратегический выбор архитектуры

Выбор брокера сообщений — это не просто техническое решение или вопрос личных предпочтений команды, а стратегический выбор архитектуры, который определяет масштабируемость, сетевую задержку, надежность и совокупную стоимость владения (TCO — Total Cost of Ownership) всей системы на годы вперед.

NATS занимает уникальную нишу в современной инженерии: это одновременно сверхлегкая «нервная система» микросервисов и мощная персистентная платформа для потоков событий. Понимание его сильных сторон и ограничений критично для принятия взвешенного решения.

Ключевые сценарии, где NATS сияет

1. Микросервисная коммуникация с низкой задержкой

Если ваша архитектура состоит из десятков или сотен микросервисов, которые непрерывно взаимодействуют между собой, NATS предоставляет идеальный транспортный слой, заменяющий зоопарк из REST, gRPC, Consul и Envoy:

Go 
// Service-to-service communication через NATS
type OrderService struct {
    nc *nats.Conn
}

func (s *OrderService) CreateOrder(ctx context.Context, order Order) error {
    // 1. Асинхронно публикуем событие создания заказа в шину (Fire-and-forget)
    data, _ := json.Marshal(order)
    if err := s.nc.Publish("orders.created", data); err != nil {
        return fmt.Errorf("publish failed: %w", err)
    }

    // 2. Синхронный вызов для проверки наличия товара (request-reply RPC)
    resp, err := s.nc.Request("inventory.check", []byte(order.SKU), 500*time.Millisecond)
    if err != nil {
        return fmt.Errorf("inventory check failed: %w", err)
    }

    if string(resp.Data) != "available" {
        return ErrItemNotAvailable
    }

    return nil
}

Почему NATS:

  • Задержка <1 мс для внутренних межсервисных вызовов;
  • Встроенный request-reply поверх асинхронного транспорта без внешнего Service Mesh;
  • Простая маршрутизация через иерархические subject'ы вместо сложного конфигурирования exchanges и bindings.
graph LR
    OS["Order Service"] -->|"orders.created"| NS["NATS Server"]
    IS["Inventory Service"] -->|"inventory.check"| NS
    NS -->|"orders.created"| ES["Email Service"]
    NS -->|"orders.created"| AS["Analytics Service"]
    NS -->|"inventory.check<br>reply"| IS

    style NS fill:#1e293b,stroke:#6366f1,stroke-width:2px,color:#f8fafc
    style OS fill:#131d33,stroke:#38bdf8,stroke-width:2px,color:#f8fafc
    style IS fill:#131d33,stroke:#38bdf8,stroke-width:2px,color:#f8fafc
    style ES fill:#131d33,stroke:#10b981,stroke-width:2px,color:#f8fafc
    style AS fill:#131d33,stroke:#10b981,stroke-width:2px,color:#f8fafc

2. Real-time приложения и live-обновления

Чат-приложения, котировки акций, спортивная телеметрия, онлайн-игры, дашборды — везде, где во главу угла ставится мгновенная доставка событий с минимальным джиттером (jitter):

Go 
// Live price updates для трейдинговой платформы
func (p *PricePublisher) PublishUpdate(symbol string, price PriceUpdate) {
    subject := fmt.Sprintf("market.price.%s", symbol)
    data, _ := json.Marshal(price)

    // Fire-and-forget: скорость важнее гарантии доставки
    _ = p.nc.Publish(subject, data)
}

// Подписка на обновления в реальном времени
func (c *PriceConsumer) Subscribe(symbol string, handler func(PriceUpdate)) error {
    subject := fmt.Sprintf("market.price.%s", symbol)
    _, err := c.nc.Subscribe(subject, func(msg *nats.Msg) {
        var update PriceUpdate
        if err := json.Unmarshal(msg.Data, &update); err != nil {
            log.Printf("Unmarshal error: %v", err)
            return
        }
        handler(update)
    })
    return err
}

3. Service discovery и health monitoring

Встроенная динамическая маршрутизация позволяет реализовать самоорганизующийся кластер сервисов без внешних реестров (Consul, Zookeeper, etcd):

Go 
// Регистрация сервиса при старте
func RegisterService(nc *nats.Conn, serviceName, instanceID, address string) error {
    heartbeatSubject := fmt.Sprintf("service.%s.heartbeat", serviceName)

    // Периодическая отправка heartbeat
    go func() {
        ticker := time.NewTicker(30 * time.Second)
        defer ticker.Stop()

        for range ticker.C {
            payload := fmt.Sprintf(`{"instance":"%s","addr":"%s","ts":%d}`, 
                instanceID, address, time.Now().Unix())
            _ = nc.Publish(heartbeatSubject, []byte(payload))
        }
    }()

    return nil
}

// Обнаружение активных инстансов сервиса
func DiscoverInstances(nc *nats.Conn, serviceName string) ([]Instance, error) {
    inbox := nats.NewInbox()
    ch := make(chan *nats.Msg, 10)

    sub, err := nc.ChanSubscribe(inbox, ch)
    if err != nil {
        return nil, err
    }
    defer sub.Unsubscribe()

    // Запрос ко всем инстансам
    if err := nc.PublishRequest(
        fmt.Sprintf("service.%s.discover", serviceName),
        inbox,
        []byte("{}"),
    ); err != nil {
        return nil, err
    }

    var instances []Instance
    timeout := time.After(2 * time.Second)

    for {
        select {
        case msg := <-ch:
            var inst Instance
            if err := json.Unmarshal(msg.Data, &inst); err == nil {
                instances = append(instances, inst)
            }
        case <-timeout:
            return instances, nil
        }
    }
}

4. Edge computing и IoT

Встраиваемые устройства, Raspberry Pi, банкоматы, контроллеры умных фабрик с жесткими ограничениями по оперативной памяти (64–128 МБ) и нестабильным мобильным интернетом:

Go 
// IoT device с NATS client
type SensorDevice struct {
    nc       *nats.Conn
    sensorID string
}

func (d *SensorDevice) StartReporting(interval time.Duration) {
    ticker := time.NewTicker(interval)

    go func() {
        for range ticker.C {
            reading := d.readSensor()
            subject := fmt.Sprintf("iot.%s.telemetry", d.sensorID)

            // Минимальный payload, no ack needed
            _ = d.nc.Publish(subject, reading.ToBytes())
        }
    }()
}

Почему NATS для IoT:

  • Клиент весит ~1 МБ, не тянет тяжелые зависимости;
  • Работает поверх простого TCP, без сложных протоколов;
  • Встроенная поддержка reconnect и буферизации (buffer flush) при восстановлении соединения после сбоя в сети.

Когда NATS — не лучший выбор

Архитектурная зрелость заключается в умении видеть границы применимости инструмента:

1. Критически важные финансовые транзакции без JetStream

Go 
// НЕЛЬЗЯ использовать Core NATS для этого:
nc.Publish("bank.transfer", []byte(`{"from":"A","to":"B","amount":1000}`))
// Если сообщение потеряется — деньги "исчезнут"

Решение: Обязательно использовать JetStream с персистентным хранилищем и подтверждением доставки:

Go 
ack, err := js.Publish("bank.transfer", data, nats.ExpectLastSequence(seq))
if err != nil {
    // Retry или компенсационная логика
}
// Гарантия: сообщение сохранено в stream

2. Долгосрочное хранение и аналитика больших данных

Если ваша задача — хранить десятки терабайтов событий месяцами или годами для OLAP-аналитики, обучения ML-моделей или построения Data Lake, NATS не заменит специализированные решения:

Требование NATS JetStream Kafka
Хранение 10 ТБ данных ⚠️ Возможно, но не оптимизировано ✅ Отлично, сегментированные логи
Сложные запросы к истории ❌ Нет встроенного query engine ✅ KSQL, Streams API, Spark/Flink коннекторы
Ретеншн по времени/размеру ✅ Есть ✅ Более гибкие политики и tiered storage

3. Сложная маршрутизация с приоритетами и трансформацией

Если бизнес-логика требует динамической маршрутизации на основе заголовков, приоритизации очередей (Priority Queues) и сложных цепочек Dead-Letter с кастомным TTL:

Go 
// RabbitMQ лучше подходит для:
channel.QueueDeclare("priority_queue", true, false, false, false, 
    amqp.Table{"x-max-priority": 10})

channel.Publish("exchange", "routing.key", true, false, 
    amqp.Publishing{
        Priority: 5,
        Expiration: "30000", // 30 секунд TTL
        Headers: amqp.Table{"retry-count": 3},
    })

Decision Framework: чек-лист выбора

flowchart TD
    Start["Нужен брокер сообщений?"] --> Q1["Критична ли задержка <1мс?"]

    Q1 -->|"Да"| Q2["Нужна ли персистентность?"]
    Q1 -->|"Нет"| Q3["Нужна ли сложная маршрутизация?"]

    Q2 -->|"Нет"| NATS_Core["✅ NATS Core"]
    Q2 -->|"Да"| NATS_JS["✅ NATS JetStream"]

    Q3 -->|"Да"| Q4["Нужны приоритеты/TTL/DLQ?"]
    Q3 -->|"Нет"| Q5["Объём данных >1ТБ?"]

    Q4 -->|"Да"| RabbitMQ["✅ RabbitMQ"]
    Q4 -->|"Нет"| NATS_JS

    Q5 -->|"Да"| Kafka["✅ Kafka"]
    Q5 -->|"Нет"| NATS_JS

    style Start fill:#1e293b,stroke:#6366f1,stroke-width:2px,color:#f8fafc
    style NATS_Core fill:#064e3b,stroke:#10b981,stroke-width:2px,color:#f8fafc
    style NATS_JS fill:#064e3b,stroke:#10b981,stroke-width:2px,color:#f8fafc
    style RabbitMQ fill:#7c2d12,stroke:#f97316,stroke-width:2px,color:#f8fafc
    style Kafka fill:#1e1b4b,stroke:#818cf8,stroke-width:2px,color:#f8fafc

Архитектурные паттерны с NATS

1. Event Bus для микросервисов

Единая внутренняя шина доменных событий:

Go 
// Centralized event bus через NATS
type EventBus struct {
    nc *nats.Conn
}

func (eb *EventBus) Publish(event DomainEvent) error {
    subject := fmt.Sprintf("events.%s.%s", 
        event.AggregateType(), 
        event.EventType())

    data, err := json.Marshal(event)
    if err != nil {
        return err
    }

    return eb.nc.Publish(subject, data)
}

func (eb *EventBus) Subscribe(aggregateType string, handler EventHandler) error {
    subject := fmt.Sprintf("events.%s.>", aggregateType)

    _, err := eb.nc.Subscribe(subject, func(msg *nats.Msg) {
        var event GenericEvent
        if err := json.Unmarshal(msg.Data, &event); err != nil {
            log.Printf("Unmarshal error: %v", err)
            return
        }
        handler.Handle(event)
    })
    return err
}

2. CQRS с разделением команд и событий

Go 
// Command side — синхронный request-reply
cmdResp, err := nc.Request("commands.create.order", cmdData, 2*time.Second)

// Event side — асинхронный pub/sub
nc.Subscribe("events.order.created", func(msg *nats.Msg) {
    updateReadModel(msg.Data) // Обновление query-модели
})

3. Saga-оркестрация через события

Go 
// Каждый шаг Saga публикует событие для следующего шага
func (s *OrderSaga) ReserveInventory(order Order) error {
    // Пытаемся зарезервировать товар
    resp, err := s.nc.Request("inventory.reserve", order.ToBytes(), 5*time.Second)
    if err != nil {
        // Публикуем событие неудачи для компенсации
        s.nc.Publish("saga.order.failed", 
            []byte(fmt.Sprintf(`{"orderId":"%s","step":"inventory"}`, order.ID)))
        return err
    }

    // Успех — публикуем событие для следующего шага
    s.nc.Publish("saga.order.inventory.reserved", order.ToBytes())
    return nil
}

Performance Benchmarks: когда цифры имеют значение

Go 
func benchmarkBrokers() {
    // NATS Core: ~1.2M msg/sec, p99 latency 0.3ms
    // NATS JetStream: ~400K msg/sec, p99 latency 2ms
    // RabbitMQ: ~80K msg/sec, p99 latency 15ms  
    // Kafka: ~900K msg/sec, p99 latency 8ms

    // Тест для NATS
    start := time.Now()
    for i := 0; i < 100000; i++ {
        nc.Publish("bench.test", []byte("payload"))
    }
    duration := time.Since(start)

    fmt.Printf("NATS: %d msg/sec, avg latency: %v\n", 
        100000/duration*time.Second, 
        duration/100000)
}

Экосистема и интеграции

NATS выигрывает за счёт единой экосистемы, заменяющей сразу несколько внешних СУБД:

Go 
// Один клиент для всего: pub/sub, request-reply, JetStream, KV, Object Store
nc, _ := nats.Connect("nats://localhost:4222")

// Core NATS
nc.Publish("event", data)

// Request-Reply  
nc.Request("api.call", data, timeout)

// JetStream
js, _ := nc.JetStream()
js.Publish("stream.subject", data)

// KV Store (NATS 2.6+) — встроенный Key-Value сторедж поверх JetStream
kv, _ := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "config"})
kv.Put("feature.flag", []byte("enabled"))

// Object Store (NATS 2.7+) — хранилище файлов и артефактов без S3
obs, _ := js.CreateObjectStore(&nats.ObjectStoreConfig{Bucket: "files"})
obs.Put("report.pdf", fileReader)

Преимущество: Не нужно подключать и поддерживать 5 разных библиотек и протоколов — один официальный клиент nats.go покрывает все потребности микросервиса.

Итог: матрица выбора

Сценарий Рекомендация Обоснование
Микросервисы, низкая задержка ✅ NATS Core <1ms latency, simple routing, 0 overhead
Event sourcing с гарантиями ✅ NATS JetStream Persistence + replay + ordering + Raft
Real-time notifications ✅ NATS Core Fire-and-forget, high throughput
Financial transactions ✅ JetStream или Kafka Exactly-once, audit trail, persistent log
Big data analytics ✅ Kafka Scale, KSQL, long-term retention, Spark/Flink
Complex routing rules ✅ RabbitMQ Exchanges, priorities, DLQ, TTL
Edge/IoT devices ✅ NATS Core Lightweight client (~1MB), auto-reconnect
Service mesh communication ✅ NATS Core Built-in discovery, low overhead, no Envoy

Золотое правило: Начинайте с простейшего решения, которое полностью покрывает текущие требования. NATS позволяет начать с Core и эволюционировать к JetStream по мере роста требований к надежности — без смены технологического стека и клиентских библиотек.

Что дальше?

Мы полностью завершили детальный разбор экосистемы NATS. В следующем подразделе — 1. Pub Sub — мы поднимемся на уровень выше и разберем фундаментальные архитектурные паттерны асинхронного взаимодействия: Pub/Sub, Work Queue, Transactional Outbox, Inbox, распределенные саги и хореографию микросервисов.