Travailler avec Apache Kafka en Go (Golang) : bibliothèques et exemples

Travailler avec Apache Kafka en Go (Golang) : bibliothèques, exemples et bonnes pratiques

Apache Kafka est une plateforme distribuée de streaming de données, devenue le standard de facto pour la construction de systèmes d'échange de messages à haute charge. Dans l'écosystème Go (Golang), il existe plusieurs bibliothèques matures pour interagir avec Kafka. Dans cet article, nous allons examiner les deux plus populaires : confluent-kafka-go (le client officiel de Confluent) et sarama (un client Go pur). Vous apprendrez à configurer un producteur et un consommateur, à gérer les erreurs et à choisir l'outil adapté à votre projet.

Pourquoi Kafka et Go forment-ils une excellente paire ?

Go a été conçu pour créer des applications serveur rapides, concurrentes et fiables. Kafka, quant à lui, exige un débit élevé et une latence minimale. Les goroutines et les canaux de Go sont parfaitement adaptés au traitement asynchrone des flux de messages. De plus, le typage statique de Go permet d'éviter de nombreuses erreurs dès la compilation, ce qui est crucial pour les systèmes de production.

1. Bibliothèque confluent-kafka-go

Il s'agit d'une surcouche du client C librdkafka, offrant des performances maximales et un support complet de toutes les fonctionnalités de Kafka (transactions, idempotence, protocoles de sécurité). Recommandée pour les systèmes à haute charge.

Installation

Nécessite la présence de librdkafka sur le système :

# macOSbrew install librdkafka

# Ubuntu/Debiansudo apt-get install librdkafka-dev

# Ensuite, installation du package Gogo get -u github.com/confluentinc/confluent-kafka-go/kafka

Exemple de producteur (Producer)

Envoie un message dans le 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()

    // Canal pour les notifications de livraison    go func() {        for e := range p.Events() {            switch ev := e.(type) {            case *kafka.Message:                if ev.TopicPartition.Error != nil {                    fmt.Printf("Erreur de livraison : %v", ev.TopicPartition.Error)                } else {                    fmt.Printf("Message livré dans %s [%d] offset %d",                        *ev.TopicPartition.Topic, ev.TopicPartition.Partition, ev.TopicPartition.Offset)                }            }        }    }()

    topic := "orders"    message := []byte("Nouvelle commande #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("Erreur d'envoi : %v", err)    }

    // Attendre l'envoi de tous les messages    p.Flush(15 * 1000)}

Exemple de consommateur (Consumer)

Lit les messages du topic orders dans le groupe 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) // attente infinie        if err != nil {            fmt.Printf("Erreur de réception : %v", err)            continue        }        fmt.Printf("Message reçu : %s (clé : %s) depuis %s [%d] of