Arbeit mit Kafka in Go (Golang): Bibliothek segmentio/kafka-go

Arbeit mit Kafka in Go (Golang) mit der Bibliothek segmentio/kafka-go

Apache Kafka ist eine verteilte Plattform für die Datenstromverarbeitung, die sich zum De-facto-Standard für den Aufbau hochlastiger Nachrichtenaustauschsysteme entwickelt hat. Bei der Entwicklung mit Go (Golang) ist segmentio/kafka-go eine der beliebtesten und zuverlässigsten Bibliotheken für die Interaktion mit Kafka. In diesem Artikel werden wir im Detail besprechen, wie man diese Bibliothek installiert, konfiguriert und verwendet, um Producer und Consumer zu erstellen.

Warum segmentio/kafka-go?

Es gibt mehrere Go-Bibliotheken für Kafka auf dem Markt (z. B. Sarama und Confluent Go Client). segmentio/kafka-go zeichnet sich durch folgende Vorteile aus:

\n
• Reines Go — erfordert keine C-Abhängigkeiten (im Gegensatz zu Confluent), was Build und Deployment vereinfacht.
\n
• Unterstützung von Kafka Protocol v2 — effiziente Komprimierung, Batching und Arbeit mit Transaktionen.
\n
• Einfache API — intuitive Schnittstellen für Producer, Consumer und Topic-Verwaltung.
\n
• Aktive Community — regelmäßige Updates und gute Dokumentation.
\n

Installation der Bibliothek

Fügen Sie die Bibliothek zu Ihrem Projekt hinzu, indem Sie go get verwenden:

go get github.com/segmentio/kafka-go

Stellen Sie sicher, dass eine Kafka-Instanz läuft (lokal oder über Docker). Beispiel einer minimalen Docker Compose-Konfiguration:

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

Erstellen eines Producers

Der Producer sendet Nachrichten an ein Kafka-Topic. Betrachten wir ein einfaches Beispiel:

package main\n\nimport (\n    "context"\n    "fmt"\n    "log"\n    "time"\n\n    "github.com/segmentio/kafka-go"\n)\n\nfunc main() {\n    // Producer-Konfiguration\n    writer := &kafka.Writer{\n        Addr:     kafka.TCP("localhost:9092"),\n        Topic:    "test-topic",\n        Balancer: &kafka.LeastBytes{}, // Lastverteilung nach Nachrichtengröße\n        BatchSize: 10,                 // Batch von 10 Nachrichten senden\n        BatchTimeout: 100 * time.Millisecond,\n    }\n    defer writer.Close()\n\n    // Nachricht senden\n    err := writer.WriteMessages(context.Background(),\n        kafka.Message{\n            Key:   []byte("key-1"),\n            Value: []byte("Hallo, Kafka von Go!"),\n        },\n    )\n    if err != nil {\n        log.Fatal("Fehler beim Senden der Nachricht:", err)\n    }\n\n    fmt.Println("Nachricht erfolgreich gesendet")\n}

Wichtige Parameter von kafka.Writer:

\n
• Addr — Adresse des Kafka-Brokers.
\n
• Topic — Name des Topics.
\n
• Balancer — Strategie zur Verteilung der Nachrichten auf Partitionen (LeastBytes, RoundRobin, Hash).
\n
• BatchSize und BatchTimeout — Optimierung der Leistung.
\n

Erstellen eines Consumers

Der Consumer liest Nachrichten aus einem Topic. Einfachstes Beispiel mit manueller Offset-Verwaltung:

package main\n\nimport (\n    "context"\n    "fmt"\n    "log"\n\n    "github.com/segmentio/kafka-go"\n)\n\nfunc main() {\n    // Reader (Consumer)-Konfiguration\n    reader := kafka.NewReader(kafka.ReaderConfig{\n        Brokers:   []string{"localhost:9092"},\n        Topic:     "test-topic",\n        GroupID:   "my-consumer-group",\n        MinBytes:  10, // Minimale Batch-Größe\n        MaxBytes:  10e6, // 10 MB\n        StartOffset: kafka.FirstOffset, // Beginn