Utiliser le SDK Go pour envoyer et s'abonner à des messages - ApsaraMQ for Kafka

Utilisez le SDK Go pour vous connecter à une instance Message Queue for Apache Kafka via un endpoint, puis envoyez et abonnez-vous aux messages.

Prérequis

Installez Go. Pour plus d'informations, consultez la page Installer Go.

Remarque

Le projet kafka-confluent-go-demo ne prend pas en charge Windows.

Préparer la configuration

Facultatif : Téléchargez le certificat racine SSL. Ce certificat est requis pour se connecter à l'instance via un endpoint SSL.
Accédez au dépôt aliware-kafka-demos, cliquez sur l'icône , puis sélectionnez Download ZIP dans la liste déroulante pour télécharger le projet de démonstration. Une fois le téléchargement terminé, décompressez le fichier.
Dans le projet de démonstration décompressé, localisez le dossier kafka-confluent-go-demo et transférez-le vers le répertoire /home de votre système Linux.

Connectez-vous à votre système Linux, accédez au répertoire /home/kafka-confluent-go-demo et modifiez le fichier de configuration conf/kafka.json.

{
  "topic": "XXX",
  "group.id": "XXX",
  "bootstrap.servers" : "XXX:XX,XXX:XX,XXX:XX",
  "security.protocol" : "plaintext",
  "sasl.mechanism" : "XXX",
  "sasl.username" : "XXX",
  "sasl.password" : "XXX"
}

Paramètre	Description
topic	Nom du topic de votre instance. Vous trouverez ce nom sur la page Topics de la Message Queue for Apache Kafka console.
group.id	ID du group de consommateurs de votre instance. Vous trouverez cet ID sur la page Groups de la Message Queue for Apache Kafka console. Remarque Ce paramètre est facultatif pour l'envoi de messages avec producer.go, mais obligatoire pour l'abonnement aux messages avec consumer.go.
bootstrap.servers	Adresse IP et port de l'endpoint. Ces informations figurent dans la section Endpoint Information de la page Instance Details de la Message Queue for Apache Kafka console.
security.protocol	Protocole d'authentification utilisateur SASL (Simple Authentication and Security Layer). La valeur par défaut est plaintext. La valeur dépend du type d'endpoint : Endpoint par défaut : plaintext. Endpoint SSL : sasl_ssl. Endpoint SASL : sasl_plaintext.
sasl.mechanism	Mécanisme de sécurité utilisé pour l'envoi et l'abonnement aux messages. La valeur dépend du type d'endpoint : Endpoint par défaut : ce paramètre n'est pas requis. Endpoint SSL : PLAIN. Endpoint SASL : pour le mécanisme PLAIN, définissez ce paramètre sur PLAIN. Pour le mécanisme SCRAM, définissez-le sur SCRAM-SHA-256.
sasl.username	Nom d'utilisateur SASL. Ce paramètre est requis si vous utilisez un endpoint SSL ou SASL. Remarque Si la fonctionnalité ACL n'est pas activée pour l'instance ApsaraMQ for Kafka, récupérez le nom d'utilisateur et le mot de passe de l'utilisateur SASL depuis les paramètres Username et Password de la section Configuration Information de la page Instance Details dans la ApsaraMQ for Kafka console. Si la fonctionnalité ACL est activée pour l'instance ApsaraMQ for Kafka, assurez-vous que l'utilisateur SASL est autorisé à envoyer et recevoir des messages via l'instance. Pour plus d'informations, consultez la rubrique Accorder des autorisations aux utilisateurs SASL.
sasl.password	Mot de passe SASL. Ce paramètre est requis si vous utilisez un endpoint SSL ou SASL.

Envoyer des messages

Exécutez la commande suivante pour envoyer des messages à l'aide du fichier producer.go.

go run -mod=vendor producer/producer.go

Exemple de fichier producer.go :

package main

import (
    "encoding/json"
    "fmt"
    "github.com/confluentinc/confluent-kafka-go/kafka"
    "os"
    "path/filepath"
)

type KafkaConfig struct {
    Topic      string `json:"topic"`
    GroupId    string `json:"group.id"`
    BootstrapServers    string `json:"bootstrap.servers"`
    SecurityProtocol string `json:"security.protocol"`
    SslCaLocation string `json:"ssl.ca.location"`
    SaslMechanism string `json:"sasl.mechanism"`
    SaslUsername string `json:"sasl.username"`
    SaslPassword string `json:"sasl.password"`
}

// The config parameter must be a pointer to a struct; otherwise, the function panics.
func loadJsonConfig() *KafkaConfig {
    workPath, err := os.Getwd()
    if err != nil {
        panic(err)
    }
    configPath := filepath.Join(workPath, "conf")
    fullPath := filepath.Join(configPath, "kafka.json")
    file, err := os.Open(fullPath);
    if (err != nil) {
        msg := fmt.Sprintf("Can not load config at %s. Error: %v", fullPath, err)
        panic(msg)
    }

    defer file.Close()

    decoder := json.NewDecoder(file)
    var config = &KafkaConfig{}
    err = decoder.Decode(config);
    if (err != nil) {
        msg := fmt.Sprintf("Decode json fail for config file at %s. Error: %v", fullPath, err)
        panic(msg)
    }
    json.Marshal(config)
    return  config
}

func doInitProducer(cfg *KafkaConfig) *kafka.Producer {
    fmt.Print("init kafka producer, it may take a few seconds to init the connection\n")
    //common arguments
    var kafkaconf = &kafka.ConfigMap{
        "api.version.request": "true",
        "message.max.bytes": 1000000,
        "linger.ms": 10,
        "retries": 30,
        "retry.backoff.ms": 1000,
        "acks": "1"}
    kafkaconf.SetKey("bootstrap.servers", cfg.BootstrapServers)

    switch cfg.SecurityProtocol {
        case "plaintext" :
            kafkaconf.SetKey("security.protocol", "plaintext");
        case "sasl_ssl":
            kafkaconf.SetKey("security.protocol", "sasl_ssl");
            kafkaconf.SetKey("ssl.ca.location", "conf/ca-cert.pem");
            kafkaconf.SetKey("sasl.username", cfg.SaslUsername);
            kafkaconf.SetKey("sasl.password", cfg.SaslPassword);
            kafkaconf.SetKey("sasl.mechanism", cfg.SaslMechanism);
            kafkaconf.SetKey("enable.ssl.certificate.verification", "false");
            kafkaconf.SetKey("ssl.endpoint.identification.algorithm", "None")
    case "sasl_plaintext":
            kafkaconf.SetKey("sasl.mechanism", "PLAIN")
            kafkaconf.SetKey("security.protocol", "sasl_plaintext");
            kafkaconf.SetKey("sasl.username", cfg.SaslUsername);
            kafkaconf.SetKey("sasl.password", cfg.SaslPassword);
            kafkaconf.SetKey("sasl.mechanism", cfg.SaslMechanism)
    default:
            panic(kafka.NewError(kafka.ErrUnknownProtocol, "unknown protocol", true))
    }

    producer, err := kafka.NewProducer(kafkaconf)
    if err != nil {
        panic(err)
    }
    fmt.Print("init kafka producer success\n")
    return producer;
}

func main() {
    // Choose the correct protocol
    // 9092 for PLAINTEXT
    // 9093 for SASL_SSL, need to provide sasl.username and sasl.password
    // 9094 for SASL_PLAINTEXT, need to provide sasl.username and sasl.password
    cfg := loadJsonConfig();
    producer := doInitProducer(cfg)

    defer producer.Close()

    // Delivery report handler for produced messages
    go func() {
        for e := range producer.Events() {
            switch ev := e.(type) {
            case *kafka.Message:
                if ev.TopicPartition.Error != nil {
                    fmt.Printf("Delivery failed: %v\n", ev.TopicPartition)
                } else {
                    fmt.Printf("Delivered message to %v\n", ev.TopicPartition)
                }
            }
        }
    }()

    // Produce messages to topic (asynchronously)
    topic := cfg.Topic
    for _, word := range []string{"Welcome", "to", "the", "Confluent", "Kafka", "Golang", "client"} {
        producer.Produce(&kafka.Message{
            TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
            Value:          []byte(word),
        }, nil)
    }

    // Wait for message deliveries before shutting down
    producer.Flush(15 * 1000)
}

S'abonner à des messages

Exécutez la commande suivante pour vous abonner aux messages à l'aide du fichier consumer.go.

go run -mod=vendor consumer/consumer.go

Exemple de fichier consumer.go :

package main

import (
    "encoding/json"
    "fmt"
    "github.com/confluentinc/confluent-kafka-go/kafka"
    "os"
    "path/filepath"
)
type KafkaConfig struct {
    Topic      string `json:"topic"`
    GroupId    string `json:"group.id"`
    BootstrapServers    string `json:"bootstrap.servers"`
    SecurityProtocol string `json:"security.protocol"`
    SaslMechanism string `json:"sasl.mechanism"`
    SaslUsername string `json:"sasl.username"`
    SaslPassword string `json:"sasl.password"`
}

// The config parameter must be a pointer to a struct; otherwise, the function panics.
func loadJsonConfig() *KafkaConfig {
    workPath, err := os.Getwd()
    if err != nil {
        panic(err)
    }
    configPath := filepath.Join(workPath, "conf")
    fullPath := filepath.Join(configPath, "kafka.json")
    file, err := os.Open(fullPath);
    if (err != nil) {
        msg := fmt.Sprintf("Can not load config at %s. Error: %v", fullPath, err)
        panic(msg)
    }

    defer file.Close()

    decoder := json.NewDecoder(file)
    var config = &KafkaConfig{}
    err = decoder.Decode(config);
    if (err != nil) {
        msg := fmt.Sprintf("Decode json fail for config file at %s. Error: %v", fullPath, err)
        panic(msg)
    }
    json.Marshal(config)
    return  config
}

func doInitConsumer(cfg *KafkaConfig) *kafka.Consumer {
    fmt.Print("init kafka consumer, it may take a few seconds to init the connection\n")
    //common arguments
    var kafkaconf = &kafka.ConfigMap{
        "api.version.request": "true",
        "auto.offset.reset": "latest",
        "heartbeat.interval.ms": 3000,
        "session.timeout.ms": 30000,
        "max.poll.interval.ms": 120000,
        "fetch.max.bytes": 1024000,
        "max.partition.fetch.bytes": 256000}
    kafkaconf.SetKey("bootstrap.servers", cfg.BootstrapServers);
    kafkaconf.SetKey("group.id", cfg.GroupId)

    switch cfg.SecurityProtocol {
    case "plaintext" :
        kafkaconf.SetKey("security.protocol", "plaintext");
    case "sasl_ssl":
        kafkaconf.SetKey("security.protocol", "sasl_ssl");
        kafkaconf.SetKey("ssl.ca.location", "./conf/ca-cert.pem");
        kafkaconf.SetKey("sasl.username", cfg.SaslUsername);
        kafkaconf.SetKey("sasl.password", cfg.SaslPassword);
        kafkaconf.SetKey("sasl.mechanism", cfg.SaslMechanism);
        kafkaconf.SetKey("ssl.endpoint.identification.algorithm", "None");
        kafkaconf.SetKey("enable.ssl.certificate.verification", "false")
    case "sasl_plaintext":
        kafkaconf.SetKey("security.protocol", "sasl_plaintext");
        kafkaconf.SetKey("sasl.username", cfg.SaslUsername);
        kafkaconf.SetKey("sasl.password", cfg.SaslPassword);
        kafkaconf.SetKey("sasl.mechanism", cfg.SaslMechanism)

    default:
        panic(kafka.NewError(kafka.ErrUnknownProtocol, "unknown protocol", true))
    }

    consumer, err := kafka.NewConsumer(kafkaconf)
    if err != nil {
        panic(err)
    }
    fmt.Print("init kafka consumer success\n")
    return consumer;
}

func main() {

    // Choose the correct protocol
    // 9092 for PLAINTEXT
    // 9093 for SASL_SSL, need to provide sasl.username and sasl.password
    // 9094 for SASL_PLAINTEXT, need to provide sasl.username and sasl.password
    cfg := loadJsonConfig();
    consumer := doInitConsumer(cfg)

    consumer.SubscribeTopics([]string{cfg.Topic}, nil)

    for {
        msg, err := consumer.ReadMessage(-1)
        if err == nil {
            fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value))
        } else {
            // The client will automatically try to recover from all errors.
            fmt.Printf("Consumer error: %v (%v)\n", err, msg)
        }
    }

    consumer.Close()
}