--- title: "Первый консьюмер: Reader" description: "Первый kafka consumer на Go: kafka.NewReader и ReadMessage, цикл чтения, поля kafka.Message, SetOffset и Lag, чтение партиции и ловушки Reader." course: "Kafka в Go: продюсеры и консьюмеры на kafka-go" url: "https://golangmaster.ru/courses/kafka-go/lessons/06-pervyj-consumer/" published: "2026-09-19" updated: "2026-09-29" minutes: 18 --- # Первый консьюмер: Reader *Курс «Kafka в Go: продюсеры и консьюмеры на kafka-go», урок 6 из 25. ~18 минут. Обновлено 29.09.2026. Источник: https://golangmaster.ru/courses/kafka-go/lessons/06-pervyj-consumer/* Первый kafka reader собирается из трёх частей: kafka.NewReader с конфигом ReaderConfig, блокирующий ReadMessage и вечный цикл чтения консьюмера — каркас, на котором стоит вся читающая сторона Kafka. Продюсеры из прошлых уроков уже пишут события в брокер — пора их забирать. Заодно разберём поля kafka.Message, SetOffset, Lag и Stats — метрики, без которых чтение остаётся слепым. К концу урока у вас будет каркас, в который вписывается любой kafka consumer golang-сервиса, от демона телеметрии до биллинга. Ситуация до боли знакомая: платежи идут в топик payments, события копятся, а читателя нет. Данные лежат на диске брокера и никуда не денутся — в этом сила журнала из первого урока, — но чтобы из лога что-то получить, нужен процесс, который ходит к брокеру и вычитывает партицию по оффсетам. В kafka-go для этого есть Reader: высокоуровневый клиент чтения, который сам держит соединение, сам запрашивает батчи сообщений и сам переживает разрывы сети. Мы напишем консьюмера, читающего одну партицию напрямую — это самый честный способ увидеть механику: где лежит сообщение, какой у него оффсет и как двигается позиция чтения. Группы консьюмеров добавят сверху распределение партиций, а коммиты — память о прочитанном; об этом уроки седьмой и восьмой. Здесь же — фундамент: конфиг, блокирующий вызов, цикл и метрики. Ключи и партиционирование из [пятого урока](/courses/kafka-go/lessons/05-klyuchi-particionirovanie/) уже разложили события по партициям — теперь разберём, как одну партицию вычитать целиком и что произойдёт с позицией при рестарте. ### kafka.NewReader: ReaderConfig и первый kafka consumer Консьюмер начинается с kafka.NewReader, который принимает ReaderConfig и паникует, если конфиг невалиден. Минимальный набор — Brokers, Topic и либо Partition, либо GroupID: это два режима работы, взаимоисключающие на уровне Validate. Чтение конкретной партиции — режим без состояния: вы говорите, откуда читать, и Reader не коммитит ничего. Чтение группой — режим с координацией, о нём следующий урок. Дефолты конфига стоят того, чтобы их знать: MinBytes равен 1 — брокер отвечает, как только есть хоть байт данных; MaxBytes — примерно мегабайт на батч, чтобы ответ брокера не разорвал память; MaxWait — 10 секунд ожидания данных в незаполненном батче. QueueCapacity внутренней очереди — 100 сообщений: они выравнивают скорость fetch и скорость вашего обработчика. Рядом стоят и менее заметные поля: MaxAttempts ограничивает тремя число попыток соединения, прежде чем ошибка уйдёт наружу, а ReadBatchTimeout — десятью секундами выборку из уже полученного батча. Эти цифры объясняют поведение, которое без них выглядит загадкой: почему консьюмер на пустом топике просыпается раз в десять секунд и почему большой MinBytes на низконагруженном топике добавляет задержку — брокер копит данные до минимума, прежде чем ответить. Для первого kafka reader оставьте дефолты как есть: они разумны, а тюнинг батчей — дело измерений, о которых поговорим в десятом уроке на продюсерской стороне. ### Консьюмер одной партиции целиком ```go package main import ( "context" "fmt" "log" "github.com/segmentio/kafka-go" ) func main() { ctx := context.Background() reader := kafka.NewReader(kafka.ReaderConfig{ Brokers: []string{"localhost:9092"}, Topic: "payments", Partition: 0, }) defer reader.Close() if err := reader.SetOffset(kafka.FirstOffset); err != nil { log.Fatal(err) } for { m, err := reader.ReadMessage(ctx) if err != nil { log.Printf("reading stopped: %v", err) break } fmt.Printf("offset=%d partition=%d key=%s value=%s\n", m.Offset, m.Partition, m.Key, m.Value) } } ``` Каркас чтения без группы: ReaderConfig требует адресов, топика и номера партиции, а MinBytes, MaxBytes и MaxWait kafka-go подставит сам. SetOffset с kafka.FirstOffset отправляет позицию к самому старому живому сообщению — как в predict второго урока. Дальше вечный for: ReadMessage блокируется до события, ошибка любого рода, включая io.EOF после Close, выводится в лог и гасит цикл. На пустом топике вызов просто ждёт. ### ReadMessage: блокирующий вызов и поля kafka.Message ReadMessage читает и возвращает следующее сообщение, блокируя горутину, пока его нет: на живом потоке это микросекунды, на пустой партиции — хоть до утра. Отменить ожидание можно контекстом — так останавливают читателя при выключении сервиса. Возвращаемое значение — kafka.Message, и его поля — это адрес события в логе: Topic и Partition говорят, откуда, Offset — позиция, Key и Value — байты, Time — время записи брокером, Headers — служебные пары ключ-значение, которые продюсер приложил сам: трейс-идентификаторы и метки маршрутизации живут именно там, а не в Value. Время записи брокером, а не обработки: на нём строится задержка сквозной доставки. Поле HighWaterMark показывает последний оффсет, видимый брокеру, — из него считается лаг. Ошибку ReadMessage оборачивает в префикс fetching message, поэтому проверяйте через errors.Is и errors.As, а не сравнением строк. По устройству метод — тонкая обёртка: внутри это FetchMessage, который при группе сопровождается CommitMessages, и ошибка коммита приедет с префиксом committing message. Особый случай — io.EOF: его Reader возвращает после Close, и это штатный сигнал остановки, а не сбой. Если программа прочитала лог до конца, ReadMessage не вернёт управление с ошибкой — он повиснет и будет ждать новых данных, потому что чтение журнала в Kafka бесконечно по определению. Отсюда практическое правило вычитки истории: ограничивайте её не чтением до конца, а оффсетом или контекстом с дедлайном из восемнадцатого урока — иначе процесс застрянет в бесконечном ожидании после последнего события. ### Проверка: что вернёт ReadMessage после Close **Вопрос:** Reader прочитал поток, затем main вызвал reader.Close() в другой горутине. Что вернёт заблокированный ReadMessage? - ✅ io.EOF — штатный сигнал, что поток закрыт - Панику — читать закрытый Reader нельзя - Ошибку сети — Close рвёт соединение с брокером - Последнее сообщение ещё раз Close останавливает Reader, и ReadMessage возвращает io.EOF — документированный способ завершить цикл чтения. Поэтому в цикле различают io.EOF (тихий выход) и остальные ошибки (лог с решением — останавливаться или пересоздать Reader). ### Цикл чтения консьюмера: где выход и как не зациклиться Цикл консьюмера живёт весь срок жизни сервиса, и единственная ветка выхода в нём — ошибка от ReadMessage. Обработка этой ошибки — место, где новички ломают продакшен. Первый грех: логировать ошибку и продолжать, не отличая io.EOF от сбоя. После Close такой цикл превращается в горловину: Reader мёртв, ReadMessage мгновенно возвращает io.EOF, лог раздувается на гигабайты, процесс жжёт CPU впустую. Второй грех — выход по любой ошибке без стратегии: сеть мигнула, брокер перезапустился, и читатель умер вместе с сервисом, хотя kafka-go умеет переподключаться сам. Разумная схема выглядит так: io.EOF — тихий выход, контекст отменён — тихий выход, остальное — лог и либо выход с падением процесса, чтобы оркестратор перезапустил, либо пересоздание Reader с паузой. Контекст тут — главный инструмент: отменённый контекст гасит блокирующий вызов мгновенно, поэтому выключение сервиса выглядит как отмена, а не как убийство горутин с риском недоделанной обработки. Внутри Reader между попытками fetch уже стоит бэкофф от 100 миллисекунд до секунды, так что при живом Reader шторма запросов не будет. Определять политику перезапуска заранее — часть контракта консьюмера: ошибка в этой ветке видна только в час ночи, и смена решения на ходу стоит дороже десяти минут проектирования. > **GroupID в конфиге меняет поведение ReadMessage.** Как только в ReaderConfig появился GroupID, ReadMessage начинает коммитить оффсет сам — документация честно предупреждает, что оффсет может закоммититься раньше, чем сообщение обработано. Для быстрых обработчиков это незаметно, для медленных и дорогих — риск потерять события при падении. На словах «закоммитился раньше» держится вся тема девятого урока про гарантии: кто подтвердил позицию до работы, тот выбрал at-most-once, как бы он ни был удивлён после инцидента. Полный разбор коммитов, CommitInterval и FetchMessage — восьмой урок; сейчас достаточно знать, что поведение чтения зависит от наличия GroupID. ### Что Reader берёт на себя - Соединения: реконнекты к брокерам и повторные попытки fetch внутри себя. - Батчи: запросы MinBytes/MaxBytes/MaxWait и очередь QueueCapacity на 100 сообщений. - Протокол: разбор fetch-ответов, оффсетов и меток времени без вашего участия. - Группы: при GroupID — heartbeats, join и синхронизацию, об этом следующий урок. - Метрики: счётчики Stats и лаг из коробки, без внешних инструментов. - Позицию: SetOffset и Offset без группы — сдвигайте чтение в обе стороны. ### SetOffset, Lag и Stats: позиция и метрики читателя Без группы вы владеете позицией напрямую, и это стоит потренировать на живом топике. SetOffset принимает kafka.FirstOffset, kafka.LastOffset или конкретное число — это переигрывание истории из [урока про устройство брокера](/courses/kafka-go/lessons/02-koncepty-brokera/) в чистом виде. Offset возвращает текущую позицию — следующий оффсет, который будет прочитан. Lag — расстояние до конца лога, вычисленное по последнему fetch: на живом потоке цифра достаточно свежая, на пустом устаревает, потому что fetch не выполняется. Для честной цифры есть ReadLag — отдельный запрос к брокеру, который вернёт актуальное отставание; ценой будет лишний round-trip, поэтому зовут его по таймеру, а не на каждое сообщение. С GroupID картина другая: SetOffset и ReadLag вернут ошибку unavailable when GroupID is set, а Offset и Lag — минус единицу, потому что позиция группы живёт на брокере, и клиенту её не отдают. Наконец, Stats отдаёт структуру ReaderStats со счётчиками Fetches, Messages, Bytes, Errors и Rebalances — их стоит экспортировать в мониторинг, прежде чем городить свой учёт: встроенные метрики уже считают то же самое, а параллельный учёт в своих обёртках — только источник расхождений. Разница между ReadLag по таймеру и встроенным Lag видна на графике за минуты — на тихом топике встроенная цифра стоит на месте и обманывает, пока не придёт новое сообщение. ### Меню: позиция и лаг без группы ```go // следующий оффсет и отставание по последнему fetch fmt.Println(reader.Offset(), reader.Lag()) // точный лаг отдельным запросом к брокеру lag, err := reader.ReadLag(ctx) if err != nil { log.Printf("lag: %v", err) } else { fmt.Println("lag:", lag) } // переиграть историю заново — позиция крутится назад err = reader.SetOffset(kafka.FirstOffset) fmt.Println("rewind:", err) ``` Чтение партиции без группы даёт прямой доступ к позиции: Offset — следующий оффсет, Lag — отставание по последнему fetch, ReadLag — свежая цифра отдельным запросом ListOffsets. SetOffset двигает позицию в обе стороны, поэтому повторное чтение истории — штатная операция. С GroupID все три приёма недоступны: у группы позиция одна и живёт на брокере. > **Живое место: Partition и GroupID одновременно.** История из код-ревью: конфиг консьюмера копировался из двух примеров, и в ReaderConfig остались и Partition: 0, и GroupID: billing. В рантайме это паника NewReader с текстом either Partition or GroupID may be specified, but not both — Validate честно ругается, но только при старте процесса, а не в IDE. Симптом узнаваем: сервис не поднимается вообще, и стек указывает в конструктор. Симметричная ловушка — SetOffset при живой группе: ошибка unavailable when GroupID is set возвращается уже в рантайме, посреди работающего цикла, если кто-то добавил подстройку позиции, забыв про режим группы. Правило простое: определились с режимом — уберите противоположное поле из конфига вовсе, а не закомментируйте. ### Методы Reader одним столом | Метод | Что делает | С GroupID | | --- | --- | --- | | ReadMessage(ctx) | Читает следующее сообщение; с группой — ещё и коммитит | да — коммит внутри вызова | | FetchMessage(ctx) | Читает без коммита — ручное подтверждение | да — парная к CommitMessages | | SetOffset(int64) | Ставит позицию чтения: FirstOffset, LastOffset, число | нет — вернёт ошибку | | Lag() / ReadLag(ctx) | Отставание от конца лога: кэш против точного запроса | нет: Lag вернёт -1, ReadLag — ошибку | | Stats() | Счётчики Fetches, Messages, Bytes, Errors, Rebalances | да — включая ребалансы | | Close() | Останавливает Reader; ReadMessage вернёт io.EOF | нет явного выхода: координатор заметит по SessionTimeout | ### Локальный запуск консьюмера: песочница против Docker Честно о формате: песочница сайта сеть не поднимает, поэтому этот код запускается локально: Go и Docker с брокером из третьего урока, два терминала — в одном продюсер из четвёртого, в другом этот Reader. Вы увидите своими глазами, как Offset растёт, а Lag тает, и что ReadMessage на пустом топике молчит, пока вы не пришлёте сообщение. Интерактивы ниже — тематические программы на чистом Go: они тренируют позицию, ошибки и лаг на эмуляции лога, и их вывод совпадает с тем, что покажет настоящий брокер на тех же данных. Если брокер ещё не поднят — интерактивов хватит для проверки в конце урока, а живой прогон останется пятиминуткой вместо вечера непонимания. ### Проверка: Partition и GroupID вместе **Вопрос:** В ReaderConfig заданы одновременно Partition: 0 и GroupID: billing. Что произойдёт? - ✅ NewReader паникует: поля взаимоисключающие по Validate - GroupID победит, Partition тихо проигнорируется - Partition победит, GroupID тихо проигнорируется - Reader будет чередовать оба режима чтения ReaderConfig.Validate возвращает ошибку either Partition or GroupID may be specified, but not both, а NewReader на невалидном конфиге паникует. Режимы чтения не смешиваются: либо адресная партиция без коммитов, либо группа с координацией. ### С какого оффсета пойдёт чтение? ```go reader := kafka.NewReader(kafka.ReaderConfig{ Brokers: []string{"localhost:9092"}, Topic: "payments", Partition: 0, }) reader.SetOffset(5) m1, _ := reader.ReadMessage(ctx) m2, _ := reader.ReadMessage(ctx) fmt.Println(m1.Offset, m2.Offset) ``` **Что выведет программа?** - ✅ 5 и 6: чтение начинается с заданного оффсета и идёт по логу дальше - 0 и 1: ReadMessage всегда начинает партицию с начала - -2 и -1: SetOffset проигнорирован, работают FirstOffset и LastOffset - Ошибку: оффсета 5 в партиции нет SetOffset задаёт стартовую позицию чтения, и ReadMessage продолжает с неё последовательно, увеличивая оффсет на каждом шаге. Число 5 означает «начни с позиции 5»; константы FirstOffset и LastOffset просто сокращения для крайних значений лога. ### Соберите консьюмера партиции Допишите конструктор Reader, его закрытие и метод чтения. ```go reader := kafka.___(kafka.ReaderConfig{ Brokers: []string{"localhost:9092"}, Topic: "payments", Partition: 0, }) defer reader.___() m, err := reader.___(ctx) if err != nil { log.Printf("stop: %v", err) return } ``` Подсказка: Конструктор NewReader, defer с Close, а блокирующее чтение — ReadMessage с контекстом. Правильные ответы в пропусках: `NewReader`, `Close`, `ReadMessage` ### Соберите цикл чтения по строкам Расставьте строки консьюмера в порядке исполнения: создание Reader, отложенное закрытие, цикл, чтение, обработка ошибки, печать сообщения. Правильный порядок строк: 1. `reader := kafka.NewReader(kafka.ReaderConfig{Brokers: []string{"localhost:9092"}, Topic: "payments", Partition: 0})` 2. `defer reader.Close()` 3. `for {` 4. ` m, err := reader.ReadMessage(ctx)` 5. ` if err != nil {` 6. ` log.Printf("stop: %v", err)` 7. ` break` 8. ` }` 9. ` fmt.Println(m.Partition, m.Offset, string(m.Key))` 10. `}` ### Карточки: Reader изнутри - **kafka.NewReader** — Конструктор Reader с ReaderConfig: Brokers, Topic и Partition либо GroupID. Паникует на невалидном конфиге — взаимоисключающие поля ловит Validate. - **ReadMessage** — Блокирующее чтение следующего сообщения. С GroupID заодно коммитит оффсет. Отменяется контекстом, после Close возвращает io.EOF. - **kafka.Message** — Событие с адресом: Topic, Partition, Offset, Key, Value, Time, Headers, HighWaterMark. Key и Value — байты, остальное — метаданные лога. - **MinBytes и MaxBytes** — Границы батча в fetch: минимум — брокер копит данные, максимум — потолок ответа. Дефолты 1 байт и 1 МБ объясняют задержки на тихих топиках. - **SetOffset** — Ставит позицию чтения без группы: FirstOffset, LastOffset или число. С GroupID недоступен — у группы позиция живёт на брокере. - **Lag** — Отставание читателя от конца лога: HighWaterMark минус текущий оффсет. Точная цифра — ReadLag отдельным запросом. Один консьюмер на партицию — рабочая машина для вычитки истории и точечных потоков. Дальше масштаб: в [уроке про группы консьюмеров](/courses/kafka-go/lessons/07-gruppy-consjumerov/) GroupID превратит набор Reader'ов в группу, брокер начнёт делить партиции между членами, а нагрузка — расползаться по инстансам. Затем в [уроке про коммиты](/courses/kafka-go/lessons/08-kommit-ofsetov/) разберём, куда девается позиция чтения при рестарте и почему ReadMessage с группой — это уже решение о гарантиях доставки. Проверьте домашнее задание на живом брокере: без пары прогонов в терминале поля kafka.Message остаются абстракцией. Ключи решают партицию, партиция решает порядок — Reader лишь честно показывает результат этих решений, и теперь вы умеете это показание читать без лога координатора и догадок. ## Коротко: что запомнить - Reader — высокоуровневое чтение: реконнекты, fetch-батчи и очередь внутри, вам — цикл и обработка ошибок. - Partition и GroupID взаимоисключающие: NewReader паникует, если заданы оба. - ReadMessage блокируется до сообщения; io.EOF после Close — штатный сигнал остановки цикла. - kafka.Message — адрес события: Topic, Partition, Offset плюс байты Key и Value. - Без группы позиция ваша: SetOffset двигает её, Offset показывает, Lag считает отставание. - Дефолты MinBytes 1, MaxBytes 1 МБ, MaxWait 10 секунд объясняют паузы на тихих топиках. - Код урока — локальный Go и брокер в Docker; песочница показывает тематические программы на чистом Go. ## Домашнее задание Против брокера из третьего урока запустите консьюмер партиции из урока и продюсер из четвёртого. Отправьте десять платежей и убедитесь, что Reader видит их с оффсетами 0-9. Затем перезапустите консьюмера с SetOffset(5): первые пять событий должны приехать повторно — это и есть переигрывание истории. После этого снимите Stats до и после минуты чтения и выпишите, какие счётчики выросли: Fetches, Messages, Bytes. Финал — сломайте конфиг: добавьте в него GroupID рядом с Partition и убедитесь, что NewReader паникует с сообщением either Partition or GroupID. Заодно запишите, какой из методов таблицы выше оказался самым полезным: этот выбор пригодится в уроке про мониторинг лага. ## Вопросы и ответы по уроку ### ReadMessage и FetchMessage — когда какой? ReadMessage — чтение с самокоммитом при группе: короткий цикл, готовые гарантии по умолчанию. FetchMessage — чтение без коммита: им пользуются, когда подтверждение нужно после обработки. Без GroupID оба метода ведут себя одинаково — коммитить просто нечего, поэтому выбор между ними — вопрос читаемости, а не семантики. ### Почему консьюмер на тихом топике отвечает с задержкой? Сработали дефолты fetch: MinBytes 1 байт, но брокер держит запрос до MaxWait 10 секунд, ожидая данных для батча. Если задержка критична, уменьшайте MaxWait или MinBytes — ценой станет больше пустых запросов к брокеру. ### Можно ли читать несколько партиций одним Reader без группы? Нет: режим без группы привязан к одной партиции через Partition. Несколько партиций — либо Reader на каждую в своей горутине, либо GroupID: группу распределит партиции между членами сама, и Reader переключится на новый режим чтения после ребаланса. Дублей при этом не будет — каждая партиция достанется ровно одному члену.