C# (.NET) में Kafka के साथ कार्य: Confluent.Kafka लाइब्रेरी

C# (.NET) में Kafka के साथ कार्य: Confluent.Kafka लाइब्रेरी

Apache Kafka एक वितरित डेटा स्ट्रीमिंग प्लेटफ़ॉर्म है जिसका व्यापक रूप से उच्च-लोड मैसेजिंग सिस्टम, इवेंट प्रोसेसिंग और माइक्रोसर्विसेज इंटीग्रेशन के निर्माण के लिए उपयोग किया जाता है। .NET इकोसिस्टम में Kafka के साथ कार्य करने के लिए मुख्य और सबसे अधिक समर्थित लाइब्रेरी Confluent.Kafka है - Confluent का आधिकारिक क्लाइंट, जो उच्च-प्रदर्शन C लाइब्रेरी librdkafka पर आधारित है।

इस लेख में, हम Confluent.Kafka की प्रमुख क्षमताओं, इंस्टॉलेशन, प्रोड्यूसर और कंज्यूमर कॉन्फ़िगरेशन, त्रुटि प्रबंधन और C# में व्यावहारिक कोड उदाहरणों पर चर्चा करेंगे।

इंस्टॉलेशन और कनेक्शन

लाइब्रेरी NuGet के माध्यम से वितरित की जाती है। इंस्टॉल करने के लिए, कमांड चलाएँ:

dotnet add package Confluent.Kafka

इंस्टॉलेशन के बाद, अपने कोड में using निर्देश जोड़ें:

using Confluent.Kafka;using Confluent.Kafka.Admin;using System;using System.Threading;using System.Threading.Tasks;

मूल अवधारणाएँ

• Producer (प्रोड्यूसर) — Kafka टॉपिक्स में संदेश भेजता है।
• Consumer (कंज्यूमर) — टॉपिक्स से संदेश पढ़ता है।
• Topic (टॉपिक) — नामित चैनल जहाँ संदेश प्रकाशित होते हैं।
• Broker (ब्रोकर) — Kafka सर्वर जो डेटा संग्रहीत करता है।
• Partition (पार्टीशन) — टॉपिक के अंदर समानांतरता की इकाई।

प्रोड्यूसर बनाना (Producer)

प्रोड्यूसर Kafka को संदेश भेजता है। एक बुनियादी उदाहरण देखें:

var config = new ProducerConfig{    BootstrapServers = "localhost:9092"};

using (var producer = new ProducerBuilder<string, string>(config).Build()){    try    {        var deliveryResult = await producer.ProduceAsync(            "my-topic",            new Message<string, string> { Key = "key1", Value = "Hello Kafka from .NET!" }        );

        Console.WriteLine($"पार्टीशन [{deliveryResult.Partition}] पर ऑफ़सेट [{deliveryResult.Offset}] पर भेजा गया");    }    catch (ProduceException<string, string> e)    {        Console.WriteLine($"भेजने में त्रुटि: {e.Error.Reason}");    }}

इस उदाहरण में:

• BootstrapServers — Kafka क्लस्टर का पता।
• ProduceAsync — संदेश का एसिंक्रोनस भेजना।
• Key — पार्टीशन निर्धारित करने के लिए वैकल्पिक कुंजी।
• Value — संदेश का मुख्य भाग (JSON, स्ट्रिंग, बाइनरी डेटा हो सकता है)।

कंज्यूमर बनाना (Consumer)

कंज्यूमर टॉपिक्स की सदस्यता लेता है और संदेशों को संसाधित करता है। एक महत्वपूर्ण पैरामीटर GroupId है, जो उपभोक्ता समूह को परिभाषित करता है:

var config = new ConsumerConfig{    BootstrapServers = "localhost:9092",    GroupId = "my-consumer-group",    AutoOffsetReset = AutoOffsetReset.Earliest,    EnableAutoCommit = false};

using (var consumer = new ConsumerBuilder<string, string>(config).Build()){    consumer.Subscribe("my-topic");

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

    try    {        while (true)        {            var consumeResult = consumer.Consume(cts.Token);

            Console.WriteLine($"संदेश प्राप्त हुआ: Key = {consumeResult.Message.Key}, Value = {consumeResult.Message.Value}");

            // मैन्युअल पुष्टिकरण (commit)            consumer.Commit(consumeResult);        }    }    catch (OperationCanceledException)    {        Console.WriteLine("उपभोग रोक दिया गया।");    }    finally    {        consumer.Close();    }}

मुख्य बिंदु:

• AutoOff