Работа с Apache Kafka в Go (Golang): библиотеки и примеры

Работа с Apache Kafka в Go (Golang): библиотеки, примеры и лучшие практики

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

Почему Kafka и Go — отличная пара?

Go создавался для создания быстрых, конкурентных и надёжных серверных приложений. Kafka, в свою очередь, требует высокой пропускной способности и минимальной задержки. Go-горутины и каналы идеально подходят для асинхронной обработки потоков сообщений. Кроме того, статическая типизация Go помогает избежать многих ошибок на этапе компиляции, что критично для production-систем.

1. Библиотека confluent-kafka-go

Это обёртка над C-клиентом librdkafka, обеспечивающая максимальную производительность и полную поддержку всех возможностей Kafka (транзакции, идемпотентность, протоколы безопасности). Рекомендуется для высоконагруженных систем.

Установка

Требует наличия librdkafka в системе:

# macOS
brew install librdkafka

# Ubuntu/Debian
sudo apt-get install librdkafka-dev

# Затем установка Go-пакета
go get -u github.com/confluentinc/confluent-kafka-go/kafka

Пример продюсера (Producer)

Отправляет сообщение в топик orders:

package main

import (
    "fmt"
    "github.com/confluentinc/confluent-kafka-go/kafka"
)

func main() {
    p, err := kafka.NewProducer(&kafka.ConfigMap{
        "bootstrap.servers": "localhost:9092",
        "acks":              "all",
    })
    if err != nil {
        panic(err)
    }
    defer p.Close()

    // Канал для уведомлений о доставке
    go func() {
        for e := range p.Events() {
            switch ev := e.(type) {
            case *kafka.Message:
                if ev.TopicPartition.Error != nil {
                    fmt.Printf("Ошибка доставки: %v\n", ev.TopicPartition.Error)
                } else {
                    fmt.Printf("Сообщение доставлено в %s [%d] offset %d\n",
                        *ev.TopicPartition.Topic, ev.TopicPartition.Partition, ev.TopicPartition.Offset)
                }
            }
        }
    }()

    topic := "orders"
    message := []byte("Новый заказ #12345")
    err = p.Produce(&kafka.Message{
        TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
        Value:          message,
        Key:            []byte("order_key"),
    }, nil)

    if err != nil {
        fmt.Printf("Ошибка отправки: %v\n", err)
    }

    // Ожидаем отправки всех сообщений
    p.Flush(15 * 1000)
}

Пример консюмера (Consumer)

Читает сообщения из топика orders в группе order-processors:

package main

import (
    "fmt"
    "github.com/confluentinc/confluent-kafka-go/kafka"
)

func main() {
    c, err := kafka.NewConsumer(&kafka.ConfigMap{
        "bootstrap.servers": "localhost:9092",
        "group.id":          "order-processors",
        "auto.offset.reset": "earliest",
    })
    if err != nil {
        panic(err)
    }
    defer c.Close()

    err = c.SubscribeTopics([]string{"orders"}, nil)
    if err != nil {
        panic(err)
    }

    for {
        msg, err := c.ReadMessage(-1) // бесконечное ожидание
        if err != nil {
            fmt.Printf("Ошибка получения: %v\n", err)
            continue
        }
        fmt.Printf("Получено сообщение: %s (ключ: %s) из %s [%d] offset %d\n",
            string(msg.Value), string(msg.Key), *msg.TopicPartition.Topic,
            msg.TopicPartition.Partition, msg.TopicPartition.Offset)

        // Здесь — бизнес-логика обработки заказа
    }
}

2. Библиотека sarama (Shopify/sarama)

Полностью реализована на Go, не требует внешних зависимостей. Легче в развёртывании, но может уступать confluent-kafka-go в производительности на очень высоких нагрузках. Отлично подходит для микросервисов и средних проектов.

Установка

go get -u github.com/Shopify/sarama

Пример продюсера с синхронной отправкой

package main

import (