Как работать с Kafka?

Основы работы с Kafka в Go-разработке

Что такое Apache Kafka?

Apache Kafka — это распределенная потоковая платформа для обработки событий в реальном времени, построенная на основе публикации-подписки. В Go-экосистеме для работы с Kafka используются несколько популярных библиотек:

1. Sarama — наиболее распространенная Go-библиотека
2. Confluent Kafka Go — официальная библиотека от Confluent
3. kafka-go — альтернативная библиотека с другим API

Основные концепции Kafka

Производители (Producers) отправляют сообщения в топики (Topics), которые разделены на партиции (Partitions). Потребители (Consumers) читают сообщения из топиков, объединяясь в группы потребителей (Consumer Groups) для распределения нагрузки.

Установка и настройка Sarama

// Установка библиотеки
// go get github.com/IBM/sarama

import "github.com/IBM/sarama"

// Конфигурация продюсера
config := sarama.NewConfig()
config.Producer.RequiredAcks = sarama.WaitForAll
config.Producer.Retry.Max = 5
config.Producer.Return.Successes = true

Создание производителя (Producer)

func createProducer(brokers []string) (sarama.SyncProducer, error) {
    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll
    config.Producer.Retry.Max = 5
    config.Producer.Return.Successes = true
    
    producer, err := sarama.NewSyncProducer(brokers, config)
    if err != nil {
        return nil, fmt.Errorf("failed to create producer: %w", err)
    }
    
    return producer, nil
}

// Отправка сообщения
func sendMessage(producer sarama.SyncProducer, topic, key, value string) error {
    msg := &sarama.ProducerMessage{
        Topic: topic,
        Key:   sarama.StringEncoder(key),
        Value: sarama.StringEncoder(value),
    }
    
    partition, offset, err := producer.SendMessage(msg)
    if err != nil {
        return fmt.Errorf("failed to send message: %w", err)
    }
    
    log.Printf("Message sent to partition %d at offset %d", partition, offset)
    return nil
}

Создание потребителя (Consumer)

func createConsumer(brokers []string, groupID string) (sarama.ConsumerGroup, error) {
    config := sarama.NewConfig()
    config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRange
    config.Consumer.Offsets.Initial = sarama.OffsetOldest
    
    consumer, err := sarama.NewConsumerGroup(brokers, groupID, config)
    if err != nil {
        return nil, fmt.Errorf("failed to create consumer: %w", err)
    }
    
    return consumer, nil
}

// Обработчик сообщений
type ConsumerHandler struct {
    Ready chan bool
}

func (h *ConsumerHandler) Setup(sarama.ConsumerGroupSession) error {
    close(h.Ready)
    return nil
}

func (h *ConsumerHandler) Cleanup(sarama.ConsumerGroupSession) error {
    return nil
}

func (h *ConsumerHandler) ConsumeClaim(
    session sarama.ConsumerGroupSession, 
    claim sarama.ConsumerGroupClaim,
) error {
    for message := range claim.Messages() {
        log.Printf("Message received: topic=%s, partition=%d, offset=%d, key=%s, value=%s",
            message.Topic, message.Partition, message.Offset, 
            string(message.Key), string(message.Value))
        
        session.MarkMessage(message, "")
    }
    
    return nil
}

Паттерны работы с Kafka в Go

1. Пакетная обработка сообщений

func processBatch(messages []*sarama.ConsumerMessage) error {
    var batch []string
    for _, msg := range messages {
        batch = append(batch, string(msg.Value))
    }
    
    // Обработка батча
    return processDataBatch(batch)
}

2. Обработка с контролем ошибок

func resilientConsumer(handler sarama.ConsumerGroupHandler) error {
    for {
        if err := consumer.Consume(ctx, topics, handler); err != nil {
            log.Printf("Consumer error: %v", err)
            time.Sleep(retryInterval)
            continue
        }
    }
}

3. Производитель с ретраями

func resilientProducerSend(producer sarama.SyncProducer, msg *sarama.ProducerMessage, maxRetries int) error {
    for i := 0; i < maxRetries; i++ {
        _, _, err := producer.SendMessage(msg)
        if err == nil {
            return nil
        }
        
        if i < maxRetries-1 {
            time.Sleep(time.Duration(math.Pow(2, float64(i))) * time.Second)
        }
    }
    
    return fmt.Errorf("failed after %d retries", maxRetries)
}

Лучшие практики работы с Kafka в Go

• Конфигурация тюнинг: Настройка batch.size, linger.ms, compression.type для оптимизации производительности
• Управление подключениями: Использование пулов подключений и graceful shutdown
• Мониторинг: Интеграция метрик через expvar или Prometheus
• Обработка ошибок: Реализация повторных попыток и схемы circuit breaker
• Сериализация: Использование эффективных форматов (Protobuf, Avro) вместо JSON
• Тестирование: Использование kafka-go или testcontainers для интеграционных тестов

Пример полной реализации

type KafkaService struct {
    producer sarama.SyncProducer
    consumer sarama.ConsumerGroup
    config   *KafkaConfig
}

func NewKafkaService(cfg *KafkaConfig) (*KafkaService, error) {
    service := &KafkaService{config: cfg}
    
    // Инициализация продюсера
    producer, err := createProducer(cfg.Brokers)
    if err != nil {
        return nil, err
    }
    service.producer = producer
    
    // Инициализация консьюмера
    consumer, err := createConsumer(cfg.Brokers, cfg.GroupID)
    if err != nil {
        return nil, err
    }
    service.consumer = consumer
    
    return service, nil
}

func (k *KafkaService) StartConsuming(ctx context.Context, topics []string) error {
    handler := &ConsumerHandler{Ready: make(chan bool)}
    
    go func() {
        for {
            select {
            case <-ctx.Done():
                return
            default:
                if err := k.consumer.Consume(ctx, topics, handler); err != nil {
                    log.Printf("Consumption error: %v", err)
                }
            }
        }
    }()
    
    <-handler.Ready
    return nil
}

Заключение

Работа с Kafka в Go требует понимания как внутренней механики Kafka, так и особенностей Go-рантайма. Ключевые аспекты: правильная настройка конфигурации, обработка ошибок, управление ресурсами и мониторинг. Sarama предоставляет наиболее полный функционал, но для production-систем важно добавлять дополнительные слои абстракции, логирования и мониторинга. Современные практики включают использование контекстов для управления временем жизни операций и интеграцию с OpenTelemetry для трейсинга распределенных систем.