--- title: "Сериализация: JSON в сообщениях" description: "Сериализация сообщений Kafka на Go: json.Marshal в Value, теги struct, общий пакет контрактов для продюсера и консьюмера, Headers и ошибки декодирования." course: "Kafka в Go: продюсеры и консьюмеры на kafka-go" url: "https://golangmaster.ru/courses/kafka-go/lessons/11-json-soobshcheniya/" published: "2026-09-19" updated: "2026-09-29" minutes: 18 --- # Сериализация: JSON в сообщениях *Курс «Kafka в Go: продюсеры и консьюмеры на kafka-go», урок 11 из 25. ~18 минут. Обновлено 29.09.2026. Источник: https://golangmaster.ru/courses/kafka-go/lessons/11-json-soobshcheniya/* В связке kafka json golang всё держится на двух функциях стандартной библиотеки: json.Marshal у продюсера и json.Unmarshal у консьюмера, потому что Value в kafka.Message — это просто срез байтов. Разберём сериализацию сообщений как контракт: теги struct, общий пакет, поведение при расхождении схем и цену формата. Продюсер из четвёртого урока отправлял то, что ему положили: []byte("charge-100") — и этого хватало, пока сообщений было немного и форма была очевидна. Как только событие обрастает полями — идентификатор заказа, аккаунт, сумма, время, — возникает вопрос, на который Kafka сама не отвечает: в каком виде байты лежат в партиции. Брокеру всё равно, для него Value — непрозрачный срез; соглашение о форме — дело продюсера и консьюмера, и это соглашение называется сериализацией сообщений. В Go-стеке самый частый ответ — JSON из стандартной библиотеки: json.Marshal собирает структуру в байты перед отправкой, json.Unmarshal читает их на другом конце. Формат выбирается один раз и живёт годами: смена формы Value — это миграция всех читателей топика, а не правка одного поля. Поэтому разберём не «как вызвать Marshal» — это одна строка, — а то, что вокруг неё: теги, общий пакет контрактов, поведение при расхождении схем и точки, где JSON начинает вредить. ### kafka json golang: Value в kafka.Message это []byte Вернёмся к анатомии сообщения из второго урока с новой стороны. kafka.Message несёт несколько полей, и для сериализации значимы три: Key — маршрут по партициям, Value — полезная нагрузка, Headers — пары ключ-значение для метаданных. Поле Value объявлено как []byte, и это не упрощение kafka-go, а честная позиция брокера: Kafka не знает про ваши домены и не обязана знать. Связка «kafka json golang» устроена поэтому тривиально: json.Marshal(order) возвращает байты, они кладутся в Value, консьюмер вызывает json.Unmarshal(m.Value, &event) — и событие снова структура. Сериализационного слоя в kafka-go нет, и это плюс: библиотека не тянет ни protobuf-рантайм, ни avro-схемы, а решение о формате остаётся за вами и меняется без вопросов к клиенту. Заметим и цену: JSON избыточен — имена полей повторяются в каждом сообщении, числа без схемы гадают типы, а деньги во float64 живут с погрешностью. Для сервисных потоков это допустимо, для больших объёмов — нет; альтернативы разберём ниже, а пока зафиксируем правило: то, что вы положили в Value, интерпретирует только ваш же код, и чем раньше оба конца согласуют форму, тем дешевле каждое следующее изменение. ### Общий пакет контрактов: форма в одном месте ```go // events/events.go — пакет импортируют продюсер и консьюмер package events import "encoding/json" type OrderPaid struct { OrderID string `json:"order_id"` AccountID string `json:"account_id"` AmountCents int64 `json:"amount_cents"` } func (e OrderPaid) Marshal() ([]byte, error) { return json.Marshal(e) } func ParseOrderPaid(value []byte) (OrderPaid, error) { var e OrderPaid err := json.Unmarshal(value, &e) return e, err } ``` Пакет events — единственное место, где живёт форма события: структура, теги и две обёртки над Marshal и Unmarshal. Продюсер вызывает OrderPaid.Marshal и кладёт байты в Value, консьюмер — ParseOrderPaid на выходе ReadMessage. Деньги сразу в целых копейках: amount_cents int64 не страдает погрешностью float64. Теги задают нижний регистр с подчёркиваниями — форму, которую прочитает и соседний сервис на Python, и отладочный скрипт. > **Живое место: поле переименовали в двух сервисах по-разному.** Классика расхождения схем: команда продюсера переименовала Amount в TotalAmount, а общий пакет не обновила — сервисы в разных репозиториях. Консьюмер продолжил декодировать старый ключ, и JSON не ругнулся: Unmarshal молча оставил Amount нулевым. Биллинг неделю писал нулевые суммы, пока бухгалтерия не заметила; никакой ошибки на линии не было — контракт просто разъехался. Лечение дисциплиной: форма события живёт в одном пакете, изменения проходят через ревью, а на стыках стоят тесты, которые маршаллят свежим продюсером и декодируют свежим консьюмером. Разные репозитории — не оправдание: пакет выносится в отдельный модуль или генерируется из схемы, и тогда «переименовать поле» превращается в релиз с миграцией. Тема версионирования схем — двадцать второй урок курса, а механика тегов — следующая. ### Теги json: поведение, о котором легко забыть Тег json — это спецификация формы, и у него есть поведение, которое лучше знать до инцидента. Без тега encoding/json использует имя поля как есть: OrderID превратится в «OrderID» с заглавной буквы, а консьюмер на другом языке ждёт order_id — поэтому теги пишут на обеих сторонах, даже если кажется, что и так совпадёт. omitempty пропускает пустые значения при маршаллинге, но ноль — легитимное значение для суммы, скидки или счётчика: если ноль нужно отличать от отсутствия, берите указатели, а не флаг. Неэкспортированные поля не сериализуются вовсе: маленькая буква в имени поля — и поле исчезло из сообщения без предупреждений. При декодировании неизвестные поля игнорируются — это фундамент обратной совместимости: продюсер добавляет поля, и старый консьюмер не падает. Но игнорирование работает в обе стороны: опечатка в теге не вызовет ошибки ни у кого, события просто потеряют поле. Самый дешёвый тест контракта — unit-тест, который маршалит и тут же декодирует структуру и сверяет все поля: он же документирует событие лучше любого комментария. ### Проверка: новое поле и старый консьюмер **Вопрос:** Продюсер добавил в событие новое поле, консьюмер декодирует сообщение старой структурой. Что произойдёт? - ✅ Поле молча игнорируется: Unmarshal пропускает неизвестные ключи - Unmarshal вернёт ошибку: схемы не совпали - Консьюмер упадёт с паникой на неизвестном поле - Поле декодируется в ближайшее по имени поле структуры encoding/json при декодировании пропускает ключи, для которых нет поля в структуре: обратная совместимость на добавление полей встроена в стандартную библиотеку. Ошибкой это не является — поэтому расхождения схем ловятся тестами и мониторингом, а не паниками. Удалять и переименовывать поля так же безболезненно нельзя. ### Сериализация сообщений как релизный процесс Идея общего пакета проста, а экономия огромна: структуры событий живут в одном пакете — events, contracts, как ни назовите, — и оба сервиса импортируют его. Один процесс маршалит, второй декодирует, форма существует в единственном экземпляре, и правка проходит одним коммитом через оба сервиса, если они в одном репозитории. Внутри монорепозитория это самая дешёвая схема из возможных: компилятор сам проверяет использование полей, а rename-рефакторинг переименовывает и продюсер, и консьюмер за один заход. С отдельными репозиториями честнее сразу признать: контракт — внешний интерфейс, его изменение — релиз с миграцией, а не правка. Вот где встаёт поле Headers из kafka.Message: тип события и версия схемы едут в заголовках, Value декодируется по ним, и два формата спокойно сосуществуют в одном топике. Полноценный schema registry kafka-go не предоставляет — при потребности подключают внешний реестр или protobuf; но для большинства сервисных потоков общий пакет закрывает задачу целиком, а заголовки дают место для развилки на время перехода. ### Продюсер и консьюмер на одном пакете ```go // продюсер: структура в байты одним вызовом event := events.OrderPaid{OrderID: order.ID, AccountID: acc.ID, AmountCents: amount} value, err := event.Marshal() if err != nil { log.Printf("marshal order %s: %v", order.ID, err) return } err = writer.WriteMessages(ctx, kafka.Message{ Key: []byte(order.ID), Value: value, }) // консьюмер: байты обратно в структуру m, err := reader.ReadMessage(ctx) if err != nil { log.Printf("read: %v", err) break } paid, err := events.ParseOrderPaid(m.Value) if err != nil { log.Printf("broken message %s/%d/%d: %v", m.Topic, m.Partition, m.Offset, err) continue } process(paid) ``` Обе стороны используют один пакет events: продюсер маршалит структуру и кладёт байты в Value рядом с ключом заказа, консьюмер декодирует Value той же формой. Ошибка маршаллинга не убивает поток — событие логируется и пропускается. Ошибка декодирования у консьюмера — признак битого или чужого сообщения: адрес топик-партиция-оффсет в логе позволяет найти виновника вручную, а обработка продолжается со следующего события. ### Правила сериализации одним списком - Value — срез байтов: форма содержимого — договорённость продюсера и консьюмера, брокер её не знает. - Форма события живёт в одном пакете: Marshal и Unmarshal обеих сторон собираются из одной структуры. - Теги пишутся всегда: без тега поле уезжает с заглавной буквы и не читается снаружи. - Ноль и отсутствие — разные вещи: нужны оба — берите указатели, а не omitempty. - Неизвестные поля при декодировании игнорируются: добавление поля безопасно, переименование — миграция. - Деньги — в целых копейках int64: float64 рано или поздно даст 19.990000000000002. - Тип и версия события — в Headers: Value декодируется по заголовку, а не по угадыванию. > **Headers: метаданные мимо Value.** Поле Headers в kafka.Message — срез пар Key string и Value []byte, который едет рядом с телом и не смешивается с ним. Привычный ход — тип события и версия схемы в заголовках: читатель может посмотреть заголовок, не декодируя тело, маршрутизатор разрулит разные форматы, а миграция схем получит место для развилки. Заголовки не сериализуются в Value и не требуют отдельной обработки на брокере. Но это не schema registry и не шифрование: они валидны ровно до тех пор, пока им верит продюсер. Как место для служебных метаданных, которые нужны до декодирования, — самое то; для проверки форм и правды о схеме всё равно нужны тесты из общего пакета. ### Проверка: обрезанное сообщение **Вопрос:** В Value оказалось обрезанное сообщение — сеть порезала байты посреди JSON. Что вернёт json.Unmarshal? - ✅ Ошибку *json.SyntaxError: байты не образуют корректный JSON - nil: Unmarshal прочитает начало и молча остановится - Панику: Unmarshal не переживает обрезанные данные - Ошибку только если обрезка попала в середину строки Некорректный JSON даёт *json.SyntaxError на первом же нарушении грамматики. Отсюда правило потребителя: ветка err после Unmarshal обязательна, а сообщение с синтаксической ошибкой — кандидат в яд-сообщения, о которых подробнее говорит девятнадцатый урок курса. ### Что вернёт Unmarshal при несовпадении типов? ```go type Produced struct { OrderID string `json:"order_id"` Amount float64 `json:"amount"` } type Consumed struct { OrderID string `json:"order_id"` Amount int `json:"amount"` } // в Value лежит: {"order_id":"o-1","amount":19.99} var consumed Consumed err := json.Unmarshal(value, &consumed) fmt.Println(consumed.OrderID, consumed.Amount, err != nil) ``` **Что выведет программа?** - ✅ o-1 0 true: ошибка UnmarshalTypeError, остальные поля декодированы - o-1 19.99 false: дробное обрежется до int без ошибок - ошибка, и OrderID тоже останется пустым - паника на конвертации float64 в int json.Unmarshal фиксирует несовместимые типы как *json.UnmarshalTypeError, но остальные поля декодирует: OrderID заполнится, Amount останется нулевым, а err будет не nil. Несовместимый тип одного поля не роняет всё событие — и это повод не игнорировать err: работать с частично заполненной структурой опаснее, чем обработать ошибку. ### Когда JSON — не тот выбор JSON остаётся дефолтом, потому что читается человеком, дебажится консольными утилитами и не тянет зависимостей, — но у него есть пределы, и лучше узнать о них до внедрения. Первый — размер: имена полей повторяются в каждом сообщении, и на потоках в десятки тысяч событий в секунду вы платите трафиком и диском за повторение order_id миллион раз. Второй — скорость: рефлексия encoding/json не бесплатна, а Writer копит батчи по байтовому лимиту из десятого урока — чем меньше сообщение, тем больше событий уезжает в одном produce-запросе. Третий — типы: у JSON нет десятичных дробей, время живёт строкой или числом, и каждая такая вещь требует конвенции. Альтернативы старые как сам Kafka: protobuf даёт компактность и строгие типы за счёт генерации кода из схемы, avro — эволюцию схем с реестром, msgpack — компромисс по размеру без отдельной схемы. kafka-go честно равнодушен: Value останется []byte при любом выборе, менять придётся только места Marshal и Unmarshal — ещё один аргумент за общий пакет из прошлого раздела. Начинайте с JSON, переходите тогда, когда цифры, а не мода, скажут, что формат стал узким местом. ### Форматы Value одним столом | Формат | Чем сериализуется | Когда брать | | --- | --- | --- | | JSON, encoding/json | Стандартная библиотека: Marshal и Unmarshal | Сервисные потоки, отладка, старт проекта | | protobuf | Генерация кода из .proto, строгие типы | Большие объёмы и межъязыковые контракты | | avro | Схемы с реестром и эволюцией версий | Долгоживущие потоки с изменяемой схемой | | []byte без формата | Свои кодеки поверх среза байтов | Внутренние потоки с готовым бинарным payload | ### Яд-сообщения: когда сериализация ломает чтение Худший сценарий сериализации происходит не на продюсере, а на консьюмере: одно сообщение, которое не декодируется, — и группа стоит. Механика проста: ReadMessage возвращает событие, Unmarshal падает, код логирует ошибку и делает continue — но оффсет ещё не закоммичен, и после рестарта группа начнёт с того же места. Топик зацикливается на битом сообщении, лаг растёт, дежурный смотрит в лог с тысячей одинаковых строк. Отсюда правило: ветка ошибки декодирования — это решение о судьбе события, а не строчка лога. Варианты — пропустить с ручным коммитом оффсета, отправить в карантинный топик или ждать фикса с паузой; развёрнутый разбор с DLQ-паттерном — в девятнадцатом уроке. json.Valid позволяет отсечь грубую порчу до Unmarshal, а залогированные топик, партиция и оффсет — найти сообщение консольной утилитой и понять, что в нём лежит. В песочнице это не воспроизвести: нужен локальный брокер из Docker-стенда третьего урока — там и накатайте сценарий с битым Value. ### Соберите обёртки общего пакета Допишите тип возврата маршаллинга, имя ключа в теге и функцию декодирования. ```go type OrderPaid struct { OrderID string `json:"___"` } func Marshal(e OrderPaid) (___, error) { return json.Marshal(e) } func Parse(value []byte) (OrderPaid, error) { var e OrderPaid err := json.___(value, &e) return e, err } ``` Подсказка: Ключ в нижнем регистре через подчёркивание, Marshal возвращает срез байтов, а байты обратно в структуру собирает json.Unmarshal. Правильные ответы в пропусках: `order_id`, `[]byte`, `Unmarshal` ### Соберите круговорот события Расставьте строки полного цикла: продюсер маршалит и отправляет, консьюмер читает, декодирует и обрабатывает — с ветками ошибок на своих местах. Правильный порядок строк: 1. `value, err := event.Marshal()` 2. `if err != nil { log.Printf("marshal: %v", err); return }` 3. `err = writer.WriteMessages(ctx, kafka.Message{Key: key, Value: value})` 4. `m, err := reader.ReadMessage(ctx)` 5. `if err != nil { log.Printf("read: %v", err); break }` 6. `paid, err := events.ParseOrderPaid(m.Value)` 7. `if err != nil { log.Printf("broken %d: %v", m.Offset, err); continue }` 8. `process(paid)` ### Карточки: сериализация сообщений - **kafka.Message.Value** — Срез []byte: брокер не знает его форму; JSON, protobuf или свой формат — выбор продюсера и консьюмера. - **Общий пакет контрактов** — Пакет со структурами событий, который импортируют обе стороны: Marshal и Unmarshal собираются из одной формы. - **Тег json** — Задаёт имя поля в JSON; без тега поле уезжает с заглавной буквы. omitempty пропускает пустые значения. - **Неизвестные поля** — Unmarshal пропускает их без ошибки: добавление поля обратно совместимо, переименование и смена типа — миграция. - **kafka.Message.Headers** — Срез пар Key string и Value []byte для метаданных: тип и версия события едут мимо тела сообщения. - **Яд-сообщение** — Сообщение, которое не декодируется: без ветки решения группа зациклится на нём после рестарта. Форма Value теперь под контролем: структуры в общем пакете, теги выписаны, декодирование с веткой ошибки, а формат выбран осознанно, а не по привычке. Дальше по курсу: [шестой урок про первого консьюмера](/courses/kafka-go/lessons/06-pervyj-consumer/) напомнит, как устроен цикл чтения, в который встаёт ParseOrderPaid; [девятнадцатый урок про ошибки консьюмера](/courses/kafka-go/lessons/19-oshibki-consumer/) доведёт яд-сообщения до карантинного топика; [двадцать второй урок про эволюцию схем](/courses/kafka-go/lessons/22-evolyuciya-shem/) покажет, как менять контракт без остановки потока. Формат честности прежний: код урока запускается локально против брокера из Docker, а интерактивы ниже тренируют механику Marshal, Unmarshal и тегов на программах в чистом Go — прямо в песочнице, без сети и брокера. ## Коротко: что запомнить - Value в kafka.Message это []byte: форма содержимого — контракт продюсера и консьюмера, а не забота брокера. - json.Marshal перед отправкой и json.Unmarshal при чтении — минимальная сериализация сообщений в Go. - Форма события живёт в одном пакете: структуры, теги и обёртки — единственный источник правды. - Теги задают имена полей: без тега поле уезжает с заглавной буквы и не читается снаружи. - Неизвестные поля при декодировании игнорируются: добавление поля обратно совместимо, переименование — миграция. - Деньги — в целых копейках int64: float64 накапливает погрешность на каждом сообщении. - Ветка ошибки Unmarshal — решение о судьбе события: пропустить с коммитом, отправить в карантин или ждать фикс. ## Домашнее задание Вынесите структуры событий платёжного сервиса в пакет events: OrderPaid и OrderRefunded с тегами в snake_case и суммой в копейках. Напишите unit-тест контракта: Marshal свежим продюсером, Unmarshal свежим консьюмером, сверка всех полей — плюс тест на сообщение с лишним неизвестным полем, оно должно декодироваться без ошибки. Против брокера из третьего урока отправьте десять событий и убедитесь консьюмером, что формы совпали. Финал: сломайте тег на консьюмерной стороне (order_id в orderid) и опишите по логам, что вы увидите и на каком этапе это поймается. ## Вопросы и ответы по уроку ### Нужно ли сжимать JSON перед отправкой? kafka-go умеет сжатие на уровне продюсера через CompressionCodec в WriterConfig, и JSON сжимается хорошо: повторяющиеся имена полей — его слабость и лучший друг gzip. Кодеки — константы kafka.Gzip, kafka.Snappy, kafka.Lz4 и kafka.Zstd, и какой из них выиграет — предмет замера на вашем payload: форматы выигрывают на разных типах данных. Но сжатие — тема производительности, а не контракта: включайте его, когда измерение покажет, что трафик стоит денег. Для консьюмера разница невидима — брокер отдаёт уже разжатые байты, и формат Value не меняется. ### Почему не взять gob или binary.Marshal вместо JSON? gob привязывает оба конца к Go: утилиты консоли, дашборды и скрипты на других языках останутся за бортом, а однажды появится консьюмер на другом языке — и придётся переписывать. JSON из стандартной библиотеки честнее как общий язык экосистемы, а для скорости и компактности существуют protobuf и avro. Смена формата позже — миграция всех читателей топика, поэтому выбирайте сразу с мыслью о внешних потребителях, даже если их пока нет. ### Как декодировать события разных типов из одного топика? Тип и версия — в Headers: kafka.Message.Headers, срез пар строка-байты, читается без декодирования тела. По заголовку выбираете структуру и вызываете нужный Parse. Второй путь — поле-дискриминатор внутри самого JSON, но оно требует декодировать Value хотя бы в json.RawMessage, чтобы заглянуть. Оба пути работают; заголовки чище, потому что не трогают тело события.