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

مقدمة عن عميل Apache Kafka لجافا

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

هذه المكتبة لا غنى عنها عند بناء بنى الخدمات المصغرة، وأنظمة معالجة السجلات، وخطوط أنابيب ETL، وأنظمة التحليلات في الوقت الفعلي، وأي سيناريوهات أخرى تتطلب نقل بيانات موثوقًا وقابلاً للتوسع وتحمل الأخطاء بين المكونات.

تثبيت عميل Kafka لجافا

يتم توزيع عميل Kafka لجافا عبر مستودع Maven Central. لإضافة المكتبة إلى مشروعك، يجب تحديد التبعية في ملف 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'

بعد إضافة التبعية وتحديث المشروع، ستكون جميع فئات المكتبة متاحة للاستيراد.

بداية سريعة: المنتج والمستهلك

لنلقِ نظرة على مثال بسيط لإرسال واستقبال رسالة.

المنتج (المرسل)

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

public class SimpleProducer {
    public static void main(String[] args) {
        // 1. إعداد الخصائص
        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");

        // 2. إنشاء المنتج
        Producer<String, String> producer = new KafkaProducer<>(props);

        // 3. إرسال رسالة
        ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key1", "مرحباً، كافكا!");
        producer.send(record);

        // 4. إغلاق المنتج
        producer.close();
    }
}

المستهلك (المستقبل)

import org.apache.kafka.clients.consumer.*;
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("bootstrap.servers", "localhost:9092");
        props.put("group.id", "my-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "earliest");

        // 2. إنشاء المستهلك
        Consumer<String, String> consumer = new KafkaConsumer<>(props);

        // 3. الاشتراك في الموضوع
        consumer.subscribe(Collections.singletonList("my-topic"));

        // 4. قراءة الرسائل
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }
    }
}