ApsaraMQ for Kafka:Utiliser le SDK pour C# pour envoyer et recevoir des messages

Prérequis

.NET est installé. Pour plus d'informations, consultez la page Install .NET.

Installer la bibliothèque de dépendances C#

dotnet add package -v 1.5.2 Confluent.Kafka

Créer les fichiers de configuration

1. (Facultatif) Téléchargez le certificat racine SSL (Secure Sockets Layer). Si vous utilisez l'endpoint SSL pour vous connecter à votre instance ApsaraMQ for Kafka, vous devez installer ce certificat.

2. Configurez les fichiers producer.cs et consumer.cs.

   Tableau 1. Paramètres

   Paramètre	Description
   BootstrapServers	L'endpoint SSL de l'instance ApsaraMQ for Kafka. Vous pouvez obtenir cet endpoint dans la section Endpoint Information de la page Instance Details de la ApsaraMQ for Kafka console.
   SslCaLocation	Le chemin d'accès au certificat racine SSL que vous avez téléchargé. Ce paramètre est requis uniquement si vous utilisez l'endpoint SSL pour vous connecter à l'instance ApsaraMQ for Kafka.
   SaslMechanism	Le mécanisme de sécurité à utiliser pour envoyer et recevoir des messages. Si vous utilisez l'endpoint SSL pour vous connecter à l'instance ApsaraMQ for Kafka, définissez ce paramètre sur SaslMechanism.Plain. Si vous utilisez l'endpoint SASL (Simple Authentication and Security Layer) pour vous connecter à l'instance ApsaraMQ for Kafka, définissez ce paramètre sur SaslMechanism.Plain pour spécifier le mécanisme PLAIN ou sur SaslMechanism.ScramSha256 pour spécifier le mécanisme SCRAM (Salted Challenge Response Authentication Mechanism).
   SecurityProtocol	Le protocole de sécurité à utiliser pour envoyer et recevoir des messages. Si vous utilisez l'endpoint SSL pour vous connecter à l'instance ApsaraMQ for Kafka, définissez ce paramètre sur SecurityProtocol.SaslSsl. Si vous utilisez l'endpoint SASL pour vous connecter à l'instance ApsaraMQ for Kafka, définissez ce paramètre sur SecurityProtocol.SaslPlaintext si le mécanisme PLAIN est utilisé, ou sur SecurityProtocol.SaslPlaintext si le mécanisme SCRAM est utilisé.
   SaslUsername	Le nom d'utilisateur de l'utilisateur SASL. Si vous utilisez l'endpoint par défaut pour vous connecter à l'instance ApsaraMQ for Kafka, ce paramètre n'est pas disponible. Remarque Si la fonctionnalité ACL n'est pas activée pour l'instance ApsaraMQ for Kafka, vous pouvez obtenir le nom d'utilisateur et le mot de passe de l'utilisateur SASL à partir des paramètres Username et Password dans la section Configuration Information de la page Instance Details de 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 page Accorder des autorisations aux utilisateurs SASL.
   SaslPassword	Le mot de passe de l'utilisateur SASL. Si vous utilisez l'endpoint par défaut pour vous connecter à l'instance ApsaraMQ for Kafka, ce paramètre n'est pas disponible.
   topic	Le nom du topic. Vous pouvez obtenir le nom du topic sur la page Topics de la ApsaraMQ for Kafka console.
   GroupId	L'ID de groupe. Vous pouvez obtenir l'ID de groupe sur la page Groups de la ApsaraMQ for Kafka console.

Envoyer des messages

Exécutez la commande suivante pour lancer producer.cs et envoyer des messages :

dotnet run producer.cs

L'exemple de code suivant illustre le fichier producer.cs.

using System;
using Confluent.Kafka;

class Producer
{
    public static void Main(string[] args)
    {
        var conf = new ProducerConfig {
            BootstrapServers = "XXX,XXX,XXX",
            SslCaLocation = "XXX/only-4096-ca-cert.pem",
            SaslMechanism = SaslMechanism.Plain,
            SecurityProtocol = SecurityProtocol.SaslSsl,
            SslEndpointIdentificationAlgorithm = SslEndpointIdentificationAlgorithm.None,
            SaslUsername = "XXX",
            SaslPassword = "XXX",
            };

        Action<DeliveryReport<Null, string>> handler = r =>
            Console.WriteLine(!r.Error.IsError
                ? $"Delivered message to {r.TopicPartitionOffset}"
                : $"Delivery Error: {r.Error.Reason}");

        string topic ="XXX";

        using (var p = new ProducerBuilder<Null, string>(conf).Build())
        {
            for (int i=0; i<100; ++i)
            {
                p.Produce(topic, new Message<Null, string> { Value = i.ToString() }, handler);
            }
            p.Flush(TimeSpan.FromSeconds(10));
        }
    }
}

Recevoir des messages

Exécutez la commande suivante pour lancer consumer.cs et recevoir des messages :

dotnet run consumer.cs

L'exemple de code suivant illustre le fichier consumer.cs.

using System;
using System.Threading;
using Confluent.Kafka;

class Consumer
{
    public static void Main(string[] args)
    {
        var conf = new ConsumerConfig {
            GroupId = "XXX",
            BootstrapServers = "XXX,XXX,XXX",
            SslCaLocation = "XXX/only-4096-ca-cert.pem",
            SaslMechanism = SaslMechanism.Plain,
            SslEndpointIdentificationAlgorithm = SslEndpointIdentificationAlgorithm.None,
            SecurityProtocol = SecurityProtocol.SaslSsl,
            SaslUsername = "XXX",
            SaslPassword = "XXX",
            AutoOffsetReset = AutoOffsetReset.Earliest
        };

        string topic = "XXX";

        using (var c = new ConsumerBuilder<Ignore, string>(conf).Build())
        {
            c.Subscribe(topic);

            CancellationTokenSource cts = new CancellationTokenSource();
            Console.CancelKeyPress += (_, e) => {
                e.Cancel = true;
                cts.Cancel();
            };

            try
            {
                while (true)
                {
                    try
                    {
                        var cr = c.Consume(cts.Token);
                        Console.WriteLine($"Consumed message '{cr.Value}' at: '{cr.TopicPartitionOffset}'.");
                    }
                    catch (ConsumeException e)
                    {
                        Console.WriteLine($"Error occured: {e.Error.Reason}");
                    }
                }
            }
            catch (OperationCanceledException)
            {
                c.Close();
            }
        }
    }
}