Sarama Go: методы, примеры, установка

Что такое Sarama и зачем она нужна?

Sarama — это высокопроизводительная, чистая Go-библиотека для работы с Apache Kafka. Она предоставляет полный набор инструментов для взаимодействия с кластером Kafka, включая продюсеров (producers) для отправки сообщений, консюмеров (consumers) для чтения сообщений, а также административные функции для управления топиками, партициями и конфигурациями.

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

Зачем использовать Sarama?

Чистый Go: Sarama написана полностью на Go без внешних зависимостей на C/C++, что упрощает сборку и развертывание.
Высокая производительность: Библиотека оптимизирована для работы с большими объемами данных и поддерживает асинхронную отправку сообщений.
Гибкость: Поддерживает различные режимы работы продюсеров (синхронный, асинхронный) и консюмеров (простой, группа консюмеров).
Активное сообщество: Sarama — одна из самых популярных Kafka-библиотек для Go с большим количеством контрибьюторов и обширной документацией.
Полный контроль: Дает разработчику полный контроль над конфигурацией, включая управление оффсетами, сериализацию/десериализацию и обработку ошибок.

Установка

Установка Sarama в проект Go осуществляется стандартным способом через менеджер пакетов go get.

go get github.com/IBM/sarama

После выполнения этой команды библиотека будет добавлена в ваш go.mod файл, и вы сможете импортировать ее в своем коде.

Важно: Убедитесь, что у вас установлен Go версии 1.18 или выше, так как Sarama использует современные возможности языка.

Быстрый старт: отправка и получение сообщений

Давайте напишем простейший пример, который отправит сообщение в топик Kafka и прочитает его.

1. Продюсер (Producer) — отправка сообщения

package main

import (
    "fmt"
    "log"

    "github.com/IBM/sarama"
)

func main() {
    // 1. Конфигурация продюсера
    config := sarama.NewConfig()
    config.Producer.Return.Successes = true // Нужно для синхронного режима
    config.Producer.RequiredAcks = sarama.WaitForAll // Ожидание подтверждения от всех реплик

    // 2. Создание синхронного продюсера
    producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
    if err != nil {
        log.Fatalf("Не удалось создать продюсера: %v", err)
    }
    defer producer.Close()

    // 3. Создание сообщения
    msg := &sarama.ProducerMessage{
        Topic: "my-topic",
        Key:   sarama.StringEncoder("my-key"),
        Value: sarama.StringEncoder("Привет, Kafka!"),
    }

    // 4. Отправка сообщения
    partition, offset, err := producer.SendMessage(msg)
    if err != nil {
        log.Fatalf("Не удалось отправить сообщение: %v", err)
    }

    fmt.Printf("Сообщение отправлено в партицию %d с оффсетом %d\n", partition, offset)
}

2. Консюмер (Consumer) — получение сообщения

package main

import (
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"

    "github.com/IBM/sarama"
)

func main() {
    // 1. Конфигурация консюмера
    config := sarama.NewConfig()
    config.Consumer.Return.Errors = true
    config.Consumer.Offsets.Initial = sarama.OffsetOldest // Читать с самого начала

    // 2. Создание консюмера
    consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, config)
    if err != nil {
        log.Fatalf("Не удалось создать консюмера: %v", err)
    }
    defer consumer.Close()

    // 3. Подписка на партицию
    partitionConsumer, err := consumer.ConsumePartition("my-topic", 0, sarama.OffsetNewest)
    if err != nil {
        log.Fatalf("Не удалось подписаться на партицию: %v", err)
    }
    defer partitionConsumer.Close()

    // 4. Обработка сигналов для graceful shutdown
    signals := make(chan os.Signal, 1)
    signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)

    // 5. Чтение сообщений
    for {
        select {
        case msg := <-partitionConsumer.Messages():
            fmt.Printf("Получено сообщение: ключ=%s, значение=%s, оффсет=%d\n",
                string(msg.Key), string(msg.Value), msg.Offset)
        case err := <-partitionConsumer.Errors():
            log.Printf("Ошибка консюмера: %v", err)
        case <-signals:
            fmt.Println("Завершение работы...")
            return
        }
    }
}

Основные методы и функции Sarama (10+ штук)

1. `sarama.NewConfig()`

Что делает: Создает новый объект конфигурации с настройками по умолчанию. Это отправная точка для настройки любого клиента Sarama.

Параметры: Нет.

Пример:

config := sarama.NewConfig()
config.Producer.Timeout = 10 * time.Second
config.Consumer.MaxWaitTime = 500 * time.Millisecond

2. `sarama.NewSyncProducer(brokers, config)`

Что делает: Создает синхронного продюсера. Каждый вызов SendMessage блокируется до получения подтверждения от брокера.

Параметры:

brokers []string — список адресов брокеров Kafka (например, []string{"localhost:9092"}).
config *sarama.Config — объект конфигурации.

Пример:

producer, err := sarama.NewSyncProducer([]string{"kafka1:9092", "kafka2:9092"}, config)
if err != nil {
    log.Fatal(err)
}
defer producer.Close()

3. `producer.SendMessage(msg)`

Что делает: Отправляет одно сообщение в Kafka и возвращает партицию и оффсет, куда оно было записано.

Параметры:

msg *sarama.ProducerMessage — сообщение с полями Topic, Key, Value и опционально Headers.

Пример:

msg := &sarama.ProducerMessage{
    Topic: "orders",
    Key:   sarama.StringEncoder("order-123"),
    Value: sarama.StringEncoder(`{"user_id": 42, "amount": 100.50}`),
}
partition, offset, err := producer.SendMessage(msg)

4. `producer.SendMessages(msgs)`

Что делает: Отправляет пакет сообщений за один вызов. Это эффективнее, чем отправлять каждое сообщение по отдельности.

Параметры:

msgs []*sarama.ProducerMessage — срез сообщений.

Пример:

msgs := []*sarama.ProducerMessage{
    {Topic: "logs", Value: sarama.StringEncoder("log1")},
    {Topic: "logs", Value: sarama.StringEncoder("log2")},
}
err := producer.SendMessages(msgs)

5. `sarama.NewAsyncProducer(brokers, config)`

Что делает: Создает асинхронного продюсера. Сообщения отправляются в фоновом режиме, а результаты (успех/ошибка) приходят через каналы.

Параметры: Аналогично NewSyncProducer.

Пример:

producer, err := sarama.NewAsyncProducer([]string{"localhost:9092"}, config)
defer producer.Close()

// Отправка
producer.Input() <- &sarama.ProducerMessage{
    Topic: "events",
    Value: sarama.StringEncoder("async event"),
}

// Обработка результатов
go func() {
    for success := range producer.Successes() {
        fmt.Printf("Успешно отправлено: %v\n", success)
    }
}()
go func() {
    for err := range producer.Errors() {
        fmt.Printf("Ошибка: %v\n", err)
    }
}()

6. `sarama.NewConsumer(brokers, config)`

Что делает: Создает простого консюмера, который может читать сообщения из указанной партиции топика.

Параметры:

brokers []string — список брокеров.
config *sarama.Config — конфигурация.

Пример:

consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, config)
if err != nil {
    log.Fatal(err)
}
defer consumer.Close()

7. `consumer.ConsumePartition(topic, partition, offset)`

Что делает: Подписывается на конкретную партицию топика, начиная с указанного оффсета.

Параметры:

topic string — имя топика.
partition int32 — номер партиции.
offset int64 — начальный оффсет (sarama.OffsetNewest, sarama.OffsetOldest или конкретное значение).

Пример:

pc, err := consumer.ConsumePartition("my-topic", 0, sarama.OffsetOldest)
if err != nil {
    log.Fatal(err)
}
defer pc.Close()

for msg := range pc.Messages() {
    fmt.Printf("Сообщение: %s\n", msg.Value)
}

8. `sarama.NewConsumerGroup(brokers, groupID, config)`

Что делает: Создает консюмер-группу. Это более продвинутый способ потребления, который автоматически балансирует нагрузку между консюмерами в группе.

Параметры:

brokers []string — список брокеров.
groupID string — идентификатор группы консюмеров.
config *sarama.Config — конфигурация.

Пример:

group, err := sarama.NewConsumerGroup([]string{"localhost:9092"}, "my-group", config)
if err != nil {
    log.Fatal(err)
}
defer group.Close()

9. `group.Consume(ctx, topics, handler)`

Что делает: Запускает потребление сообщений из указанных топиков с использованием заданного обработчика.

Параметры:

ctx context.Context — контекст для отмены.
topics []string — список топиков.
handler sarama.ConsumerGroupHandler — объект, реализующий интерфейс обработчика.

Пример:

handler := &MyHandler{}
for {
    err := group.Consume(ctx, []string{"my-topic"}, handler)
    if err != nil {
        log.Printf("Ошибка потребления: %v", err)
    }
}

// Реализация обработчика
type MyHandler struct{}

func (h *MyHandler) Setup(sarama.C


                
    
  Также в библиотеке
  
                
                  
                 SwiftUI для Swift: полное руководство с примерами
      
                
                  
                 Rector для PHP: полное руководство с примерами
      
                
                  
                 doctest для C++: полное руководство с примерами
      
                
                  
                 Работа с S3 и облачными хранилищами в PHP: полное руководство
      
                
                  
                 Кэширование в PHP: Redis, Memcached и APCu — полное руководство