العمل مع Kafka في C# (.NET) | Confluent.Kafka

مكتبة Confluent.Kafka للعمل مع Apache Kafka في C# (.NET)

\n\n

Apache Kafka هي منصة موزعة لتدفق البيانات أصبحت معيارًا في بناء البنى القائمة على الأحداث والخدمات المصغرة. لمطوري C# (.NET)، المكتبة الأكثر شيوعًا والمدعومة هي Confluent.Kafka - العميل الرسمي من شركة Confluent، المبني على عميل C عالي الأداء librdkafka. في هذه المقالة، سنشرح بالتفصيل كيفية تثبيت المكتبة، إنشاء منتج ومستهلك، بالإضافة إلى أفضل الممارسات.

\n\n

تثبيت Confluent.Kafka

للبدء، أضف الحزمة إلى مشروعك عبر NuGet:

dotnet add package Confluent.Kafka

أو عبر وحدة تحكم مدير الحزم:

Install-Package Confluent.Kafka

\n\n

الإعداد الأساسي للمنتج (Producer)

يرسل المنتج الرسائل إلى مواضيع Kafka. لنلق نظرة على مثال بسيط:

using Confluent.Kafka;\n\nvar config = new ProducerConfig\n{\n    BootstrapServers = "localhost:9092"\n};\n\nusing (var producer = new ProducerBuilder<string, string>(config).Build())\n{\n    var message = new Message<string, string>\n    {\n        Key = "user_123",\n        Value = "مرحبًا، Kafka!"\n    };\n\n    var deliveryReport = await producer.ProduceAsync("my-topic", message);\n    Console.WriteLine($"تم إرسال الرسالة. القسم: {deliveryReport.Partition}, الإزاحة: {deliveryReport.Offset}");\n}

\n\n

النقاط الرئيسية:

BootstrapServers — عنوان واحد أو أكثر من وسطاء Kafka.
ProduceAsync — إرسال غير متزامن مع انتظار التأكيد.
المفتاح (Key) يضمن ترتيب الرسائل لنفس المفتاح.

\n\n

منتج مع تسلسل الكائنات المعقدة

غالبًا ما يكون مطلوبًا إرسال JSON أو بيانات ثنائية بدلاً من النصوص. استخدم أداة تسلسل مخصصة أو مكتبة Newtonsoft.Json:

using System.Text.Json;\nusing Confluent.Kafka;\n\nvar config = new ProducerConfig { BootstrapServers = "localhost:9092" };\n\nvar producer = new ProducerBuilder<string, Order>(config)\n    .SetValueSerializer(new JsonSerializer<Order>())\n    .Build();\n\nvar order = new Order { Id = 1, Amount = 150.50m, Customer = "إيفان" };\nawait producer.ProduceAsync("orders", new Message<string, Order> { Key = order.Id.ToString(), Value = order });\n\npublic class Order\n{\n    public int Id { get; set; }\n    public decimal Amount { get; set; }\n    public string Customer { get; set; }\n}\n\npublic class JsonSerializer<T> : ISerializer<T>\n{\n    public byte[] Serialize(T data, SerializationContext context) =>\n        JsonSerializer.SerializeToUtf8Bytes(data);\n}

\n\n

إعداد المستهلك (Consumer)

يقرأ المستهلك الرسائل من موضوع واحد أو أكثر. مثال على الاشتراك والمعالجة:

using Confluent.Kafka;\n\nvar config = new ConsumerConfig\n{\n    BootstrapServers = "localhost:9092",\n    GroupId = "my-consumer-group",\n    AutoOffsetReset = AutoOffsetReset.Earliest\n};\n\nusing (var consumer = new ConsumerBuilder<string, string>(config).Build())\n{\n    consumer.Subscribe("my-topic");\n\n    CancellationTokenSource cts = new CancellationTokenSource();\n    Console.CancelKeyPress += (_, e) => { e.Cancel = true; cts.Cancel(); };\n\n    try\n    {\n        while (true)\n        {\n            var consumeResult = consumer.Consume(cts.Token);\n            Console.WriteLine($"تم الاستلام: Key={consumeResult.Message.Key}, Value={consumeResult.Message.Value}");\n        }\n    }\n    catch (OperationCanceledException)\n    {\n        consumer.Close();\n    }\n}

\n\n

معلمات التكوين الهامة: