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

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

Apache Kafka هي منصة موزعة لتدفق البيانات، تُستخدم على نطاق واسع لبناء أنظمة تبادل الرسائل عالية التحميل، ومعالجة الأحداث، وتكامل الخدمات المصغرة. للعمل مع Kafka في نظام .NET البيئي، المكتبة الرئيسية والأكثر دعماً هي 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