Публикация событий

Q: Зачем самодельная шина, если есть Kafka — не дешевле ли сразу брокер?

Дешевле — пока факты нужны только внутри процесса. Брокер — это сервер, конфигурация, мониторинг и ещё одна точка отказа; шина — десять строк на каналах и ноль зависимостей. Интерфейс Publisher — шов: когда факты поедут между сервисами, шина подменяется на NATS или Kafka без правок сервиса. Правило роста: шина → outbox → брокер, а не наоборот.

Q: Событие потерялось — как об этом узнать?

Три линии обороны. Метрики: счётчик дропов ErrSlowConsumer и глубина буферов из урока 07 показывают потерю в моменте. Сверка: издатель и потребитель считают свои счётчики по типам — расхождение видно на графике. Гарантия: для фактов, потеря которых недопустима, outbox — событие уже в таблице рядом с данными, и релей доопубликует; дедуп по ID гасит дубли.

Q: Порядок событий что-нибудь гарантирует?

Частично, и лучше знать границы. В шине подписчик видит факты в порядке вызова Publish — но дроп при полном буфере ломает последовательность. В Kafka порядок строг внутри партиции: ключ партиционирования (jobID) собирает факты одной задачи в одну партицию. Между партициями и подписчиками порядка нет, поэтому потребителя проектируют идемпотентным и невосприимчивым к перестановкам, а не надеются на глобальную ленту.

Опрос против события#

Посчитай цену опроса на конкретных числах. Пять систем хотят знать о завершении UppercaseAll: аудит, кэш, конвейер, вебхук, биллинг. С интервалом опроса в секунду это пять запросов в секунду вечно — даже когда событий нет ни одного; латентность знания — до секунды; каждый подписчик держит соединение, клиентский код и логику «спросил — забыл». Опрос переворачивает ответственность: получатель тратит источник, источник платит за чужое любопытство. События микросервисы связывают по-другому: источник один раз публикует факт, подписчики слушают, подключение нового интересанта не меняет издателя ни на строку — он вообще не знает, что кто-то присоединился. Цена публикации — одна операция, латентность доставки — миллисекунды, а история опроса превращается в ленту фактов, которую можно читать и постфактум. Но у события есть и своя цена, о которой опрос не знает: доставка может потеряться, порядок — перепутаться, подписчик — отстать и захлебнуться. Поэтому в этом уроке мы строим шину руками и тут же честно документируем, что она обещает, а что — нет: гарантии доставки важнее удобства конверта.

Канон: событие — конверт факта#

Канон stringsvc не ломается: Uppercase и Count остаются мгновенными, очередь и пул из урока 21 — работающими. Рядом появляется издатель, и у него есть строгая грамматика. Событие — всегда прошедшее время: stringsvc.uppercase_all.completed, а не completeJob — факты фиксируют, что случилось, а не что нужно сделать; приказ — это команда, и её место в очереди задач, а не в шине. Конверт события — четыре поля: ID (uuid, ключ дедупликации у потребителя), Type (имя факта — часть контракта: сменил — сломал всех подписчиков), OccurredAt (момент факта, а не момент доставки — они разъезжаются на лаг очереди) и Payload (данные факта в JSON: что именно произошло, идентификатор задачи, размер результата, текст ошибки). Payload несёт данные, а не инструкции: подписчик сам решает, что с фактом делать. Имена типов и форма payload — публичный контракт: меняйте аккуратно, новые версии вводите рядом со старыми (тип с суффиксом v2), как вводили новые endpoint'ы, не трогая старые. Событие неизменяемо: опубликовал — и больше не редактируешь; исправление — новое событие, а не переписывание старого.

Event и Publisher: контракт

// internal/events: контракт, общий для издателя и подписчиков
type Event struct {
        ID         string    // uuid события: ключ дедупликации у потребителя
        Type       string    // "stringsvc.uppercase_all.completed" — прошедшее время
        OccurredAt time.Time // момент факта, а не момент доставки
        Payload    []byte    // JSON: данные факта, а не инструкции
}

// Publisher — единственная зависимость сервиса от мира событий.
type Publisher interface {
        Publish(ctx context.Context, e Event) error
}

var ErrSlowConsumer = errors.New("event bus: subscriber buffer full")

Publisher — интерфейс, а не конкретная шина, по той же причине, что и Queue из урока 21: в тестах из урока 16 подставляется фейк, а в проде — шина на каналах или брокер; код сервиса не меняется ни на строку. ID события — не украшение: при at-least-once доставке потребитель увидит один факт дважды, и только ID отличает дубль от нового факта. OccurredAt фиксирует момент факта: подписчик, обработавший событие через час, всё ещё знает, когда оно произошло. ErrSlowConsumer объявлен рядом с контрактом, потому что это часть контракта: шина честно отказывается, а не молча роняет факт.

Шина на каналах: анатомия#

Первая шина, которую стоит построить, — уже в языке: map подписчиков с каналом у каждого. Subscribe(buffer) выдаёт подписчику его личный канал — буферизованный, с запасом, который подписчик выбрал сам; шина хранит только каналы и не знает, кто и как их читает. Publish раскладывает копию события в буфер каждого подписчика — и делает это без блокировки: select с default мгновенно возвращает ErrSlowConsumer, если чей-то буфер полон, вместо того чтобы затыкать издателя. Чтение — забота подписчика: range по каналу в собственной горутине, тот же канон, что и воркер пула из урока 21. Мьютекс защищает map: Publish берёт RLock (публикации параллелятся), Subscribe — полный Lock. Главное решение дизайна — политика при полном буфере: молча выбросить, заблокировать или отказать с ошибкой. Молчаливый дроп — обещание без исполнения, затор — отказ в обслуживании всех от одного медленного; честный отказ с ошибкой превращает проблему в видимую — её увидит лог из урока 06 и метрики из урока 07. Такую in-memory шину проще всего недооценить: она живёт в процессе и умирает с процессом, и об этом — честно — ниже.

Шина: Subscribe, Publish, Close

type Bus struct {
        mu   sync.RWMutex
        subs map[chan Event]struct{}
        done chan struct{}
}

func NewBus() *Bus {
        return &Bus{subs: make(map[chan Event]struct{}), done: make(chan struct{})}
}

func (b *Bus) Subscribe(buffer int) <-chan Event {
        select {
        case <-b.done: // после Close подписка — ошибка дизайна
                panic("event bus: subscribe after close")
        default:
        }
        ch := make(chan Event, buffer) // буфер — личный запас подписчика
        b.mu.Lock()
        defer b.mu.Unlock()
        b.subs[ch] = struct{}{}
        return ch
}

func (b *Bus) Publish(_ context.Context, e Event) error {
        b.mu.RLock()
        defer b.mu.RUnlock()
        for ch := range b.subs {
                select {
                case ch <- e: // слот нашёлся — доставили копию
                default:
                        return ErrSlowConsumer // честный отказ вместо затора
                }
        }
        return nil
}

func (b *Bus) Close() {
        close(b.done)
        b.mu.Lock()
        defer b.mu.Unlock()
        for ch := range b.subs {
                close(ch) // подписчики дочитывают буфер через range
                delete(b.subs, ch)
        }
}

RWMutex здесь не украшение: Publish читает map под RLock, поэтому несколько публикаций идут параллельно, а Subscribe с подписчиком-призраком исключён полным замком. Select с default — тот же детектор переполнения, что и в Pool.Put из урока 21: без него Publish блокировался бы на первом же полном канале, и один медленный подписчик остановил бы всех. Close повторяет канон дренажа: close(ch) — сигнал «больше не будет», range у подписчика добирает остаток буфера и завершается; тот же приём работал в дренаже пула. Паника в Subscribe после Close — сознательная грубость: подписка на умирающую шину — ошибка композиции, её нельзя вернуть ошибкой, потому что вызывающий уже владеет горутиной.

Издатель как middleware сервиса#

Куда встроить публикацию? Канон урока 15 подсказывает место: сервисный middleware-декоратор. EventingMiddleware оборачивает StringService: вызывает следующий слой, дожидается результата — и издаёт факт о том, что произошло. Порядок принципиален: событие издаётся после успеха, а не до — событие фиксирует свершившееся, и «завершено» до того, как работа закончилась, — это ложь, которая разъезжается с реальностью при первом же откате; то же правило честности работало в outbox-рассуждении о транзакциях. Что издавать — тоже решение: доменные факты (uppercase.completed, uppercase.failed, uppercase_all.queued), а не каждый вызов Count — шина не журнал отладки, для этого есть логгер из урока 06. Для асинхронных задач из урока 21 издаёт воркер: рядом с status.Finish он публикует uppercase_all.completed — факт доступен и аудиту, и вебхуку без единого опроса. Ошибка публикации не превращается в ошибку операции: работа свершилась, факт есть, шина подводит — значит, дроп фиксируется в логе и метриках, а клиент получает свой честный ответ; событие — побочный след операции, а не её результат.

Middleware-издатель поверх канона

type EventingMiddleware struct {
        next      StringService
        publisher events.Publisher
        logger    log.Logger // канон из урока 06
}

func (mw EventingMiddleware) Uppercase(ctx context.Context, s string) (out string, err error) {
        out, err = mw.next.Uppercase(ctx, s) // сначала факт: работа выполнена
        t := "stringsvc.uppercase.completed"
        payload := map[string]string{"len": strconv.Itoa(len(out))}
        if err != nil {
                t = "stringsvc.uppercase.failed"
                payload["err"] = err.Error()
        }
        e := events.Event{
                ID:         events.NewID(),
                Type:       t,
                OccurredAt: time.Now().UTC(),
                Payload:    marshal(payload),
        }
        if perr := mw.publisher.Publish(ctx, e); perr != nil {
                // факт уже свершился: дроп события не меняет ответ клиенту
                mw.logger.Log("during", "publish", "type", t, "err", perr)
        }
        return out, err
}

Декоратор повторяет форму loggingMiddleware и instrumentingMiddleware из урока 15: те же поля next и зависимости, то же чтение через метод интерфейса StringService. Разница в одном: middleware не меняет поведение, а свидетельствует о нём — поэтому Publish стоит после вызова next и его ошибка не поднимается наверх. Композиция в main из урока 17 становится трёхслойной: logging → eventing → service — и порядок не случаен: логгер снаружи видит и саму публикацию, включая её ошибки. Для задач из очереди тот же вызов Publish встаёт в воркер рядом с p.status.Finish — конверт собирается один раз, а издают оба мира.

Гарантии доставки: три уровня честности#

Шина обещает не всё, и это надо проговорить терминами, а не настроением. At-most-once: событие отправляется один раз, повторов нет, потеря возможна — при полном буфере подписчика, при панике, при падении процесса; in-memory шина из этого урока живёт ровно здесь, и это не дефект, а честная цена нулевых зависимостей. At-least-once: событие повторяется, пока подписчик не подтвердит обработку — доставка надёжна, но дубликаты неизбежны, и потребитель обязан быть идемпотентным: дедупликация по Event.ID в таблице обработанных фактов превращает поток дублей в один результат. Exactly-once в чистом виде — маркетинг: на практике это at-least-once плюс дедуп на стороне получателя, и лучше сразу проектировать потребителя идемпотентным, чем верить слогану брокера. Правило выбора: цена потери факта определяет уровень. Для фактов внутри процесса (обновить кэш, посчитать статистику) — at-most-once достаточно; для фактов с последствиями (списать деньги, уведомить клиента) — как минимум outbox и подтверждения; а интерфейс Publisher — тот шов, где реализация меняется без переписывания сервиса, как менялась Queue в прошлом уроке.

Outbox: публикация без лжи#

Между шиной и базой данных есть щель, в которую проваливаются события-призраки. Издал до коммита — транзакция откатилась, подписчики уже бегают с фактом, которого не было; издал после коммита — процесс упал между commit и Publish, факт есть, события нет. Обе лжи лечатся паттерном outbox: событие пишется строкой в таблицу outbox в той же транзакции, что и само изменение данных, — атомарность базы гарантирует: либо оба, либо ничего. Публикацию делает релей — отдельный воркер, который читает неотмеченные строки, публикует их в шину или брокер и помечает отправленными после подтверждения. Релей — это тот же тикер из урока 21: time.NewTicker, батч, select с остановкой, дренаж при SIGTERM. Гарантия получается at-least-once: релей может умереть после публикации и до пометки — факт уйдёт дважды, и снова спасает дедупликация по ID на потребителе. Глубина таблицы outbox — лучшая метрика здоровья конвейера: растёт — значит, релей отстал или брокер лежит, и это видно в метриках из урока 07 раньше, чем первый клиент заметит молчание вебхуков.

NATS и Kafka: когда канала стало мало#

Шина в памяти умирает с процессом — для фактов, которые нужны другим сервисам и обязаны переживать рестарты, дорога ведёт к брокеру. Запрос «event publisher go» в поиске приводит не в go-kit, а к брокерам — и это правильно: издатель у тебя уже есть, вопрос только в транспорте. NATS: лёгкий брокер с адресацией по темам; core NATS — at-most-once и очень быстрый, а JetStream добавляет персистентный поток: подтверждения, повторы, перечитывание — по сути, outbox-семантику из коробки. Kafka: распределённый журнал — топики, партиции, offset у каждого потребителя; порядок гарантирован внутри партиции, поэтому ключ партиционирования (например, jobID) — решение издателя: факты одной задачи лягут в одну партицию и придут по порядку. Второй подарок Kafka — retention: журнал хранится днями, новый подписчик может перечитать историю с начала, а не ждать свежих фактов; consumer group делит партиции между членами группы — масштабирование подписчиков без дублей. Правило роста масштаба однонаправленное: шина в процессе → outbox с релеем → брокер; в обратную сторону путь обычно не возвращается, потому что интерфейс Publisher уже всё спрятал.

Publish делает ch <- e без select и default. У одного из пяти подписчиков буфер полон. Что произойдёт?

Когда middleware-издатель должен публиковать событие stringsvc.uppercase.completed?

Процесс упал с SIGKILL, когда в буферах шины лежали три непрочитанных события. Какая гарантия у in-memory шины?

Буфер подписчика против default

Прочитайте код и выберите, что он выведет:

ch := make(chan Event, 2)
ch <- Event{Type: "a.completed"}
ch <- Event{Type: "b.completed"}

select {
case ch <- Event{Type: "c.completed"}:
        fmt.Println("delivered")
default:
        fmt.Println("dropped")
}

Собери Publish шины

Восстанови четыре пропуска публикации: чтение под замком, раздача копий и честный отказ при полном буфере.

func (b *Bus) Publish(_ context.Context, e Event) error { b.mu.RLock() defer b.mu.() for ch := range b. { select { case e: // слот нашёлся — доставили копию : return ErrSlowConsumer // честный отказ вместо затора } } return nil }

Путь события

Расставь этапы от свершившегося факта до его читателей.

Строки

Ваша программа

Нажимайте на строки слева, чтобы собрать программу

Транспорты событий: кто что обещает#

Критерий	Шина в памяти	Outbox + релей	NATS JetStream	Kafka
Гарантия доставки	at-most-once	at-least-once	at-least-once	at-least-once
Порядок фактов	порядок вызова Publish	порядок записи в таблицу	в пределах потока	внутри партиции
Переживает падение	нет	да: строка в БД	да: поток на диске	да: журнал на диске
Перечитывание истории	нет	перечитывание таблицы	да: replay потока	да: offset в ретеншене
Медленный подписчик	ErrSlowConsumer	лаг в таблице	лаг подписчика	лаг группы
Дедупликация	по ID на потребителе	по ID на потребителе	по ID / ack	по ID на потребителе
Зависимости	ни одной	только БД	сервер NATS	кластер Kafka
Когда уместен	факты внутри процесса	факты рядом с транзакцией	лёгкий надёжный поток	журнал и конвейеры
Наблюдаемость (урок 07)	дропы и глубина буферов	глубина outbox	лаг потока	лаг consumer group

Один медленный подписчик — беда всех

Шина коллективна по определению: пять подписчиков делят один вызов Publish, и буфер каждого — общий риск. Блокирующая отправка превращает одного отставшего в затор для всех; молчаливый дроп превращает его в тихую потерю фактов. Выход из кода этого урока — отказ с ErrSlowConsumer — работает, только если его видно: логируй дропы с типом события, поднимай счётчик, ставь алерт. И не давай подписчикам буферы на глазок: буфер — личный запас на короткий всплеск, а не место, где медленный потребитель прячет свою медленность. Подписчик, который регулярно ловит ErrSlowConsumer, требует не буфера побольше, а отдельного разговора о своей скорости.

Метрики шины

Шина без метрик — чёрный ящик с тайной жизнью. Четыре счётчика закрывают вопрос: событий издано по типам, дропов ErrSlowConsumer по типам, число активных подписчиков, возраст старейшего события в буфере — как возраст старейшей задачи в очереди из урока 21. Для outbox добавь глубину таблицы и возраст старейшей неотправленной строки: они растут, когда релей отстал или брокер лежит. Всё это — счётчики и гейджи из урока 07; алерт на рост дропов поставит тебя в известность раньше, чем подписчик напишет, что данных не хватает.

Честно о запуске кода

Шина, издатель и подписчики живут в реальном процессе вместе с брокерами и потребителями, в песочнице курса этот код не исполняется — интерактивы урока построены на чтении кода. Локально всё оживает за минуту: собери Bus, подцепи два подписчика (лог и фейковый вебхук), затормози одного time.Sleep'ом и посмотри на ErrSlowConsumer в логе, а затем Ctrl+C и прочитай дренаж шины в секвенции остановки до exit 0.

Термины урока

Термин

Событие

нажмите, чтобы перевернуть

Определение

Конверт факта в прошедшем времени: ID, Type, OccurredAt, Payload; не приказ, а свидетельство о свершившемся; неизменяемо после публикации.

1 / 6

Что дальше#

Сервис научился свидетельствовать о фактах: конверт, шина, middleware-издатель, outbox и карта транспортов от канала до Kafka — события разгружают опрос и развязывают зависимости. Но при этом каждый входящий запрос в stringsvc по-прежнему анонимен: endpoint честно выполнит Uppercase для кого угодно, кто дотянулся до порта, а событие в шине не несёт имени того, кто заказал работу. Урок 23-bezopasnost закроет последнюю дыру канона: аутентификация — JWT на endpoint, identity в контексте, 401 против 403, HS256 против RS256 и токены, которые едут между сервисами вместе с запросом. Карта курса — на странице курса; очередь, воркеры и тикер релея — в уроке 21-async; декораторы, поверх которых стоит издатель, — в уроке 15-service-middleware; метрики дропов и лагов — в уроке 07-metriki; дренаж шины в секвенции остановки — в уроке 19-graceful.

Песочница урока

Редактируйте код и запускайте его прямо здесь — интерпретатор работает в браузере, ничего устанавливать не нужно. За каждый запуск — +2 XP.

Вывод

Запустите код, чтобы увидеть вывод.

Коротко: что запомнить

Событие — конверт факта в прошедшем времени: ID, Type, OccurredAt, Payload; команда приказывает, событие свидетельствует.
In-memory шина — map каналов под RWMutex: Subscribe выдаёт личный буфер, Publish раскладывает копии без блокировки.
Полный буфер подписчика — ErrSlowConsumer, а не затор и не молчаливый дроп: отказ должен быть видимым.
Издатель — сервисный middleware из урока 15: Publish после успеха, ошибка шины не меняет ответ операции.
Гарантии: at-most-once у шины в памяти, at-least-once требует дедупа по Event.ID, exactly-once — at-least-once плюс дедуп.
Outbox: событие в таблице рядом с транзакцией, релей на тикере публикует и помечает — без событий-призраков.
NATS JetStream — лёгкий персистентный поток; Kafka — журнал с порядком внутри партиции и перечитыванием по offset.
Дренаж шины — тот же канон урока 19: close каналов, range дочитывает буферы, затем закрытие зависимостей.

Домашнее задание

Поставь события в stringsvc. Первое: собери пакет internal/events с Event, Publisher и Bus из урока; подцепи двух подписчиков — лог-подписчика пишет факты через kitlog из урока 06, вебхук-фейк печатает уведомления. Второе: добавь EventingMiddleware поверх StringService и проверь композицию logging → eventing → service из урока 17; убедись, что failed-факт несёт текст ошибки, а успешный — размер результата. Третье: затормози вебхук-подписчика time.Sleep'ом, подави десять событий подряд и поймай ErrSlowConsumer — дроп должен появиться и в логе, и в счётчике метрик из урока 07. Четвёртое: собери таблицу outbox с колонками id, type, payload, occurred_at, sent_at и релей на тикере из урока 21, который публикует и помечает строки; разорви связь между commit и Publish, чтобы увидеть at-least-once и дедуп по ID на потребителе. Пятое: запусти остановку с непустыми буферами и убедись по логам, что bus.Close стоит после дренажа HTTP из урока 19 и все события дочитаны до exit 0.

Вопросы и ответы по уроку

Зачем самодельная шина, если есть Kafka — не дешевле ли сразу брокер?

Дешевле — пока факты нужны только внутри процесса. Брокер — это сервер, конфигурация, мониторинг и ещё одна точка отказа; шина — десять строк на каналах и ноль зависимостей. Интерфейс Publisher — шов: когда факты поедут между сервисами, шина подменяется на NATS или Kafka без правок сервиса. Правило роста: шина → outbox → брокер, а не наоборот.

Событие потерялось — как об этом узнать?

Три линии обороны. Метрики: счётчик дропов ErrSlowConsumer и глубина буферов из урока 07 показывают потерю в моменте. Сверка: издатель и потребитель считают свои счётчики по типам — расхождение видно на графике. Гарантия: для фактов, потеря которых недопустима, outbox — событие уже в таблице рядом с данными, и релей доопубликует; дедуп по ID гасит дубли.

Порядок событий что-нибудь гарантирует?

Частично, и лучше знать границы. В шине подписчик видит факты в порядке вызова Publish — но дроп при полном буфере ломает последовательность. В Kafka порядок строг внутри партиции: ключ партиционирования (jobID) собирает факты одной задачи в одну партицию. Между партициями и подписчиками порядка нет, поэтому потребителя проектируют идемпотентным и невосприимчивым к перестановкам, а не надеются на глобальную ленту.

Термины урока

Ключевые понятия из этого урока — короткие определения в глоссарии. Наведите на термин, чтобы вспомнить, откройте карточку для подробностей и ссылки на практику.

Опрос против события#

Канон: событие — конверт факта#

Event и Publisher: контракт

Шина на каналах: анатомия#

Шина: Subscribe, Publish, Close

Издатель как middleware сервиса#

Middleware-издатель поверх канона

Гарантии доставки: три уровня честности#

Outbox: публикация без лжи#

NATS и Kafka: когда канала стало мало#

Буфер подписчика против default

Собери Publish шины

Путь события

Транспорты событий: кто что обещает#

Термины урока

Что дальше#

Песочница урока

Коротко: что запомнить

Домашнее задание

Вопросы и ответы по уроку

Термины урока

Изучайте дальше

gRPC-Go: высокопроизводительный RPC