Event-driven Go: Watermill, Kafka и async-обработка

После HTTP-сервиса и persistence у тебя есть синхронная картина мира: клиент шлёт запрос, сервер сходил в БД, ответил. Этот урок про другую половину: когда не должен ждать ответа здесь и сейчас и когда нужно, чтобы события доехали до десятков подписчиков.

Темы: что такое event-driven, чем pub/sub отличается от RPC, как Watermill упрощает работу с Kafka/RabbitMQ/NATS, и какие паттерны (CQRS, outbox, saga) делают это надёжным.

Главный источник, academy.threedots.tech/trainings/watermill-quickstart и официальная документация watermill.io.

1. Зачем нужен event-driven

Стандартный синхронный HTTP-сценарий:

клиент -> orders -> users -> payments -> notifications
                       |          |             |
                       |          v             |
                       +---->  inventory  <----+

Шесть RPC-вызовов в цепочку. Один тормозит, тормозит вся цепь. Один упал, упал весь запрос. Каждый сервис знает про всех, кому передавать дальше. Это нормально на старте, но плохо масштабируется.

Альтернатива: orders публикует событие order.created. Дальше никто никого не дёргает. Любой сервис подписывается на это событие и реагирует:

orders --(order.created)--> [pub/sub]
                              ├─-> payments
                              ├─-> inventory
                              ├─-> notifications
                              └─-> analytics

Что меняется:

• Слабая связность. orders не знает о payments. Чтобы добавить новую реакцию (например, audit-лог), создаёшь нового подписчика, никого не трогая.
• Resilience. Если notifications лежит, очередь буферизует событие, и оно доедет, когда notifications вернётся. orders не падает.
• Масштабирование. Каждый подписчик на своей скорости.
• Async-обработка. Долгие операции (видео-конвертация, отправка пуша) не держат HTTP-handler.

Минусы тоже честные: eventual consistency (между событием и реакцией секунды), сложнее дебажить (нет stack trace через систему), нужна инфраструктура (Kafka/RabbitMQ/NATS).

Правило практики: синхронный HTTP до тех пор, пока хватает. На event-driven переезжают, когда сервисов больше пяти, цепочки больше двух хопов, и нужна resilience. Не раньше.

2. Брокеры: Kafka, RabbitMQ, NATS, Redis Streams

Шорт-лист на 2026:

• Apache Kafka. Дефолт для high-throughput event log-ов. Партиционирование по ключу, retention неделями, потребители (consumer groups) переигрывают историю. Сложновата в эксплуатации (ZooKeeper или KRaft, тюнинг).
• RabbitMQ. Классический message broker. Routing keys, exchanges, очереди. Удобнее для классических task queue (worker берёт работу), хуже для event log.
• NATS (с JetStream). Лёгкий, быстрый, с JetStream даёт persistence. Хороший выбор для cloud-native, особенно когда не нужен Kafka-уровень retention.
• Redis Streams. Если уже есть Redis, можно начать с него. На небольших нагрузках, отлично. На больших, лимит памяти.
• Google Pub/Sub / AWS SQS+SNS. Managed-варианты в облаках, без забот по инфраструктуре.

В Go под все эти брокеры есть нативные клиенты (segmentio/kafka-go, Shopify/sarama, streadway/amqp, nats-io/nats.go). И есть слой поверх, Watermill.

3. Watermill: pub/sub за интерфейсом

Watermill (watermill.io) от ThreeDotsLabs скрывает разницу между брокерами за двумя интерфейсами:

type Publisher interface {
    Publish(topic string, messages ...*Message) error
    Close() error
}

type Subscriber interface {
    Subscribe(ctx context.Context, topic string) (<-chan *Message, error)
    Close() error
}

Это всё. Любой брокер, для которого есть Watermill-плагин (Kafka, RabbitMQ, NATS, Redis, Postgres LISTEN/NOTIFY, in-memory gochannel), реализует эти два интерфейса.

Поэтому код прикладной логики не зависит от брокера. Меняешь импорт, меняется транспорт. Это особенно полезно для тестов: подменяешь Kafka на in-memory gochannel, и unit-тест бежит без Docker.

Watermill построен пирамидой:

        EventBus / EventProcessor          ← cqrs (high-level)
               Router                      ← message.Router (mid-level)
       Publisher / Subscriber              ← pub/sub (low-level)

Чем выше, тем меньше boilerplate. Но и меньше контроля. Идиома, начинать с Router, опускаться в pub/sub только когда надо.

4. Уровень 1: Publisher и Subscriber

import (
    "github.com/ThreeDotsLabs/watermill"
    "github.com/ThreeDotsLabs/watermill/message"
    "github.com/ThreeDotsLabs/watermill-kafka/v3/pkg/kafka"
)

logger := watermill.NewSlogLogger(nil)

publisher, err := kafka.NewPublisher(kafka.PublisherConfig{
    Brokers:   []string{"kafka:9092"},
    Marshaler: kafka.DefaultMarshaler{},
}, logger)
if err != nil {
    return err
}
defer publisher.Close()

msg := message.NewMessage(watermill.NewUUID(), []byte(`{"order_id":"42"}`))
msg.Metadata.Set("source", "orders-service")

if err := publisher.Publish("OrderCreated", msg); err != nil {
    return fmt.Errorf("publish: %w", err)
}

Что внутри *message.Message:

• UUID, для дедупликации и трейсинга.
• Payload, []byte. Кодируешь сам (JSON, protobuf, gob).
• Metadata, map[string]string. Аналог HTTP-заголовков: correlation_id, source, version.

Подписка:

subscriber, err := kafka.NewSubscriber(kafka.SubscriberConfig{
    Brokers:       []string{"kafka:9092"},
    ConsumerGroup: "notifications-service",
    Unmarshaler:   kafka.DefaultMarshaler{},
}, logger)
if err != nil { return err }
defer subscriber.Close()

messages, err := subscriber.Subscribe(ctx, "OrderCreated")
if err != nil { return err }

for msg := range messages {
    var ev OrderCreated
    if err := json.Unmarshal(msg.Payload, &ev); err != nil {
        slog.Error("unmarshal", "err", err)
        msg.Nack()
        continue
    }
    if err := sendEmail(ev); err != nil {
        slog.Error("send email", "err", err)
        msg.Nack()
        continue
    }
    msg.Ack()
}

5. Ack и Nack: ключевые слова

Это два самых важных метода *Message. Брокер не выкинет сообщение, пока ты явно не подтвердил обработку.

• msg.Ack(), обработка прошла. Брокер удаляет (или сдвигает offset). Следующее сообщение приходит.
• msg.Nack(), обработка не прошла. Брокер пытается отдать снова (политика зависит от брокера).

Самые частые ошибки:

• Забыли Ack. Подписчик получает одно сообщение и виснет навсегда. Если видишь “consumer обработал одно событие и стоит”, это оно.
• Ack-нули битое сообщение. Сообщение потеряно навсегда. Если хочется убрать сломанное сообщение из очереди, сначала залогируй (или отправь в DLQ), потом Ack.
• Nack-аем неисправимую ошибку. Бесконечный цикл: брокер отдаёт, мы фейлим, отдаёт снова. Это poison message, его нужно вытаскивать в Dead Letter Queue.

6. HTTP -> Watermill: аналогия

Удобно держать в голове такую таблицу:

HTTP	Watermill
Request	Message
Body	Payload
Headers	Metadata
200 OK	Ack
500 Error	Nack
HTTP-клиент	Publisher
HTTP-сервер	Subscriber
Router (mux)	Router

Большинство решений из HTTP-мира (middleware, retry, timeout) переезжает один в один.

7. Уровень 2: Router

Свой for msg := range messages хорош для одного handler. Когда подписчиков становится несколько, ставишь *message.Router. Это аналог http.ServeMux, только для очередей.

router, err := message.NewRouter(message.RouterConfig{}, logger)
if err != nil { return err }

router.AddMiddleware(
    middleware.CorrelationID,
    middleware.Recoverer,
    middleware.Retry{
        MaxRetries:      3,
        InitialInterval: 100 * time.Millisecond,
        Logger:          logger,
    }.Middleware,
)

router.AddNoPublisherHandler(
    "send-welcome-email",   // имя handler-а (для логов)
    "UserRegistered",       // топик подписки
    subscriber,
    func(msg *message.Message) error {
        var ev UserRegistered
        if err := json.Unmarshal(msg.Payload, &ev); err != nil {
            return err  // вернёшь ошибку = Nack
        }
        return sendWelcomeEmail(ctx, ev.Email)
        // nil = Ack
    },
)

if err := router.Run(ctx); err != nil { return err }

Что даёт Router из коробки:

• Middleware. Recoverer (паника не убивает консьюмера), Retry (с экспоненциальным backoff), CorrelationID (прокидывает корреляционный id через цепочку), Throttle, IgnoreErrors. Можно писать свои.
• AddHandler vs AddNoPublisherHandler. Первый для “handler берёт топик A, выдаёт в топик B”. Второй для consumer-only.
• Graceful shutdown. Один ctx, все подписчики корректно закрываются.

Идиома, возвращать ошибку из handler. Router сам сделает Nack и применит retry. Не вызывай msg.Nack() руками внутри Router.