Technical Tutorials

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

فهم التجريدات الأساسية

قبل الغوص في الكود، من الضروري فهم اللبنات الأساسية لـ Kafka. على عكس طابور التقليدي الذي يُستهلك ويُتخلص منه، يحتفظ Kafka بالرسائل لفترة قابلة للتكوين، مما يسمح لعدة خدمات بقراءة نفس البيانات بوتيرتها الخاصة. تشمل التجريدات الرئيسية ما يلي:

الموضوع (Topic): فئة أو اسم تغذية تُنشر إليها السجلات.
التجزئة (Partition): يتم تقسيم الموضوعات إلى تجزئات للسماح بالتوازي وقابلية التوسع. كل تجزئة هي تسلسل مرتب وغير قابل للتغيير للسجلات.
المنتج (Producer): تطبيق ينشر (يكتب) السجلات إلى موضوعات Kafka.
المستهلك (Consumer): تطبيق يشترك (يقرأ) في الموضوعات ويعالج السجلات.
مجموعة المستهلكين (Consumer Group): مجموعة من المستهلكين الذين يقرأون مجتمعة جميع تجزئات موضوع ما، مما يتيح موازنة الحمل والمعالجة المتوازية.

إعداد البيئة

للمتابعة مع الأمثلة العملية في هذا المنشور، ستحتاج إلى عامل تشغيل Kafka (Broker) نشط ومكتبة عميل Confluent Kafka. تأكد من تثبيت Java 11 أو إصدار أحدث. أضف التبعية التالية إلى ملف pom.xml إذا كنت تستخدم Maven:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.4.0</version>
</dependency>

بناء منتج (Producer)

المسؤول عن إرسال البيانات إلى Kafka هو المنتج. يقوم بتسلسل المفاتيح والقيم إلى بايتات ويحدد التجزئة التي سيتم الكتابة إليها. فيما يلي مثال بسيط لمنتج Kafka يرسل رسالة نصية بسيطة إلى موضوع باسم test-topic.

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class KafkaProducerExample {
    public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        try (Producer<String, String> producer = new KafkaProducer<>(props)) {
            ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", "key1", "Hello Kafka!");
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    System.out.println("Error sending message: " + exception.getMessage());
                } else {
                    System.out.println("Sent to partition: " + metadata.partition() + 
                                       ", offset: " + metadata.offset());
                }
            });
            producer.flush();
        }
    }
}

لاحظ استخدام flush() لضمان إرسال جميع الرسائل المخزنة مؤقتًا قبل خروج التطبيق. في بيئة الإنتاج، ستتعامل عادةً مع الاستدعاءات غير المتزامنة بشكل أكثر قوة وتدير دورة حياة المنتج ككائن أحادي (Singleton).

بناء مستهلك (Consumer)

المستهلكون أكثر تعقيدًا قليلاً لأنهم يجب أن يديروا الإزاحات (Offsets) ويتعاملوا مع أخطاء معالجة الرسائل المحتملة. يشترك المستهلك في موضوع ويستعلم عن الرسائل الجديدة في حلقة تكرار.

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        try (Consumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList("test-topic"));
            
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Consumed record: key=%s, value=%s, partition=%d%n",
                                      record.key(), record.value(), record.partition());
                }
            }
        }
    }
}

أفضل الممارسات للإنتاج

بينما تكون الأمثلة المذكورة أعلاه وظيفية، تتطلب أنظمة الإنتاج اعتبارات إضافية. استخدم دائمًا acks=all في تكوين المنتج الخاص بك لضمان بقاء الرسائل محفوظة في جميع النسخ المتطابقة المتزامنة (in-sync replicas)، مما يضمن المتانة. بالنسبة للمستهلكين، فكر في تعطيل الالتزام التلقائي (auto-commit) والالتزام بالإزاحات يدويًا فقط بعد معالجة الرسائل بنجاح لتحقيق دلالات الدقة الواحدة (exactly-once) أو لمنع فقدان البيانات أثناء الأعطال. بالإضافة إلى ذلك، قم دائمًا بتنفيذ منطق إعادة المحاولة وفواصل الدائرة (circuit breakers) للتعامل مع فشل الشبكة العابرة بسلاسة.

الخاتمة

يُعد Apache Kafka أداة قوية، وعندما يتم فهمها بشكل صحيح، يمكنها تحويل طريقة تعامل تطبيقاتك مع البيانات. من خلال إتقان نمط المنتج-المستهلك وفهم استراتيجيات التجزئة، يمكنك بناء أنظمة مرنة وقابلة للتوسع ومنفصلة عن بعضها البعض. مع تقدمك، استكشف الميزات المتقدمة مثل Kafka Streams للتحليل في الوقت الفعلي وسجل المخططات (Schema Registry) لإدارة عقود البيانات لتقوية بنيتك المعمارية القائمة على الأحداث بشكل أكبر.