Arbeiten mit Apache Kafka in Go (Golang): Bibliotheken und Beispiele

Arbeiten mit Apache Kafka in Go (Golang): Bibliotheken, Beispiele und bewährte Praktiken

Apache Kafka ist eine verteilte Plattform für die Datenstromverarbeitung, die sich zum De-facto-Standard für den Aufbau hochlastiger Nachrichtensysteme entwickelt hat. Im Go-Ökosystem (Golang) gibt es mehrere ausgereifte Bibliotheken für die Interaktion mit Kafka. In diesem Artikel betrachten wir die beiden beliebtesten: confluent-kafka-go (offizieller Client von Confluent) und sarama (reiner Go-Client). Sie erfahren, wie Sie einen Producer und Consumer einrichten, Fehler behandeln und das passende Werkzeug für Ihr Projekt auswählen.

Warum Kafka und Go ein hervorragendes Paar sind

Go wurde entwickelt, um schnelle, nebenläufige und zuverlässige Serveranwendungen zu erstellen. Kafka wiederum erfordert hohen Durchsatz und minimale Latenz. Go-Goroutinen und Channels eignen sich ideal für die asynchrone Verarbeitung von Nachrichtenströmen. Darüber hinaus hilft die statische Typisierung von Go, viele Fehler bereits zur Kompilierzeit zu vermeiden, was für Produktionssysteme entscheidend ist.

1. Bibliothek confluent-kafka-go

Dies ist ein Wrapper um den C-Client librdkafka, der maximale Leistung und vollständige Unterstützung aller Kafka-Funktionen (Transaktionen, Idempotenz, Sicherheitsprotokolle) bietet. Empfohlen für hochlastige Systeme.

Installation

Erfordert die Installation von librdkafka im System:

# macOSbrew install librdkafka

# Ubuntu/Debiansudo apt-get install librdkafka-dev

# Anschließend Installation des Go-Paketsgo get -u github.com/confluentinc/confluent-kafka-go/kafka

Beispiel für einen Producer

Sendet eine Nachricht an das Topic 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()

    // Kanal für Zustellungsbenachrichtigungen    go func() {        for e := range p.Events() {            switch ev := e.(type) {            case *kafka.Message:                if ev.TopicPartition.Error != nil {                    fmt.Printf("Zustellungsfehler: %v", ev.TopicPartition.Error)                } else {                    fmt.Printf("Nachricht zugestellt an %s [%d] offset %d",                        *ev.TopicPartition.Topic, ev.TopicPartition.Partition, ev.TopicPartition.Offset)                }            }        }    }()

    topic := "orders"    message := []byte("Neue Bestellung #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("Sendefehler: %v", err)    }

    // Warten auf das Senden aller Nachrichten    p.Flush(15 * 1000)}

Beispiel für einen Consumer

Liest Nachrichten aus dem Topic orders in der Gruppe 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) // unendliches Warten        if err != nil {            fmt.Printf("Empfangsfehler: %v", err)            continue        }        fmt.Printf("Nachricht empfangen: %s (Schlüssel: %s) aus %s [%d] of