--- title: "Первый продюсер: Writer" description: "Первый kafka producer на Go: kafka.NewWriter и WriterConfig, WriteMessages с контекстом, AllowAutoTopicCreation, ошибки записи и Close — построчный разбор." course: "Kafka в Go: продюсеры и консьюмеры на kafka-go" url: "https://golangmaster.ru/courses/kafka-go/lessons/04-pervyj-producer/" published: "2026-09-19" updated: "2026-09-29" minutes: 18 --- # Первый продюсер: Writer *Курс «Kafka в Go: продюсеры и консьюмеры на kafka-go», урок 4 из 25. ~18 минут. Обновлено 29.09.2026. Источник: https://golangmaster.ru/courses/kafka-go/lessons/04-pervyj-producer/* Первый kafka producer на Go собирается из двух частей: kafka.NewWriter с конфигом брокера и вызова WriteMessages, который отдаёт сообщение в топик и ждёт подтверждения. В этом уроке пишем продюсера целиком: WriterConfig построчно, механика записи между Go-кодом и диском брокера, обработка ошибок и таймаутов, роль Close и типовые грабли первого дня. Брокер из прошлого урока крутится на localhost:9092, топик orders с тремя партициями создан — самое время отправить в него первое событие из Go. Продюсер в kafka-go называется Writer, и это не случайное имя: клиент ведёт себя как надёжный писатель — копит сообщения, дописывает их в лог брокера и возвращает управление только после подтверждения записи. На первый взгляд всё просто: конструктор, вызов, готово. Но именно здесь живут самые дорогие ошибки новичка: кто-то создаёт Writer на каждое сообщение и тонет в соединениях, кто-то игнорирует возвращённую ошибку и теряет события молча, кто-то не ставит таймаут и зависает на деплое против упавшего брокера. Разберём каждую часть честно: от WriterConfig до судьбы ошибки при недоступном брокере, чтобы продюсер выглядел одинаково уверенно и в учебном примере, и в боевом сервисе. К концу у вас будет каркас, который можно копировать в сервис и не краснеть на code review. ### kafka.NewWriter: WriterConfig и подключение к брокеру Writer создаётся функцией kafka.NewWriter, которой передаётся kafka.WriterConfig — и вот первый сюрприз: NewWriter ничего не подключает. Конструктор лишь запоминает настройки; соединения с брокерами появятся при первой записи, а Writer будет поддерживать их пул дальше. Сердце конфига — Brokers, список адресов брокеров, и Topic — топик по умолчанию, куда уйдут все сообщения этого Writer. Поле Balancer решает, как сообщения распределяются по партициям топика; если его не задать, работает RoundRobin — по кругу. Батчевые настройки BatchSize и BatchTimeout влияют на производительность и подробно разобраны в десятом уроке, здесь оставим их по умолчанию. Важно правило проектирования: Writer — долгоживущий объект уровня приложения. Он потокобезопасен, держит внутреннюю очередь и пул соединений, и создавать его на каждое сообщение — самый известный антипаттерн kafka producer на golang: каждый конструктор — это новые соединения и новое ожидание, а выигрыша ноль. Собрали один Writer в main, передали в обработчики — и живёте с ним весь процесс. Список адресов в Brokers нужен не для красоты: Writer узнаёт у первого живого брокера метаданные кластера — где какой лидер партиций — и дальше общается уже напрямую с ними. Если захочется альтернативного синтаксиса — есть литерал kafka.Writer с полем Addr: библиотека поддерживает оба пути, курс держится классического NewWriter c WriterConfig. ### Продюсер целиком: от NewWriter до Close ```go package main import ( "context" "fmt" "log" "time" "github.com/segmentio/kafka-go" ) func main() { writer := kafka.NewWriter(kafka.WriterConfig{ Brokers: []string{"localhost:9092"}, Topic: "orders", }) defer writer.Close() ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() err := writer.WriteMessages(ctx, kafka.Message{ Key: []byte("order-42"), Value: []byte("created"), }) if err != nil { log.Fatal("send:", err) } fmt.Println("сообщение записано") } ``` Каркас боевого продюсера: NewWriter с адресом брокера и топиком, defer Close на выходе, контекст с таймаутом десять секунд — страховка против зависания на недоступном брокере. WriteMessages принимает kafka.Message: Value — полезная нагрузка в байтах, Key — байтовый ключ маршрутизации, который выбирает партицию и которому посвящён следующий урок. Ошибка после WriteMessages — единственный сигнал о судьбе сообщения, поэтому проверяется сразу, а не по желанию. ### Что происходит между WriteMessages и диском брокера Внутри WriteMessages скрыта конвейерная работа, и о ней полезно знать до первого инцидента. Вызов складывает сообщение во внутреннюю очередь Writer, откуда отдельные горутины собирают их в батчи по адресату — топик и партиция определяются балансировщиком. Батч уезжает по сети на лидера партиции, тот дописывает данные в свой лог и отвечает подтверждением; уровень ожидания задаётся RequiredAcks: конструктор NewWriter при нулевом значении подставляет kafka.RequireAll — запись считается состоявшейся, когда её подтвердили все реплики из ISR; на локальном брокере с одной репликой это просто подтверждение лидера, а сниженный порог kafka.RequireOne и его цена — тема урока про durability. Только после ответа брокера WriteMessages возвращает управление — поэтому «отправил» в этом коде означает «брокер подтвердил запись», а не «байты ушли в сокет». Для вызывающего кода это удобный контракт: при успехе сообщение уже в логе, и о нём можно забыть до чтения. Обратная сторона — запись синхронная по умолчанию: редкие события это устраивает, а для высоких скоростей есть асинхронный режим и батчи — оба разберём в десятом уроке. Кстати, именно поэтому продюсер здесь называется Writer: библиотека подчёркивает метафору дописывания в журнал — вы не шлёте письмо в никуда, а вносите запись в лог, который переживёт и ваш процесс. ### WriteMessages: как kafka producer отправляет сообщения Сигнатура WriteMessages принимает контекст и переменный список сообщений: WriteMessages(ctx, msg) для одиночной отправки и WriteMessages(ctx, msgs...) для пачки, собранной в срез. Сообщение — это kafka.Message с полями Key и Value типа []byte: Kafka не знает про JSON, protobuf и ваши структуры, она хранит байты, и сериализация — тема одиннадцатого урока. Key внешне похож на опцию, но на самом деле это маршрутизатор: одинаковый ключ всегда попадает в одну партицию, что и создаёт порядок событий по сущности — заказу, аккаунту, пользователю. Если WriterConfig.Topic пуст, топик можно указать в каждом сообщении через поле Message.Topic — так один Writer обслуживает несколько каналов. Пачка за один вызов экономит не только строки кода, но и сетевые поездки: одно сообщение — один produce-запрос, тысяча — один-два в зависимости от размера. Наконец, WriteMessages безопасно звать из нескольких горутин одновременно: очередь Writer сериализует работу, и мьютексы в вызывающем коде не нужны. Отдельно про размер: Value не обязан быть маленьким, но у брокера есть потолок на сообщение — по умолчанию около мегабайта, поэтому большие файлы в топик не таскают: их место в объектном хранилище, а в сообщении — ссылка. > **AllowAutoTopicCreation: удобно в dev, дорого в проде.** Поле WriterConfig.AllowAutoTopicCreation со значением true просит брокер создать топик при первой записи. Для локальных экспериментов — золотая кнопка: не нужно ходить в CreateTopics. В продакшене она же превращает опечатку в имя топика в бесшумное создание пустого канала: событие уходит в orders-typo, топик создаётся, никто ничего не замечает, пока аналитика не спросит, куда делись данные. Ставьте true только в dev-окружении и только осознанно. Проверенный приём: в CI-окружении флаг выключен, чтобы тесты падали на опечатках громко и рано. ### Меню: пачка сообщений одним вызовом ```go messages := make([]kafka.Message, 0, len(orders)) for _, o := range orders { messages = append(messages, kafka.Message{ Key: []byte(o.ID), Value: payload(o), }) } err := writer.WriteMessages(ctx, messages...) var writeErr kafka.WriteError if errors.As(err, &writeErr) { log.Printf("не улетело %s: %v", writeErr.Message.Key, writeErr.Err) } ``` Пачка собирается как обычный срез kafka.Message, после чего WriteMessages получает его через многоточие — один вызов вместо сотни. Если при отправке пачки что-то не записалось, ошибка приходит обёрнутой в kafka.WriteError, в поле Message которой лежит проблемное сообщение: удобно логировать именно его ключ, а не гадать по индексу. Про пачки, пороги и BatchTimeout — подробно в десятом уроке. ### Ошибки и таймауты: что возвращает WriteMessages Честная обработка ошибок начинается с понимания, откуда они берутся. Первая группа — недоступный брокер: Writer не сдаётся после первой попытки, а ретраит запись с нарастающим интервалом, пока жив контекст; если брокер так и не поднялся, вызов завершается ошибкой контекста — context.DeadlineExceeded при таймауте или context.Canceled при отмене. Отсюда практическое правило: контекст на запись — не формальность, а ваш дедлайн; без него продюсер может крутиться в ретраях неограниченно долго. Вторая группа — ошибки брокера на корректной связи: запись в несуществующий топик без AllowAutoTopicCreation прилетает ошибкой Unknown Topic Or Partition, отказ в доступе к каналу — своим кодом. Все они приходят обёрнутыми, и вытащить код умеет errors.As с целью kafka.Error: значение этого типа само и есть код, его сравнивают с константами вроде kafka.UnknownTopicOrPartition. Третий источник — само сообщение: превышение лимита размера (BatchBytes в WriterConfig и потолок брокера) возвращает MessageTooLargeError с индексом проблемного сообщения — его выбрасывают из пачки и отправляют отдельно, а не глушат всю пачку целиком. Каждой группе — свой ретрай и свой алерт: это тема двенадцатого урока, здесь важно видеть карту. Хорошая новость: все три группы ловятся в одном месте — ошибкой, возвращённой из WriteMessages, и различаются уже по кодам, поэтому единый обработчик с логированием и метрикой покрывает весь спектр. > **Проглотили ошибку — потеряли событие.** Живое место из аудита реального сервиса: err := writer.WriteMessages(...) с подчёркиванием вместо обработки. Всё работало, пока во время деплоя не перезапустился брокер: две минуты писали в ничто, ошибки тонули в логе, а заказы числились отправленными. Итог — сверка с биллингом и ручной докат событий. Kafka не бросает исключений: единственный сигнал о неудаче — возвращённая ошибка, и контракт простой: не проверил ошибку — согласился на потерю. Симптом у всех одинаковый: графики сообщений ровные, а внизу по течению дыра в данных, и найти её начало помогает только сверка с источником. ### Close и жизнь Writer в сервисе writer.Close() в продюсере — не церемония, а часть контракта. Close дожидается отправки накопленных батчей, завершает внутренние горутины и закрывает пул соединений: без него последние сообщения могут остаться в памяти процесса при завершении, а «молчаливая потеря» — худший из видов отказа для шины событий. Канонический порядок такой: Writer создаётся один на процесс, defer Close ставится рядом с конструктором, а при остановке сервиса закрывается явно после того, как обработчики закончили работу — приёмы выключения без потерь собраны в [уроке про грабли продакшена](/courses/kafka-go/lessons/24-grabli-prod/). Обратная сторона долгожизни — конфигурация: Writer настраивается один раз и не пересобирается на ходу, поэтому всё, что может меняться, — топики в сообщениях, ключи и значения; сам Writer и его конфиг остаются неизменными. Для тестов Writer инжектится интерфейсом, как любая зависимость, — приём из урока про тесты Kafka. Правило на память: один Writer, длинная жизнь, аккуратный Close. И держите конфиг рядом с кодом запуска: полдня дебага уходит на то, чтобы найти, где именно продюсер настроен иначе, чем вы думали. ### Анатомия записи одним столом | Элемент | Роль в записи | | --- | --- | | WriterConfig.Brokers | Адреса брокеров для начального подключения и метаданных | | WriterConfig.Topic | Топик по умолчанию; пустое значение — топик в каждом сообщении | | WriterConfig.Balancer | Правило выбора партиции; по умолчанию RoundRobin | | Message.Key | Ключ маршрутизации: одинаковый ключ — одна партиция | | Message.Value | Полезная нагрузка в байтах | | WriteMessages(ctx, ...) | Запись с ожиданием подтверждения брокера до смерти контекста | | writer.Close() | Флеш накопленных батчей и закрытие соединений | ### Чек-лист продюсера одним списком - Один Writer на процесс: конструктор — в main, соединения — по мере необходимости. - Контекст с таймаутом на каждую WriteMessages: дедлайн — страховка от вечных ретраев. - Ошибка после записи проверяется всегда: молчаливое подчёркивание — согласие на потерю. - Ключ сообщения выбирается в домене: он определяет партицию и порядок — тема следующего урока. - Close — один раз при завершении: флеш батчей и закрытие соединений. - AllowAutoTopicCreation — только в dev: в проде опечатка должна быть ошибкой, а не новым топиком. ### Проверка: когда NewWriter подключается **Вопрос:** После kafka.NewWriter(...) соединений с брокером ещё нет. Когда Writer впервые позвонит брокеру? - ✅ При первом WriteMessages — конструктор только запоминает конфиг - Сразу внутри NewWriter, в фоне - Только при вызове Close - Никогда: Writer пишет в локальный буфер на диске NewWriter ленив: он настраивает объект, а пул соединений возникает при первой записи. Проверить живость брокера через NewWriter нельзя — для этого есть kafka.Client из Admin API и его Metadata-запрос, тот же, каким CreateTopics ищет кластер; поэтому «создал и ничего не упало» ещё не значит «подключился». ### Проверка: запись в несуществующий топик **Вопрос:** Топика orders нет, AllowAutoTopicCreation не задан. Что вернёт WriteMessages? - Ничего: сообщение отправится в очередь на создание топика - ✅ Ошибку Unknown Topic Or Partition - Панику внутри библиотеки - Успех, но сообщение потеряется при рестарте брокера По умолчанию Kafka защищает от опечаток: запись в несуществующий топик возвращает ошибку Unknown Topic Or Partition, и WriteMessages честно её отдаёт. Автосоздание включается явно полем AllowAutoTopicCreation — и тогда опечатка создаёт новый пустой топик вместо ошибки, что в продакшене хуже ошибки. ### Что напечатает программа? ```go writer := kafka.NewWriter(kafka.WriterConfig{ Brokers: []string{"localhost:9092"}, Topic: "orders", }) defer writer.Close() // брокер остановлен: docker compose stop kafka ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() err := writer.WriteMessages(ctx, kafka.Message{ Key: []byte("order-1"), Value: []byte("created"), }) fmt.Println(err) ``` **Что выведет программа?** - — Writer буферизует сообщение и дождётся брокера при Close - Панику: соединение отклонено - ✅ context deadline exceeded — Writer ретраил до истечения контекста - Ошибка соединения мгновенно: Writer не делает повторных попыток Writer ретраит запись с нарастающим интервалом, пока жив контекст: три секунды попыток, затем возвращается context.DeadlineExceeded. Это поведение и есть причина ставить таймаут: без него горутина записи может ждать недоступного брокера бесконечно. Сообщение при этом не записано — после ошибки его судьбу решает ваш код. ### Соберите первый продюсер Допишите топик, конструктор сообщения и вызов записи. ```go writer := kafka.NewWriter(kafka.WriterConfig{ Brokers: []string{"localhost:9092"}, Topic: "___", }) defer writer.___() msg := kafka.___{ Key: []byte("order-42"), Value: []byte("created"), } err := writer.___(ctx, msg) ``` Подсказка: Топик из этого курса — orders, закрывает Writer метод Close, сообщение — kafka.Message, записывает — WriteMessages. Правильные ответы в пропусках: `orders`, `Close`, `Message`, `WriteMessages` ### Соберите продюсера по строкам Расставьте строки в порядке работы программы: Writer, его закрытие, контекст, сообщение, отправка, обработка ошибки, вывод. Правильный порядок строк: 1. `writer := kafka.NewWriter(kafka.WriterConfig{Brokers: []string{"localhost:9092"}, Topic: "orders"})` 2. `defer writer.Close()` 3. `ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)` 4. `defer cancel()` 5. `msg := kafka.Message{Key: []byte("order-42"), Value: []byte("created")}` 6. `err := writer.WriteMessages(ctx, msg)` 7. `if err != nil { log.Fatal("send:", err) }` 8. `fmt.Println("сообщение записано")` ### Карточки: первый продюсер - **kafka.NewWriter** — Конструктор Writer по WriterConfig. Ленив: не подключается, соединения появляются при первой записи. Создаётся один раз на процесс. - **WriterConfig** — Конфиг записи: Brokers, Topic, Balancer, батчевые пороги, RequiredAcks, AllowAutoTopicCreation. После создания не меняется. - **WriteMessages** — Запись одного сообщения или пачки с ожиданием подтверждения брокера. Ошибка — единственный сигнал о неудаче, проверять обязательно. - **kafka.Message** — Сообщение: Key и Value типа []byte, опциональные Topic и Time. Key маршрутизирует по партициям, Value — нагрузка. - **kafka.WriteError** — Обёртка ошибки при пачечной записи: в поле Message лежит конкретное сообщение, которое не записалось. - **writer.Close()** — Дожидается отправки накопленных батчей, глушит внутренние горутины и закрывает соединения. Без него — тихая потеря хвоста. Продюсер готов: один Writer на процесс, запись через WriteMessages с контекстом, ошибки обрабатываются, Close на месте. Дальше по курсу: в [уроке про ключи и партиционирование](/courses/kafka-go/lessons/05-klyuchi-particionirovanie/) разберём, как Hash-балансировщик собирает порядок по сущности и что ломается при увеличении партиций; в [уроке про батчи](/courses/kafka-go/lessons/10-producer-batchi/) выжмем из Writer скорость; про консьюмера — всё с [шестого урока](/courses/kafka-go/lessons/06-pervyj-consumer/). Код урока запускается против вашего локального брокера из третьего урока: скопируйте main.go, запустите и загляните в топик консольным консьюмером — первое сообщение в логе поставит жирную точку над теорией. Песочница сеть не эмулирует, зато интерактивы ниже тренируют сам каркас записи. После урока вернитесь к домашнему заданию: отправить пачку, сломать брокера и вернуть его — лучший способ выучить поведение Writer до первого продакшен-инцидента. ## Коротко: что запомнить - Writer — долгоживущий объект приложения: NewWriter ленив, соединения появляются при первой записи. - WriteMessages блокируется до подтверждения брокера; контекст задаёт дедлайн всей операции. - Message — это Key и Value в байтах: Key маршрутизирует по партициям, Value — полезная нагрузка. - Недоступный брокер — ретраи до смерти контекста и context.DeadlineExceeded; несуществующий топик — Unknown Topic Or Partition. - Пачка сообщений одним вызовом экономит сетевые поездки; ошибка приходит в kafka.WriteError с проблемным сообщением. - Close флешит накопленные батчи: без него хвост событий тихо умирает вместе с процессом. - Контекст — не формальность: он ограничивает ретраи Writer и возвращает DeadlineExceeded, когда брокер так и не ответил. - AllowAutoTopicCreation — инструмент dev-окружения: в проде опечатка вместо ошибки создаёт пустой топик. ## Домашнее задание Расширь продюсера из урока: отправьте в orders десять сообщений с ключами order-1..order-10 одной пачкой через WriteMessages(ctx, msgs...). Затем остановите брокер командой docker compose stop kafka и запустите отправку с таймаутом три секунды — убедитесь, что получаете context deadline exceeded, а не панику. Верните брокера через docker compose start kafka, перечитайте топик консольным консьюмером и ответьте себе: какие сообщения доехали, а какие надо отправить повторно. ## Вопросы и ответы по уроку ### Нужно ли закрывать Writer после каждой отправки? Нет — наоборот: Writer создаётся один на процесс и живёт вместе с сервисом. Close вызывается один раз при завершении работы, чтобы доехали накопленные батчи. Конструктор на каждое сообщение — антипаттерн: он плодит соединения и не даёт Writer'у собрать пачку. Если сервисов много, выносите создание Writer в конструктор приложения и передавайте его зависимостью. ### WriteMessages блокирующий — не повредит ли это моему HTTP-обработчику? По умолчанию да, вызов ждёт подтверждения брокера, поэтому в обработчиках его вызывают с контекстом с таймаутом, а при высоких скоростях уходят в асинхронный режим Writer. Для типовых нагрузок синхронная запись со сдержанным таймаутом — самый безопасный старт; ускорение — отдельная тема десятого урока. ### Как отправлять сообщения сразу в несколько топиков? Варианта два: поле Topic внутри каждого kafka.Message при пустом WriterConfig.Topic, либо отдельный Writer на топик. Первый экономит объекты и соединения, второй изолирует настройки вроде балансировщика и батчей. В обоих случаях топик обязан существовать — либо с включённым AllowAutoTopicCreation в dev. Для пачки в несколько топиков первый вариант и есть штатный путь.