Работа с Kafka в Go (Golang) | segmentio/kafka-go |…

Работа с Kafka в Go (Golang) с библиотекой segmentio/kafka-go

Apache Kafka — это распределённая платформа для потоковой передачи данных, ставшая стандартом де-факто для построения высоконагруженных систем обмена сообщениями. При разработке на Go (Golang) одной из самых популярных и надёжных библиотек для взаимодействия с Kafka является segmentio/kafka-go. В этой статье мы подробно разберём, как установить, настроить и использовать эту библиотеку для создания продюсеров и консьюмеров.

Почему segmentio/kafka-go?

На рынке существует несколько Go-библиотек для Kafka (например, Sarama и Confluent Go Client). segmentio/kafka-go выделяется следующими преимуществами:

Чистый Go — не требует C-зависимостей (в отличие от Confluent), что упрощает сборку и деплой.
Поддержка Kafka Protocol v2 — эффективное сжатие, батчинг и работа с транзакциями.
Простой API — интуитивно понятные интерфейсы для продюсеров, консьюмеров и администрирования топиков.
Активное сообщество — регулярные обновления и хорошая документация.

Установка библиотеки

Для начала работы добавьте библиотеку в ваш проект через go get:

go get github.com/segmentio/kafka-go

Убедитесь, что у вас запущен экземпляр Kafka (локально или через Docker). Пример минимальной конфигурации Docker Compose:

version: '3'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092

Создание продюсера (Producer)

Продюсер отправляет сообщения в топик Kafka. Рассмотрим базовый пример:

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "github.com/segmentio/kafka-go"
)

func main() {
    // Настройка продюсера
    writer := &kafka.Writer{
        Addr:     kafka.TCP("localhost:9092"),
        Topic:    "test-topic",
        Balancer: &kafka.LeastBytes{}, // Балансировка по размеру сообщения
        BatchSize: 10,                 // Отправлять батч из 10 сообщений
        BatchTimeout: 100 * time.Millisecond,
    }
    defer writer.Close()

    // Отправка сообщения
    err := writer.WriteMessages(context.Background(),
        kafka.Message{
            Key:   []byte("key-1"),
            Value: []byte("Hello, Kafka from Go!"),
        },
    )
    if err != nil {
        log.Fatal("Ошибка отправки сообщения:", err)
    }

    fmt.Println("Сообщение успешно отправлено")
}

Ключевые параметры kafka.Writer:

Addr — адрес брокера Kafka.
Topic — имя топика.
Balancer — стратегия распределения сообщений по партициям (LeastBytes, RoundRobin, Hash).
BatchSize и BatchTimeout — оптимизация производительности.

Создание консьюмера (Consumer)

Консьюмер читает сообщения из топика. Простейший пример с ручным управлением offset:

package main

import (
    "context"
    "fmt"
    "log"

    "github.com/segmentio/kafka-go"
)

func main() {
    // Настройка ридера (консьюмера)
    reader := kafka.NewReader(kafka.ReaderConfig{
        Brokers:   []string{"localhost:9092"},
        Topic:     "test-topic",
        GroupID:   "my-consumer-group",
        MinBytes:  10, // Минимальный размер батча
        MaxBytes:  10e6, // 10 MB
        StartOffset: kafka.FirstOffset, // Начать с самого первого сообщения
    })
    defer reader.Close()

    fmt.Println("Чтение сообщений из топика test-topic...")
    for {
        msg, err := reader.ReadMessage(context.Background())
        if err != nil {
            log.Fatal("Ошибка чтения:", err)
        }
        fmt.Printf("Получено: key=%s, value=%s, partition=%d, offset=%d\n",
            string(msg.Key), string(msg.Value), msg.Partition, msg.Offset)
    }
}

Важные поля ReaderConfig: