Kotlin में Kafka के साथ कार्य: लाइब्रेरीज़ और उदाहरण

Kotlin में Kafka के साथ कार्य: लाइब्रेरीज़ का अवलोकन और व्यावहारिक उदाहरण

Apache Kafka एक वितरित डेटा स्ट्रीमिंग प्लेटफ़ॉर्म है, जो माइक्रोसर्विस आर्किटेक्चर में डी-फ़ैक्टो मानक बन गया है। Kotlin, एक आधुनिक JVM भाषा होने के नाते, Java इकोसिस्टम के साथ पूरी तरह से संगत है, जो Kafka के लिए सभी लोकप्रिय लाइब्रेरीज़ का उपयोग करने की अनुमति देता है। इस लेख में, हम तीन मुख्य दृष्टिकोणों पर विचार करेंगे: आधिकारिक Java Kafka क्लाइंट, Spring Kafka और हल्की लाइब्रेरी Kafkakit। सभी उदाहरण Kotlin में लिखे गए हैं।

1. आधिकारिक Kafka क्लाइंट (Java क्लाइंट)

Apache की kafka-clients लाइब्रेरी वह आधार है जिस पर अन्य सभी रैपर बनाए गए हैं। यह प्रोड्यूसर और कंज़्यूमर के लिए निम्न-स्तरीय API प्रदान करती है। Kotlin के लिए, build.gradle.kts में निर्भरता जोड़ना पर्याप्त है:

implementation("org.apache.kafka:kafka-clients:3.6.0")

Kotlin में प्रोड्यूसर का उदाहरण

import org.apache.kafka.clients.producer.KafkaProducerimport org.apache.kafka.clients.producer.ProducerRecordimport java.util.Properties

fun createProducer(): KafkaProducer<String, String> {    val props = Properties().apply {        put("bootstrap.servers", "localhost:9092")        put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")        put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")    }    return KafkaProducer(props)}

fun main() {    val producer = createProducer()    val record = ProducerRecord("my-topic", "key1", "Hello from Kotlin!")    producer.send(record) { metadata, exception ->        if (exception != null) {            println("भेजने में त्रुटि: ${exception.message}")        } else {            println("partition ${metadata.partition()} offset ${metadata.offset()} में भेजा गया")        }    }    producer.close()}

Kotlin में कंज़्यूमर का उदाहरण

import org.apache.kafka.clients.consumer.ConsumerRecordsimport org.apache.kafka.clients.consumer.KafkaConsumerimport java.time.Durationimport java.util.Properties

fun createConsumer(): KafkaConsumer<String, String> {    val props = Properties().apply {        put("bootstrap.servers", "localhost:9092")        put("group.id", "kotlin-group")        put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")        put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")        put("auto.offset.reset", "earliest")    }    return KafkaConsumer(props)}

fun main() {    val consumer = createConsumer()    consumer.subscribe(listOf("my-topic"))

    while (true) {        val records: ConsumerRecords<String, String> = consumer.poll(Duration.ofMillis(100))        records.forEach { record ->            println("प्राप्त: key=${record.key()}, value=${record.value()}, offset=${record.offset()}")        }    }}

2. Kotlin के साथ Spring Kafka

यदि आप Spring Boot का उपयोग कर रहे हैं, तो Spring Kafka एक स्वाभाविक विकल्प है। यह एनोटेशन और स्वचालित कॉन्फ़िगरेशन के माध्यम से एक घोषणात्मक दृष्टिकोण प्रदान करता है। निर्भरता जोड़ें:

implementation("org.springframework.kafka:spring-kafka")

application.yml में कॉन्फ़िगरेशन

spring:  kafka:    bootstrap-servers: localhost:9092    producer:      key-serializer: org.apache.kafka.common.serialization.StringSerializer      value-serializer: org.apache.kafka.common.serialization.StringSerializer    consumer:      group-id: spring-kotlin-group      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer      auto-offset-reset: earliest

KafkaTemplate के साथ प्रोड्यूसर

import org.springfram