العمل مع Apache Kafka في Java: دليل شامل للمطور

العمل مع Apache Kafka في Java: من الأساسيات إلى الاستخدام المتقدم

Apache Kafka هي منصة موزعة لمعالجة البيانات المتدفقة، والتي أصبحت المعيار الفعلي لبناء بنى تعتمد على الأحداث (event-driven architectures). في هذا الدليل، سنشرح كيفية العمل بكفاءة مع Kafka باستخدام Java: بدءًا من تثبيت المكتبات الضرورية وحتى كتابة المنتج والمستهلك مع معالجة الأخطاء.

لماذا Java و Kafka ثنائي مثالي؟

Java هي إحدى اللغات الرئيسية للعمل مع Kafka لعدة أسباب:

• عميل Kafka الرسمي مكتوب بلغة Java ويوفر أقصى أداء
• نظام بيئي غني بالمكتبات (Spring Kafka, Kafka Streams, Akka Streams)
• الكتابة الصارمة (strict typing) وإمكانية الضبط الدقيق عبر كائنات التكوين

تثبيت وإضافة التبعيات

للعمل مع Kafka في Java، ستحتاج إلى إضافة التبعية في pom.xml (Maven) أو build.gradle (Gradle).

Maven

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

Gradle

implementation 'org.apache.kafka:kafka-clients:3.6.0'

إنشاء Kafka Producer (المنتج)

المنتج مسؤول عن إرسال الرسائل إلى مواضيع (topics) Kafka. إليك مثال أساسي على الإعداد:

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

public class SimpleProducer {    public static void main(String[] args) {        // 1. إعداد خصائص المنتج        Properties props = new Properties();        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());                // 2. إنشاء المنتج        KafkaProducer<String, String> producer = new KafkaProducer<>(props);                // 3. إرسال رسالة        ProducerRecord<String, String> record = new ProducerRecord<>(            "my-topic",             "key-1",             "Hello, Kafka from Java!"        );                // 4. إرسال غير متزامن مع callback        producer.send(record, (metadata, exception) -> {            if (exception == null) {                System.out.println("تم إرسال الرسالة: " +                     "topic=" + metadata.topic() +                     ", partition=" + metadata.partition() +                     ", offset=" + metadata.offset());            } else {                System.err.println("خطأ في الإرسال: " + exception.getMessage());            }        });                // 5. إغلاق المنتج        producer.close();    }}

إنشاء Kafka Consumer (المستهلك)

المستهلك يقرأ الرسائل من المواضيع. من المهم تكوين مجموعة المستهلكين بشكل صحيح لتحقيق توازن الحمل.

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

public class SimpleConsumer {    public static void main(String[] args) {        // 1. إعداد خصائص المستهلك        Properties props = new Properties();        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());        props.put(Consu