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

العمل مع Kafka في Kotlin: نظرة عامة على المكتبات وأمثلة عملية

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

1. Kafka Clients — العمل الأساسي مع المنتج والمستهلك

Apache Kafka Clients هي مكتبة Java الرسمية التي توفر واجهة برمجة تطبيقات منخفضة المستوى لإرسال واستقبال الرسائل. في Kotlin، يتم تضمينها عبر Maven أو Gradle:

// build.gradle.kts\ndependencies {\n    implementation("org.apache.kafka:kafka-clients:3.6.0")\n}

لننظر إلى مثال لإنشاء منتج (producer) في Kotlin:

import org.apache.kafka.clients.producer.KafkaProducer\nimport org.apache.kafka.clients.producer.ProducerRecord\nimport java.util.Properties\n\nfun main() {\n    val props = Properties().apply {\n        put("bootstrap.servers", "localhost:9092")\n        put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")\n        put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")\n    }\n\n    val producer = KafkaProducer<String, String>(props)\n    val record = ProducerRecord("my-topic", "key1", "Hello from Kotlin!")\n\n    producer.send(record) { metadata, exception ->\n        if (exception != null) {\n            println("خطأ في الإرسال: ${exception.message}")\n        } else {\n            println("تم الإرسال إلى القسم ${metadata.partition()} مع offset ${metadata.offset()}")\n        }\n    }\n\n    producer.close()\n}

وبالمثل، يتم إنشاء مستهلك (consumer) لقراءة الرسائل:

import org.apache.kafka.clients.consumer.ConsumerRecords\nimport org.apache.kafka.clients.consumer.KafkaConsumer\nimport java.time.Duration\nimport java.util.Properties\n\nfun main() {\n    val props = Properties().apply {\n        put("bootstrap.servers", "localhost:9092")\n        put("group.id", "my-group")\n        put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")\n        put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")\n        put("auto.offset.reset", "earliest")\n    }\n\n    val consumer = KafkaConsumer<String, String>(props)\n    consumer.subscribe(listOf("my-topic"))\n\n    while (true) {\n        val records: ConsumerRecords<String, String> = consumer.poll(Duration.ofMillis(100))\n        records.forEach { record ->\n            println("تم الاستلام: المفتاح = ${record.key()}, القيمة = ${record.value()}, offset = ${record.offset()}")\n        }\n    }\n}

2. Kafka Streams — معالجة تدفق البيانات

Kafka Streams هي مكتبة لبناء تطبيقات معالجة التدفق بناءً على Kafka. تتيح إجراء التحويلات والتجميعات وربط تدفقات البيانات مباشرة على جانب العميل. التضمين:

dependencies {\n    implementation("org.apache.kafka:kafka-streams:3.6.0")\n}

مثال لتطبيق بسيط يقرأ الرسائل من موضوع، ويحولها إلى أحرف كبيرة، ويكتبها في موضوع آخر:

import org.apache.kafka.streams.KafkaStreams\nimport org.apache.kafka.streams.StreamsBuilder\nimport org.apache.kafka.streams.StreamsConfig\nimport org.apache.kafka.streams.kstream.KStream\nimport java.util.Properties\n\nfun main() {\n    val props = Properties().apply {\n        put(StreamsConfig.APPLICATION_ID_CONFIG, "uppercase-app")\n        put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "local