العمل مع Kafka في Java: المكتبات، الإعداد وأمثلة الكود

العمل مع Kafka في Java: المكتبات، الإعداد وأمثلة الكود

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

المكتبات الرئيسية للعمل مع Kafka في Java

لتكامل تطبيقات Java مع Apache Kafka، توجد عدة مكتبات رئيسية:

\n
• Kafka Clients — المكتبة الرسمية من Apache، والتي توفر واجهة برمجة تطبيقات للمنتجين (Producer) والمستهلكين (Consumer). هذه هي الطريقة الأقل والأكثر موثوقية للتفاعل مع الوسطاء.
\n
• Spring Kafka — امتداد لإطار Spring Framework، يبسط إعداد واستخدام Kafka من خلال التعليقات التوضيحية والقوالب (KafkaTemplate، @KafkaListener).
\n
• Kafka Streams — مكتبة لبناء تطبيقات معالجة التدفق (stream processing) في Java، دون الحاجة إلى Spark أو Flink. تسمح بإجراء التصفية والتجميع ودمج التدفقات.
\n
• Akka Streams Kafka (Alpakka) — مكتبة تفاعلية للعمل مع Kafka تعتمد على Akka Streams، مناسبة للتطبيقات غير المتزامنة وغير المحظورة.
\n

في هذا الدليل، سنركز على المكتبة الرسمية Kafka Clients، لأنها الأساس لجميع الأدوات الأخرى.

إضافة التبعيات (Maven)

للبدء، أضف التبعية التالية إلى ملف pom.xml:

<dependency>\n    <groupId>org.apache.kafka</groupId>\n    <artifactId>kafka-clients</artifactId>\n    <version>3.7.0</version>\n</dependency>

إذا كنت تستخدم Gradle، أضف:

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

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

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

import org.apache.kafka.clients.producer.*;\nimport java.util.Properties;\n\npublic class SimpleProducer {\n    public static void main(String[] args) {\n        // إعداد خصائص المنتج\n        Properties props = new Properties();\n        props.put("bootstrap.servers", "localhost:9092");\n        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");\n        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");\n        props.put("acks", "all"); // انتظار التأكيد من جميع النسخ المتماثلة\n        props.put("retries", 3);  // عدد المحاولات عند حدوث خطأ\n\n        // إنشاء المنتج\n        Producer<String, String> producer = new KafkaProducer<>(props);\n\n        // إرسال رسالة\n        ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key1", "Hello, Kafka from Java!");\n        producer.send(record, (metadata, exception) -> {\n            if (exception == null) {\n                System.out.println("تم إرسال الرسالة إلى الموضوع: " + metadata.topic() +\n                                   ", القسم: " + metadata.partition() +\n                                   ", الإزاحة: " + metadata.offset());\n            } else {\n                System.err.println("خطأ في الإرسال: " + exception.getMessage());\n            }\n        });\n\n        // إغلاق المنتج\n        producer.close();\n    }\n}

في هذا المثال، نحدد عنوان الوسيط (localhost:9092)، وأدوات التسلسل للمفتاح والقيمة (في هذا