Sarama für Go: Vollständiges Tutorial mit Beispielen

Was ist Sarama und warum wird es benötigt?

Sarama ist eine leistungsstarke, reine Go-Bibliothek für die Arbeit mit Apache Kafka. Sie bietet einen vollständigen Werkzeugsatz für die Interaktion mit einem Kafka-Cluster, darunter Producer zum Senden von Nachrichten, Consumer zum Lesen von Nachrichten sowie administrative Funktionen zur Verwaltung von Topics, Partitionen und Konfigurationen.

Apache Kafka ist eine verteilte Daten-Streaming-Plattform, die für den Aufbau von Echtzeit-Datenpipelines, Nachrichtensystemen, Ereignisverarbeitung und Protokollspeicherung verwendet wird. Sarama ermöglicht Go-Anwendungen eine einfache Integration mit Kafka und gewährleistet eine zuverlässige und effiziente Datenübertragung.

Warum Sarama verwenden?

• Reines Go: Sarama ist vollständig in Go ohne externe C/C++-Abhängigkeiten geschrieben, was die Kompilierung und Bereitstellung vereinfacht.
• Hohe Leistung: Die Bibliothek ist für die Verarbeitung großer Datenmengen optimiert und unterstützt asynchrones Senden von Nachrichten.
• Flexibilität: Unterstützt verschiedene Betriebsmodi für Producer (synchron, asynchron) und Consumer (einfach, Consumergruppe).
• Aktive Community: Sarama ist eine der beliebtesten Kafka-Bibliotheken für Go mit vielen Mitwirkenden und umfangreicher Dokumentation.
• Volle Kontrolle: Gibt dem Entwickler die vollständige Kontrolle über die Konfiguration, einschließlich Offset-Verwaltung, Serialisierung/Deserialisierung und Fehlerbehandlung.

Installation

Die Installation von Sarama in einem Go-Projekt erfolgt auf dem Standardweg über den Paketmanager go get.

go get github.com/IBM/sarama

Nach Ausführung dieses Befehls wird die Bibliothek zu Ihrer go.mod-Datei hinzugefügt, und Sie können sie in Ihrem Code importieren.

Wichtig: Stellen Sie sicher, dass Go Version 1.18 oder höher installiert ist, da Sarama moderne Sprachfeatures nutzt.

Schnellstart: Senden und Empfangen von Nachrichten

Schreiben wir ein einfaches Beispiel, das eine Nachricht an ein Kafka-Topic sendet und sie wieder ausliest.

1. Producer — Nachricht senden

package main

import (
    "fmt"
    "log"

    "github.com/IBM/sarama"
)

func main() {
    // 1. Producer-Konfiguration
    config := sarama.NewConfig()
    config.Producer.Return.Successes = true // Für synchronen Modus erforderlich
    config.Producer.RequiredAcks = sarama.WaitForAll // Warte auf Bestätigung aller Replikate

    // 2. Synchronen Producer erstellen
    producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
    if err != nil {
        log.Fatalf("Producer konnte nicht erstellt werden: %v", err)
    }
    defer producer.Close()

    // 3. Nachricht erstellen
    msg := &sarama.ProducerMessage{
        Topic: "my-topic",
        Key:   sarama.StringEncoder("my-key"),
        Value: sarama.StringEncoder("Hallo, Kafka!"),
    }

    // 4. Nachricht senden
    partition, offset, err := producer.SendMessage(msg)
    if err != nil {
        log.Fatalf("Nachricht konnte nicht gesendet werden: %v", err)
    }

    fmt.Printf("Nachricht gesendet an Partition %d mit Offset %d\n", partition, offset)
}

2. Consumer — Nachricht empfangen

package main

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

    "github.com/IBM/sarama"
)

func main() {
    // 1. Consumer-Konfiguration
    config := sarama.NewConfig()
    config.Consumer.Return.Errors = true
    config.Consumer.Offsets.Initial = sarama.OffsetOldest // Von Anfang an lesen

    // 2. Consumer erstellen
    consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, config)
    if err != nil {
        log.Fatalf("Consumer konnte nicht erstellt werden: %v", err)
    }
    defer consumer.Close()

    // 3. Auf Partition abonnieren
    parti