ApsaraMQ for Kafka:Gunakan SDK untuk C# untuk mengirim dan menerima pesan

Persyaratan lingkungan

.NET telah diinstal. Untuk informasi lebih lanjut, lihat Unduh .NET.

Instal pustaka C#

dotnet add package -v 1.5.2 Confluent.Kafka

Buat file konfigurasi

1. (Opsional) Unduh sertifikat root Secure Sockets Layer (SSL). Jika Anda menggunakan titik akhir SSL untuk terhubung ke instance ApsaraMQ for Kafka, Anda harus menginstal sertifikat tersebut.

2. Konfigurasikan file producer.cs dan consumer.cs.

   Tabel 1. Parameter

   Parameter	Deskripsi
   BootstrapServers	Titik akhir SSL dari instance ApsaraMQ for Kafka. Anda dapat memperoleh titik akhir di bagian Endpoint Information pada halaman Instance Details di Konsol ApsaraMQ for Kafka.
   SslCaLocation	Lokasi sertifikat root SSL yang Anda unduh. Parameter ini hanya diperlukan jika Anda menggunakan titik akhir SSL untuk terhubung ke instance ApsaraMQ for Kafka.
   SaslMechanism	Mekanisme keamanan yang ingin Anda gunakan untuk mengirim dan menerima pesan. Jika Anda menggunakan titik akhir SSL untuk terhubung ke instance ApsaraMQ for Kafka, atur parameter ini ke SaslMechanism.Plain. Jika Anda menggunakan titik akhir Simple Authentication and Security Layer (SASL) untuk terhubung ke instance ApsaraMQ for Kafka, atur parameter ini ke SaslMechanism.Plain untuk menentukan mekanisme PLAIN atau atur parameter ini ke SaslMechanism.ScramSha256 untuk menentukan Salted Challenge Response Authentication Mechanism (SCRAM).
   SecurityProtocol	Protokol keamanan yang ingin Anda gunakan untuk mengirim dan menerima pesan. Jika Anda menggunakan titik akhir SSL untuk terhubung ke instance ApsaraMQ for Kafka, atur parameter ini ke SecurityProtocol.SaslSsl. Jika Anda menggunakan titik akhir SASL untuk terhubung ke instance ApsaraMQ for Kafka, atur parameter ini ke SecurityProtocol.SaslPlaintext jika mekanisme PLAIN digunakan, atau atur parameter ini ke SecurityProtocol.SaslPlaintext jika mekanisme SCRAM digunakan.
   SaslUsername	Nama pengguna SASL. Jika Anda menggunakan titik akhir default untuk terhubung ke instance ApsaraMQ for Kafka, parameter ini tidak tersedia. Catatan Jika fitur ACL tidak diaktifkan untuk instance ApsaraMQ for Kafka, Anda dapat memperoleh nama pengguna dan kata sandi pengguna SASL dari parameter Username dan Password di bagian Configuration Information pada halaman Instance Details di Konsol ApsaraMQ for Kafka. Jika fitur ACL diaktifkan untuk instance ApsaraMQ for Kafka, pastikan bahwa pengguna SASL memiliki otorisasi untuk mengirim dan menerima pesan menggunakan instance tersebut. Untuk informasi lebih lanjut, lihat Berikan izin kepada pengguna SASL.
   SaslPassword	Kata sandi pengguna SASL. Jika Anda menggunakan titik akhir default untuk terhubung ke instance ApsaraMQ for Kafka, parameter ini tidak tersedia.
   topic	Nama topik. Anda dapat memperoleh nama topik di halaman Topics di Konsol ApsaraMQ for Kafka.
   GroupId	ID grup. Anda dapat memperoleh ID grup di halaman Groups di Konsol ApsaraMQ for Kafka.

Kirim pesan

Jalankan perintah berikut untuk menjalankan producer.cs guna mengirim pesan:

dotnet run producer.cs

Kode contoh berikut memberikan contoh 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
                ? $"Pesan dikirim ke {r.TopicPartitionOffset}"
                : $"Kesalahan Pengiriman: {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));
        }
    }
}

Terima pesan

Jalankan perintah berikut untuk menjalankan consumer.cs guna menerima pesan:

dotnet run consumer.cs

Kode contoh berikut memberikan contoh 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($"Pesan diterima '{cr.Value}' pada: '{cr.TopicPartitionOffset}'.");
                    }
                    catch (ConsumeException e)
                    {
                        Console.WriteLine($"Kesalahan terjadi: {e.Error.Reason}");
                    }
                }
            }
            catch (OperationCanceledException)
            {
                c.Close();
            }
        }
    }
}