--- title: "Итоговый проект: событийный пайплайн" description: "Итоговый kafka проект go: событийный пайплайн заказов. Продюсер с версионированным payload, два консьюмера, DLQ, лаг, graceful shutdown и тест." course: "Kafka в Go: продюсеры и консьюмеры на kafka-go" url: "https://golangmaster.ru/courses/kafka-go/lessons/25-itogovyj-proekt/" published: "2026-09-19" updated: "2026-09-29" minutes: 24 --- # Итоговый проект: событийный пайплайн *Курс «Kafka в Go: продюсеры и консьюмеры на kafka-go», урок 25 из 25. ~24 минут. Обновлено 29.09.2026. Источник: https://golangmaster.ru/courses/kafka-go/lessons/25-itogovyj-proekt/* Курс собран в рабочую систему: сервис заказов публикует событие order.created в топик с тремя партициями, а два независимых консьюмера разбирают этот лог — инвентарь резервирует товар, аналитика копит статистику. Нового API в уроке почти нет: двадцать четыре урока выдали все детали по одной, здесь проверяется умение соединять их в конвейер, который не теряет события, не вязнет в яд-данных и останавливается по деплою без потери офсетов. Каждый узел помечен номером урока-источника: непонятный стык открывается заново за минуту, а каркас переносится в свой проект почти без правок. Прогон — локально на Docker-брокере из третьего урока: механика воспроизводится полностью, кластерные сценарии отдельно не проверяются. Три партиции, две группы и один Writer — этого достаточно, чтобы повторить в миниатюре всё, чем живёт продовый событийный конвейер. Пайплайн живёт вокруг одного топика orders с тремя партициями. Продюсер пишет событие order.created, ключом служит order_id: [hash от ключа выбирает партицию](/courses/kafka-go/lessons/05-klyuchi-particionirovanie/), все события одного заказа ложатся в одну очередь, и порядок внутри заказа гарантирован — порядок между заказами не нужен, и это осознанный размен пятого урока. Читателей два, и они не мешают друг другу: инвентарь слушает в группе inventory, аналитика — в группе analytics, а потребительские группы устроены так, что один и тот же лог читается каждой независимо, со своими офсетами и своим темпом. Внутри группы партиции распределяются между членами, поэтому масштабирование сводится к добавлению инстансов. Таймауты, ребалансы и батчи наследуются из прошлых уроков: здесь они впервые работают в одной программе, и каждое решение проверяется не схемой, а поведением системы при сбое — потерей, повтором или простоем. ### Кирпич проекта — урок-источник | Кирпич | Урок | Роль в проекте | | --- | --- | --- | | Balancer: &kafka.Hash{} и ключ order_id | 05 | события заказа — в одну партицию: порядок на заказ | | OrderEvent с полем Version | 22 | версионированный payload переживает эволюцию схемы | | GroupID у Reader | 07 | инвентарь и аналитика читают один лог независимо | | FetchMessage + CommitMessages | 08 | резерв подтверждён — офсет продвинут, не раньше | | Счётчик неудач и топик .dlq | 19 | яд-данные снимаются с конвейера после трёх попыток | | ReadLag | 21 | лаг группы виден вживую, без остановки чтения | | контекст с отменой | 18 | деплой не рвёт обработку на полуслове | | уникальный топик на прогон | 20 | тест не встречает чужие сообщения | ### Продюсер: событие с версией ```go type OrderEvent struct { Event string `json:"event"` Version int `json:"version"` OrderID string `json:"order_id"` SKU string `json:"sku"` Qty int `json:"qty"` } var writer = kafka.NewWriter(kafka.WriterConfig{ Brokers: []string{"localhost:9092"}, Topic: "orders", Balancer: &kafka.Hash{}, // ключ решает партицию }) func PublishOrderCreated(ctx context.Context, o Order) error { value, err := json.Marshal(OrderEvent{ Event: "order.created", Version: 2, OrderID: o.ID, SKU: o.SKU, Qty: o.Qty, }) if err != nil { return fmt.Errorf("order %s: %w", o.ID, err) } return writer.WriteMessages(ctx, kafka.Message{ Key: []byte(o.ID), Value: value, }) } ``` Balancer kafka.Hash направляет сообщение по ключу: один заказ — одна партиция, события не перепутаются. Поле Version — якорь эволюции схем: консьюмер читает его первым и понимает, как разбирать остальное, приём из двадцать второго урока. Writer надёжен по умолчанию: NewWriter подставляет RequireAll, ошибка WriteMessages означает недоставку пачки, и сервис возвращает её наверх вместо ложного подтверждения заказа. ### Проверка: ключ и партиция **Вопрос:** Два события заказа order_id=ord-7351 ушли с интервалом в секунду. Куда они попадут? - ✅ В одну партицию: Hash от одинакового ключа выбирает её каждый раз - В разные партиции: Writer распределяет сообщения по кругу - В одну партицию только при RequireAll, иначе по кругу - В партицию ноль: ключи читаются только консьюмером kafka.Hash хеширует байты ключа и по хешу выбирает партицию, поэтому одинаковый ключ всегда даёт одинаковую партицию — события одного заказа идут строго по порядку. Round-robin балансер разослал бы их по разным, а консьюмеру ключ маршрутизацией не управляет: решает партиционер продюсера в момент записи. ### Продюсер: контракт события order.created Событие — не строка, а контракт. Поле Event называет факт, поле Version сообщает читателю, на какой схеме данных он смотрит: вторая версия упразднила поле Address и добавила PromoCode, и старый консьюмер, встретив Version: 2, обязан либо уметь читать новую форму, либо честно отказать — но не молча взять первые поля, потеряв остальное. Дисциплина из урока про эволюцию схем: номер растёт при несовместимом изменении, совместимые — добавление необязательного поля — проходят без смены версии. Имя события и имя топика пишутся константами рядом с типом payload: опечатка в строковом литерале компиляцией не ловится, а расхождение между продюсером и консьюмером всплывёт в рантайме, когда заказ уже опубликован. Публикация идёт после сохранения заказа в базу: транзакция первична, событие — уведомление о ней. Порядок обратный создал бы заказ, которого нет: консьюмеры зарезервируют товар по данным, которые база отвергла. Ошибку WriteMessages сервис не глотает: заказ сохранён в базе, но событие не ушло — логируем и отдаём в сверочную процедуру, которая перечитает базу и допубликует недостающее. ### Консьюмер-инвентарь: обработка до коммита Инвентарь подписан группой inventory и делает ровно то, что отработано в восьмом уроке: FetchMessage достаёт запись, reserveStock резервирует SKU на складе, CommitMessages продвигает офсет — строго после успешной обработки. Порядок хранит корректность: коммит до резервирования роняет сервис между шагами — офсет продвинут, товар не зарезервирован, заказ исчез, и Kafka его не вернёт, потому что группа считает работу сделанной. После обработки наоборот: сбой до коммита вернёт запись тому же или другому члену группы, резерв выполнится второй раз, и это уже видно по order_id в логах. Повтор гасится идемпотентностью: резерв по ключу заказа выполняется один раз, вторая попытка помечает дубль, а не списывает товар заново. Ошибки классифицируются по девятнадцатому уроку: сбой сети уходит на ретрай без продвижения офсета, битый payload после трёх неудач — в orders.dlq, о котором ниже, и партиция оживает. Пауза между попытками — короткий sleep с бэкоффом: он отделяет медленного брокера от постоянного брака и не даёт циклу сжечь процессор в холостую. ### Цикл инвентаря с DLQ ```go const maxFails = 3 func runInventory(ctx context.Context, brokers []string) error { reader := kafka.NewReader(kafka.ReaderConfig{ Brokers: brokers, GroupID: "inventory", Topic: "orders", MinBytes: 1, MaxBytes: 10e6, MaxWait: time.Second, }) defer reader.Close() dlq := kafka.NewWriter(kafka.WriterConfig{Brokers: brokers}) defer dlq.Close() fails := 0 for { m, err := reader.FetchMessage(ctx) if err != nil { return ctx.Err() // читателя закрыли: чистый выход } if err := reserveStock(m); err != nil { fails++ if fails < maxFails { continue // офсет на месте: запись вернётся заново } _, err = dlq.WriteMessages(ctx, kafka.Message{ Topic: "orders.dlq", // Topic в сообщении: у Writer его нет Key: m.Key, Value: m.Value, // сырьё, без пересборки Headers: []kafka.Header{ {Key: "dlq-reason", Value: []byte("reserve x3")}, }, }) if err != nil { return err // DLQ недоступна: коммитить нельзя } } fails = 0 if err := reader.CommitMessages(ctx, m); err != nil { return err } } } ``` Цикл собирает три ветки девятнадцатого урока: успех — коммит; временный сбой — офсет на месте, запись придёт заново; третий провал — сырьё уходит в orders.dlq с причиной в заголовке, и лишь затем CommitMessages перешагивает яд. Writer для DLQ создан без поля Topic: назначение записано в каждом сообщении, и один Writer обслуживает разные топики — иначе kafka-go вернёт ошибку конфликта конфигурации. > **Живое место: коммит перед обработкой.** Инверсия порядка — самая частая правка новичков: «закоммитим сразу, чтобы не зависнуть на яде» — и конвейер начинает тихо терять заказы: сервис падает между коммитом и резервом, а Kafka считает событие обработанным. Вторая ловушка рядом: DLQ-Writer переиспользован из продюсера, где поле Topic заполнено, — WriteMessages отвечает ошибкой, ядро записи не уходит, и цикл смерти возобновляется с тем же сообщением. Оба случая всплывают не сразу: потери копятся в течение дня, а застрявший цикл выглядит как лаг, который вот-вот выровняется. Правила отсюда: коммит только после обработки, а Writer для DLQ — отдельный экземпляр без Topic в конфиге, и ошибка записи в DLQ никогда не глотается. ### Проверка: цена преждевременного коммита **Вопрос:** Резерв товара поставлен до reserveStock, CommitMessages вызван перед ним, сервис упал между двумя шагами. Что с заказом? - ✅ Заказ потерян: офсет продвинут, событие никто не перечитает - Заказ обработается повторно: офсет остался на месте - Ничего: Kafka повторит недоставленное событие автоматически - Событие уйдёт в DLQ: упавший хендлер выглядит как poison Коммит продвинул группу мимо события, и после падения чтение начнётся со следующего: резерв не выполнен, событие не вернётся — Kafka хранит то, что записано, а не то, что было нужно бизнесу. Единственный безопасный порядок — обработка, затем коммит; ценой остаётся возможный повтор при падении между шагами, и его гасит идемпотентность резерва по order_id. ### Консьюмер-аналитика: тот же лог, своя группа Аналитика читает те же события, но живёт в группе analytics: свои офсеты, свой темп, своя раскладка партиций. Инвентарь встанет на паузу — аналитика не заметит; аналитика отстанет на час — инвентарь продолжит резервировать. [Fanout через группы](/courses/kafka-go/lessons/07-gruppy-consjumerov/) стоит нуля копирований: один физический лог обслуживает сколько угодно независимых потребителей, и новый подписчик подключается без единой правки в продюсере. Настройки чтения у групп разные сознательно: инвентарю важен короткий MaxWait — резерв не должен ждать сбора пачки; аналитике важнее крупный MinBytes — статистика глотает десятки тысяч событий в секунду и берёт их батчами из десятого урока. Обработчик аналитики проще некуда: разобрать payload по версии, обновить счётчики, закоммитить. Ей не нужен строгий realtime: отставание на минуты не искажает суточные срезы, поэтому крупный MinBytes и батчи уместнее, чем погоня за каждым сообщением. Но и здесь ошибка данных после порога уходит в собственный DLQ: витрина, построенная на потоке с ядом, врёт во всех отчётах сразу. ### Что вернёт ReadLag у группы? ```go // Reader с GroupID: "inventory", топик orders. // Написали 40 000 событий, обработали 39 200. lag, err := reader.ReadLag(ctx) fmt.Println(lag, err) ``` **Что выведет программа?** - ✅ 800 nil: ReadLag сравнит конец партиций с офсетами группы - −1 nil: при групповой подписке лаг недоступен - 0 nil: лаг считается только для читателей без группы - ошибка: ReadLag работает только на партиционированном Reader Lag() при групповой подписке бессилен — возвращает минус один, потому что читатель не ведёт офсет сам. ReadLag из двадцать первого урока сделан для этого случая: он сходил за последними офсетами партиций и за закоммиченными офсетами группы и вернул разницу. Именно этот вызов вешают на метрику, чтобы алерт поднялся раньше, чем заказы застрянут в невидимой очереди. ### Наблюдаемость и остановка: лаг и контекст Конвейер, который нельзя наблюдать, нельзя запускать в проде. Лаг — его пульс: ReadLag у обоих читателей раз в десять секунд, число уезжает в метрику, рост без спада — алерт, потому что каждая единица лага — заказ, который ещё не обработан. Второй сигнал — лог самого цикла: ключ, ветка ошибки, длительность обработки; без него разбор инцидента превращается в угадывание по косвенным признакам. Остановка идёт через контекст из восемнадцатого урока: сигнал завершения отменяет корневой ctx, FetchMessage возвращается с ошибкой контекста, цикл выходит, defer закрывает Reader и DLQ-Writer, а необработанный офсет остаётся не закоммиченным — после рестарта группа начнёт с него и ничего не потеряет. Деплой превращён в процедуру: сначала перестаём принимать заказы по HTTP, затем отменяем консьюмеров, затем ждём закрытия Writer, у которого могут остаться незавершённые записи. Таймауты вокруг внешних вызовов — из того же урока: без них одна медленная партиция удержит весь пайплайн. Метрика лага без порогов бесполезна: настраиваются алерт на затяжной рост и абсолютный предел, выше которого конвейер заведомо не успевает за потоком заказов. ### Тест: уникальный топик на прогон Конвейер получает интеграционный тест из двадцатого урока, и его главный приём — изоляция: имя топика собирает fmt.Sprintf с UnixNano, двух прогонов с одинаковым названием не бывает даже при параллельном запуске пакетов. Тест сам создаёт топик через CreateTopics с NumPartitions: 3 — автосоздание брокера отдало бы конфигурацию дефолтам, а порядок здесь проверяется явно: три события одного заказа уходят продюсером, группа inventory читает их, и резерв вызывается строго в порядке отправки. Второй сценарий — ветка DLQ: в поток подмешивается payload, который reserveStock всегда отвергает, и тест ждёт, что после трёх неудач сырьё окажется в orders.dlq с заголовком-причиной, а следующее событие обработается без рестартов. t.Cleanup регистрирует DeleteTopics сразу после создания: отложенные вызовы срабатывают при любом исходе, включая t.Fatalf, и брокер остаётся чистым. Гонки между продюсером и читателем снимаются FetchMessage с таймаутом контекста — тем же приёмом, что и в боевом цикле, поэтому тестовая обвязка совпадает с продовой почти строка в строку. > **Про песочницу: конвейер вживую.** Интерактивы исполняются в песочнице курса на программах без сети, а живой прогон воспроизводится локально: брокер в Docker из третьего урока, CreateTopics создаёт orders и orders.dlq с тремя партициями, продюсер шлёт десяток заказов — два Reader с разными GroupID показывают, как инвентарь резервирует, аналитика копит статистику, и оба скроллят один лог со своей скоростью. Прервите читателя по ctrl-c в середине обработки и запустите снова: группа продолжит с последнего закоммиченного офсета — нагляднейшая демонстрация контракта «обработка до коммита». ### Чек-лист курса - Ключ решает порядок: один заказ — одна партиция, kafka.Hash на продюсере. - Событие — контракт: имя константой, Version в payload, совместимые изменения без смены версии. - Обработка до коммита: не обработано — офсет на месте; повторы гасит идемпотентность по order_id. - Группы дают fanout: новый потребитель подключается без правок продюсера. - Ошибки классифицируются: инфраструктура ретраится, данные после порога уходят в .dlq с причиной в заголовке. - Лаг — метрика номер один: ReadLag на каждом читателе, алерт на затяжной рост. - Остановка через контекст: отмена, закрытие Reader и Writer, рестарт продолжит с не закоммиченного офсета. - Тесты изолируются: уникальный топик на прогон, CreateTopics и DeleteTopics в t.Cleanup. ### Соберите продюсера проекта Допишите балансер, который направляет события по ключу, и имя события этого проекта. ```go writer := kafka.NewWriter(kafka.WriterConfig{ Brokers: []string{"localhost:9092"}, Topic: "orders", Balancer: &kafka.___, }) event := OrderEvent{ Event: "order.___", Version: 2, OrderID: o.ID, SKU: o.SKU, Qty: o.Qty, } ``` Подсказка: Балансер по хешу ключа называется kafka.Hash, а событие из этого проекта — order.created. Правильные ответы в пропусках: `Hash`, `created` ### Соберите цикл инвентаря Расставьте строки цикла: чтение, выход по контексту, обработка, порог неудач, DLQ, коммит. Правильный порядок строк: 1. `m, err := reader.FetchMessage(ctx)` 2. `if err != nil { return ctx.Err() }` 3. `err = reserveStock(m)` 4. `if err != nil { fails++ }` 5. `if fails >= maxFails {` 6. ` err = sendToDLQ(ctx, m)` 7. ` if err != nil { return err }` 8. `}` 9. `fails = 0` 10. `return reader.CommitMessages(ctx, m)` ### Карточки: узлы пайплайна - **order.created** — Событие о создании заказа: имя-константа, JSON payload с полем Version, ключ order_id в kafka.Message. - **kafka.Hash** — Балансер продюсера: хеш ключа выбирает партицию — события одного заказа идут по порядку, между заказами порядка нет. - **Fanout через группы** — Один лог обслуживает несколько consumer groups: у каждой свои офсеты и темп, копирование данных не требуется. - **orders.dlq** — Топик мёртвых писем: после трёх неудач туда уходит сырое сообщение с причиной в заголовке, затем коммит перешагивает яд. - **reader.ReadLag(ctx)** — Отставание читателя: конец партиций минус офсет группы. Работает при GroupID, в отличие от Lag(), которая вернёт минус один. - **ctx в FetchMessage** — Отмена контекста — сигнал остановки: FetchMessage вернёт ctx.Err(), цикл выйдет, defer закроет Reader и Writer. Проект собран, и это уже не учебная программа: один топик, две независимые группы, DLQ, наблюдаемость и аккуратная остановка — каркас, который в проде дорастает кластером и реестром схем, не меняя формы. Что дальше. Задачи со сроком и статусом — напоминание, отложенная отправка, повтор по расписанию — комфортнее в [asynq](/courses/asynq/): там приоритеты, ретраи и планировщик встроены, и Kafka для них избыточна. Кэш витрины и горячие остатки — территория [Redis](/courses/go-redis/): событийный пайплайн пишет в базу, Redis закрывает чтение, и связка «событие — кэш» разобрана в отдельном курсе. Если вы прошли уроки по порядку — у вас на руках работающий стенд и карта типовых граблей; откройте свой проект, создайте один топик и начните с продюсера, а остальное достраивайте по чек-листу выше. Каркас переносится как отдельный пакет: продюсер, две группы и DLQ-логика не завязаны на функцию main, и переезд в сервисную структуру сводится к разрезанию на модули. ## Коротко: что запомнить - Пайплайн: заказ → событие order.created → топик orders на три партиции → две группы консюмеров. - Ключ order_id и kafka.Hash дают порядок на заказ без глобальной упорядоченности. - Payload версионирован: поле Version говорит читателю, по какой схеме разбирать остальное. - Инвентарь: FetchMessage, резерв, CommitMessages после обработки; идемпотентность гасит повторы. - Аналитика живёт в своей группе: тот же лог, свои офсеты, ноль копирований. - Ошибки данных после трёх неудач уходят в orders.dlq с причиной в заголовке, затем коммит. - ReadLag на каждом читателе — метрика и алерт: рост лага виден раньше жалоб. - Остановка через контекст: отмена, закрытие Reader и Writer, рестарт продолжит с не закоммиченного офсета. ## Домашнее задание Соберите конвейер на Docker-брокере из третьего урока: CreateTopics создаст orders с NumPartitions: 3 и orders.dlq, продюсер опубликует серию заказов, группы inventory и analytics прочитают лог независимо. Вбросьте в поток запись, которую reserveStock всегда отвергает, и убедитесь по логам, что после трёх неудач она оказалась в DLQ с заголовком-причиной, а партиция продолжила работу. Отдельно проверьте остановку: прервите консьюмера посреди обработки и запустите заново — обработка продолжится с не закоммиченного офсета. Финал — прогон теста из этого урока против того же стенда. Усилить: повесьте ReadLag на тикер раз в десять секунд и объясните, почему Lag() у вашего читателя возвращает минус один. ## Вопросы и ответы по уроку ### Чем итоговый пайплайн отличается от продовой версии? Масштабом и краями, а не архитектурой: в проде топик с репликами и RequireAll из пятнадцатого урока, конфиги читателей выверены, DLQ под мониторингом и ретеншеном из семнадцатого, а счётчик неудач переживает рестарт — он в базе, а не в памяти процесса. Схема событий живёт в реестре схем, а не в комментарии рядом с типом, а мониторинг из двадцать первого урока — лаг, рост DLQ, длительность обработки — уезжает на дашборд. Каркас же тот же: ключ, порядок, обработка до коммита, DLQ, лаг, контекст. ### Почему два консьюмера, а не один, который делает резерв и статистику сразу? Разные судьбы отказа: инвентарь, застрявший на недоступном складе, не должен останавливать статистику, и наоборот — сломанная формула витрины не мешает резервам. Разные группы дают каждому свой темп, свои ретраи и свой DLQ, а переиспользование события бесплатно. Отдельный потребитель масштабируется отдельно: инстансов аналитики — по нагрузке витрины, инстансов инвентаря — по потоку заказов. ### Сохранится ли порядок событий одного заказа при росте числа инстансов? Да, и это свойство схемы, а не удача: ключ order_id направляет события в одну партицию, а партиция внутри группы обслуживается ровно одним членом — последовательность держится без блокировок. Следить стоит за двумя местами: число инстансов группы не выше числа партиций, иначе лишние простаивают, и балансер продюсера остаётся Hash — замена на round-robin разорвёт порядок мгновенно. Если порядок нужен строже — на группу в целом, — это уже другая архитектура: одна партиция на весь поток и цена параллелизма, о которой честно предупреждали в пятом уроке.