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

C# (.NET) में Apache Kafka के लिए Confluent.Kafka लाइब्रेरी

Apache Kafka एक डिस्ट्रीब्यूटेड डेटा स्ट्रीमिंग प्लेटफॉर्म है, जो इवेंट-ड्रिवन आर्किटेक्चर और माइक्रोसर्विसेज बनाने का मानक बन गया है। C# (.NET) डेवलपर्स के लिए सबसे लोकप्रिय और सपोर्टेड लाइब्रेरी Confluent.Kafka है - यह Confluent कंपनी का आधिकारिक क्लाइंट है, जो हाई-परफॉरमेंस C-क्लाइंट librdkafka पर बनाया गया है। इस लेख में हम विस्तार से जानेंगे कि लाइब्रेरी कैसे इंस्टॉल करें, प्रोड्यूसर और कंज्यूमर कैसे बनाएं, और बेस्ट प्रैक्टिसेज क्या हैं।

Confluent.Kafka इंस्टॉलेशन

शुरू करने के लिए, NuGet के माध्यम से अपने प्रोजेक्ट में पैकेज जोड़ें:

dotnet add package Confluent.Kafka

या पैकेज मैनेजर कंसोल से:

Install-Package Confluent.Kafka

प्रोड्यूसर (Producer) की बेसिक सेटअप

प्रोड्यूसर Kafka टॉपिक्स में मैसेज भेजता है। एक सरल उदाहरण देखें:

using Confluent.Kafka;

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

using (var producer = new ProducerBuilder<string, string>(config).Build())
{
    var message = new Message<string, string>
    {
        Key = "user_123",
        Value = "नमस्ते, Kafka!"
    };

    var deliveryReport = await producer.ProduceAsync("my-topic", message);
    Console.WriteLine($"मैसेज भेजा गया। Partition: {deliveryReport.Partition}, Offset: {deliveryReport.Offset}");
}

मुख्य बिंदु:

• BootstrapServers — एक या अधिक Kafka ब्रोकर्स का पता।
• ProduceAsync — एसिंक्रोनस भेजना जो कन्फर्मेशन का इंतजार करता है।
• Key एक ही की के लिए मैसेज की ऑर्डरिंग सुनिश्चित करता है।

जटिल ऑब्जेक्ट्स के लिए प्रोड्यूसर

अक्सर स्ट्रिंग के बजाय JSON या बाइनरी डेटा भेजना होता है। कस्टम सीरियलाइज़र या Newtonsoft.Json का उपयोग करें:

using System.Text.Json;
using Confluent.Kafka;

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

var producer = new ProducerBuilder<string, Order>(config)
    .SetValueSerializer(new JsonSerializer<Order>())
    .Build();

var order = new Order { Id = 1, Amount = 150.50m, Customer = "राहुल" };
await producer.ProduceAsync("orders", new Message<string, Order> { Key = order.Id.ToString(), Value = order });

public class Order
{
    public int Id { get; set; }
    public decimal Amount { get; set; }
    public string Customer { get; set; }
}

public class JsonSerializer<T> : ISerializer<T>
{
    public byte[] Serialize(T data, SerializationContext context) =>
        JsonSerializer.SerializeToUtf8Bytes(data);
}

कंज्यूमर (Consumer) की सेटअप

कंज्यूमर एक या अधिक टॉपिक्स से मैसेज पढ़ता है। सब्सक्राइब और प्रोसेस करने का उदाहरण:

using Confluent.Kafka;

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

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}");
        }
    }
    catch (OperationCanceledException)
    {
        consumer.Close();
    }
}

महत्वपूर्ण कॉन्फ़िगरेशन पैरामीटर:

• GroupId — कंज्यूमर ग्रुप का आइडेंटिफायर। Kafka पार्टीशन को ग्रुप के सदस्यों के बीच बांटता है।
• AutoOffsetReset — यह निर्धारित करता है कि कहां से पढ़ना शुरू करना है