Формат сообщений: JSON и структуры

Соединение установлено, очередь есть, DeliveryMode выбран — и вот первый межсервисный контракт проекта: что именно лежит в срезе байтов. Прошлые уроки гоняли строки, потому что механике брокера без разницы; живым сервисам — нет: потребитель обязан понять, что пришло, и понимать одинаково с издателем, включая версии. Это и есть формат сообщений очереди: соглашение о структуре тела, типе контента и правилах изменения. Выбор кажется мелочью, пока прод не падает после переименования поля. Строка «order.created ord-1042» из двенадцатого урока читается человеком, но требует ручного парсинга, в неё не влезает ни одно новое поле, и по ней не понять, что именно сломалось. JSON со структурами Go закрывает это дешевле всего: стандартная библиотека, читаемость в логах и management UI, привычный тулчейн. Разберём полный цикл — от структуры до Delivery и обратно — и цену этого удобства.

Rabbitmq json: маршалинг и ContentType#

Издательская сторона rabbitmq json — это json.Marshal и ContentType. Структура с тегами описывает контракт: тег переименовывает поле в ключ нижнего регистра и служит документацией в обе стороны — что отдаёт издатель и что ждёт потребитель. Marshal возвращает срез байтов, он и становится Body; ошибку маршалинга обрабатывают до публикации — в очереди она уже не случится. ContentType ставьте application/json явно: amqp091-go не заполняет его за вас, а потребитель, инструменты и будущие версии формата читают именно это поле; рядом живут ContentEncoding и метаданные в Headers, но тип контента — минимальная гигиена. DeliveryMode из девятого урока решается отдельно: JSON-байты персистентны ровно настолько, насколько persistent у Publishing и durable у очереди, — формат на выживание не влияет. Тело стоит держать компактным: очередь — не база документов, и гигантские объекты тормозят брокера и потребителей одинаково.

Разбор на потребителе и ядовитые тела#

Потребитель делает обратное: json.Unmarshal(d.Body, &order) — и здесь формат сообщений очереди встречается с реальностью. Тело может быть вовсе не JSON: битая публикация, чужой сервис в той же очереди, обрезанное сообщение после эксперимента с лимитами. Unmarshal вернёт ошибку — и это самый важный if в воркере: битое тело не станет валидным от повтора, поэтому requeue=true превращает его в ядовитое из восьмого урока — возврат, доставка, снова возврат. Правильная ветка: лог с телом и заголовками, Nack с requeue=false и путь в DLQ из шестнадцатого. Второй класс проблем — валидный JSON с неожиданной схемой: Unmarshal по умолчанию игнорирует неизвестные поля и оставляет нули там, где ожидаемые поля отсутствуют; заказ с нулевой суммой и без идентификатора пройдёт разбор и упрётся в бизнес-логику — значит, после разбора нужна валидация обязательных полей. Хотите строгий режим — Decoder с DisallowUnknownFields, но тогда каждое новое поле издателя ломает старых потребителей; чаще выбирают мягкий разбор плюс явную валидацию.

Издатель: структура, маршалинг, публикация

package main

import (
        "context"
        "encoding/json"
        "log"
        "time"

        amqp "github.com/rabbitmq/amqp091-go"
)

type Order struct {
        ID       string "json:\"id\""
        Amount   int    "json:\"amount\""
        Currency string "json:\"currency\""
        Created  string "json:\"created_at\""
}

func main() {
        conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
        if err != nil {
                log.Fatal(err)
        }
        defer conn.Close()

        ch, err := conn.Channel()
        if err != nil {
                log.Fatal(err)
        }

        _, err = ch.QueueDeclare("orders.json", true, false, false, false, nil)
        if err != nil {
                log.Fatal(err)
        }

        order := Order{
                ID:       "ord-1042",
                Amount:   1990,
                Currency: "RUB",
                Created:  time.Now().UTC().Format(time.RFC3339),
        }
        body, err := json.Marshal(order)
        if err != nil {
                log.Fatal(err)
        }

        ctx := context.Background()
        err = ch.PublishWithContext(ctx, "", "orders.json", false, false,
                amqp.Publishing{
                        ContentType:  "application/json",
                        Body:         body,
                        DeliveryMode: amqp.Persistent,
                })
        if err != nil {
                log.Fatal(err)
        }
        log.Println("опубликовано:", string(body))
}

Тег переименовывает поле в ключ нижнего регистра: ID станет id, Created — created_at. Marshal идёт до публикации, его ошибка — проблема издателя, а не повод для возврата в очереди; ContentType application/json — явное обещание потребителю, что внутри JSON. Created в UTC по RFC3339: время в контрактах хранят в одной зоне и одном формате.

Форматы тела: чем платить#

Формат	Сильные стороны	Цена
JSON	читаемость, отладка в UI, стандартная библиотека, гибкая эволюция	объём и парсинг на больших потоках
protobuf	компактность, схемы с версионированием, кодогенерация	тело не прочитать глазами, нужны proto-файлы и генерация
msgpack	компактнее JSON при схожей модели данных	меньше инструментов в логах и браузере
текстовый протокол	нулевая зависимость, тривиальный разбор	хрупкость: парсинг руками и без валидации

Переименовали поле — потребители получили нули

Кейс: издатель переименовал ключ amount в total, выкатился — и отчёты начали считать заказы на ноль рублей. Ошибок нигде нет: Unmarshal не находит знакомого ключа, оставляет поле нулевым и молчит. Так ломаются схемы без тестов. Второй вариант того же инцидента: тип поля сменили со строки на число — потребители со строкой в структуре получают ошибку разбора на каждом сообщении, очередь полнится необработанным. Лечится правилами эволюции: поля добавляют, не переименовывают и не меняют типы; новое имя означает новый ключ рядом со старым, пока все потребители не переехали. Спорная привычка — полагаться на то, что в монорепе обновят всё сразу: один сервис с долгим релизным циклом, и договорённость рассыпается. Версию схемы пишут в тело или заголовок и проверяют тестами на фикстурах, а не надеждами.

Эволюция схемы: правила безопасных изменений#

Формат сообщений очереди живёт дольше любого релиза: старые сообщения дожидаются в durable-очереди, пока их не разберёт новая версия потребителя. Правила, которые делают изменения безопасными. Аддитивность: новые поля — опциональные, с omitempty в теге, — потребители, которые о них не знают, просто игнорируют. Два шага: сначала выкатывается потребитель, умеющий читать новое поле, затем издатель начинает его писать; обратный порядок ломает старых потребителей. Переименование — тоже два шага и период совместимости: пишите оба ключа, переводите читателей, убирайте старый, когда метрики скажут, что его перестали встречать. Типы полей не меняют: число и строка — разные контракты; при необходимости заводят новое поле рядом. Версия схемы в конверте — заголовок schema-version или поле version в теле — позволяет ветвить разбор; без неё эволюция превращается в археологию. Правило проверки — тесты на фикстурах из боевого трафика: старые сообщения обязаны разбираться свежим кодом.

Потребитель: разбор и валидация

type Order struct {
        ID     string "json:\"id\""
        Amount int    "json:\"amount\""
}

func handle(body []byte) error {
        var order Order
        if err := json.Unmarshal(body, &order); err != nil {
                return fmt.Errorf("разбор тела: %w", err)
        }
        if order.ID == "" || order.Amount <= 0 {
                return fmt.Errorf("невалидный заказ: %+v", order)
        }
        log.Println("заказ валиден:", order.ID)
        return nil
}

// в цикле воркера
for d := range msgs {
        if err := handle(d.Body); err != nil {
                log.Println("в DLQ:", err)
                d.Nack(false, false)
                continue
        }
        d.Ack(false)
}

Две ступени проверки: разбор — синтаксис и типы, затем валидация — обязательные поля и бизнес-границы. Провал любой ступени уводит сообщение с requeue=false: повтор не исправит ни битый JSON, ни нулевую сумму; именно так обходится ядовитый цикл из восьмого урока.

Заголовки против тела: что куда класть#

У сообщения два места для данных, и путаница между ними стоит дороже, чем кажется. В теле — бизнес-содержимое: то, без чего обработка невозможна и что меняется по правилам эволюции схемы. В заголовках — служебные метки: MessageId для дедупликации из восьмого урока, Timestamp, идентификатор трассировки, версия схемы, тип события — то, что брокер и инфраструктура читают, не разбирая тело. Тип контента — отдельные поля в amqp.Publishing, не заголовок: ContentType и ContentEncoding живут на своём месте. Антипаттерн — бизнес-данные в заголовках: их туда тянет headers-маршрутизация из четырнадцатого урока, а остальным причинам взяться там нет; заголовки едут в каждой доставке, растят память брокера и не имеют схемы. Обратный антипаттерн — служебное в теле: trace-id внутри JSON заставляет каждого потребителя разбирать тело ради поля, которое ему не нужно. Практика: заголовок — о доставке и диагностике, тело — о домене.

Контракт между сервисами: кто владеет схемой#

Структура Order живёт в одном репозитории, а читать её будут два сервиса — и возникает вопрос владения. Общий пакет с типами кажется решением: один тип, ноль расхождений. Цена — связность: любое изменение пакета требует согласования двух команд, а при разных релизных циклах общий пакет превращается в узкое место. Спорная, но рабочая альтернатива — копия типа у потребителя: сервис владеет своей структурой разбора и не зависит от издателя вообще; контракт проверяют интеграционными тестами на фикстурах реального трафика. В монорепе с одной командой общий пакет честнее; в разделённых командах копии с тестами на контракте надёжнее. Проверка без брокера делается просто: обработчик Delivery как функция из двадцать четвёртого урока принимает срез байтов, и тест гоняет фикстуры JSON без amqp. Так формат сообщений очереди становится тестируемым артефактом, а не устной договорённостью в чате.

Код урока — локально, интерактивы — на странице

Пакет encoding/json не входит в набор песочницы — там доступны fmt, strings, strconv, math, sort, errors и time, без сети и брокера: полный код урока запускается локально против Docker-контейнера из второго урока. Быстрая проверка на живой очереди: опубликуйте структуру и откройте сообщение в management UI — там видны и ContentType, и тело JSON целиком. Интерактивы ниже проверят ContentType, ветки разбора и правила эволюции схемы.

Правила формата сообщений#

Тело — срез байтов: семантику дают ContentType application/json и структура, о которой договорились.
Ошибку Marshal решают у издателя до публикации; ошибку Unmarshal потребитель обязан обработать.
Битое тело не станет валидным от повтора: Nack с requeue=false и путь в DLQ.
Разбор без валидации пропускает нули: обязательные поля проверяют после Unmarshal.
Схему меняют аддитивно: новые поля — опциональные, переименования — через период двух ключей.
Версию схемы кладут в заголовок или конверт и проверяют тестами на боевых фикстурах.

Практика вокруг формата: шестой урок про amqp.Publishing дал поля Publishing, о которых здесь шла речь, восьмой про ack объяснил Nack с requeue=false, а шестнадцатый про dead letter выстроит очередь повторов вокруг ядовитых тел. Обработчик Delivery как функция и тесты без брокера — двадцать четвёртый урок, конверты с версиями пригодятся в итоговом проекте двадцать пятого. Глубокий разбор самого encoding/json с тегами и omitempty остался в двадцать первом уроке курса по Go; карта курса — на странице курса.

Проверка: ContentType

Издатель отправил JSON, но забыл указать ContentType. Что произойдёт?

Проверка: разбор и валидация

Воркер получил тело {"id":"ord-1042"} — валидный JSON, но поле amount отсутствует. Структура ожидает ID string и Amount int. Что вернёт json.Unmarshal?

Что напечатает разбор?

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

type Note struct {
        Text string "json:\"text\""
}

body := []byte("{"text":"привет","extra":42}")
var n Note
err := json.Unmarshal(body, &n)
fmt.Println(n.Text, err == nil)

Соберите издателя JSON

Допишите маршалинг, тип контента (строку вместе с кавычками) и разбор на потребителе.

body, err := json.(order) if err != nil { log.Fatal(err) } err = ch.PublishWithContext(ctx, "", "orders.json", false, false, amqp.Publishing{ ContentType: , Body: body, DeliveryMode: amqp.Persistent, }) var in Order err = json.(d.Body, &in)

Соберите потребителя JSON

Расставьте строки в порядке выполнения: соединение, канал, очередь, подписка, чтение, разбор.

Строки

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

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

Карточки: формат сообщений

Термин

json.Marshal

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

Определение

Превращает структуру в срез байтов для Body. Ошибка решается у издателя до публикации.

1 / 6

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

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

Вывод

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

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

Тело — срез байтов: контракт делают json.Marshal, ContentType и структура с тегами.
amqp091-go не заполняет ContentType: application/json ставят явно.
Unmarshal игнорирует незнакомые поля и оставляет нули: валидация обязательных полей идёт сразу после разбора.
Битое тело не оживёт от повтора: Nack с requeue=false и DLQ из шестнадцатого урока.
Схему меняют аддитивно и в два шага: сначала потребители, потом издатели.
Служебные метки — в заголовках, бизнес-данные — в теле: границу путать дорого.
Контракт проверяется тестами на фикстурах трафика, а не устными договорённостями.

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

Соберите издателя со структурой заказа из урока и потребителя с валидацией обязательных полей. Опубликуйте пять корректных заказов и один битый — с обрезанным телом — и убедитесь, что битый ушёл с Nack в DLQ, а воркер не остановился. Затем добавьте в структуру поле discount с omitempty: сначала выкатите потребителя, который его читает, и только потом издателя, который его пишет. В конце опубликуйте сообщение с лишним полем и сравните мягкий разбор со строгим DisallowUnknownFields.

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

Можно ли отправлять структуру как есть, без тегов?

Да: json.Marshal выведет поля их именами — ID, Amount — и потребители разберут те же имена. Контракт получается с заглавными ключами, что непривычно внешним читателям и инструментам; теги с ключами в нижнем регистре ставят сразу, потому что переименование ключей после релиза — ломающее изменение для всех потребителей.

Что делать, если сообщение опубликовано в старой схеме, а потребитель уже новый?

Проектировать разбор так, чтобы он переживал старое: опциональные поля, нули вместо отсутствующих, версия схемы в конверте. Обратная совместимость — обязанность потребителя: он доживает сообщения, которые издатель выпустил до релиза. Именно поэтому поля добавляют, а не переименовывают: старые тела должны разбираться свежим кодом.

JSON или protobuf для очереди?

Если потоки небольшие, а отладка важна — JSON: читаемость в логах и management UI экономит часы расследований. Protobuf берут под большие объёмы и строгие схемы с кодогенерацией; цена — тело не прочитать глазами и синхронизация proto-файлов между командами. Начинают обычно с JSON и мигрируют на бинарные форматы точечно, там, где парсинг стал измеримой статьёй расходов.

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

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