Kafka के साथ Kotlin में काम: लाइब्रेरी, सेटअप और उदाहरण

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

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

1. Kafka Clients — प्रोड्यूसर और कंज्यूमर के साथ बेसिक काम

Apache Kafka Clients एक आधिकारिक Java लाइब्रेरी है जो संदेश भेजने और प्राप्त करने के लिए लो-लेवल API प्रदान करती है। Kotlin में इसे Maven या Gradle के माध्यम से जोड़ा जाता है:

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

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

import org.apache.kafka.clients.producer.KafkaProducer
import org.apache.kafka.clients.producer.ProducerRecord
import java.util.Properties

fun main() {
    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")
    }

    val producer = KafkaProducer<String, String>(props)
    val record = ProducerRecord("my-topic", "key1", "Kotlin से नमस्ते!")

    producer.send(record) { metadata, exception ->
        if (exception != null) {
            println("भेजने में त्रुटि: ${exception.message}")
        } else {
            println("पार्टीशन ${metadata.partition()} में offset ${metadata.offset()} के साथ भेजा गया")
        }
    }

    producer.close()
}

इसी तरह, संदेश पढ़ने के लिए कंज्यूमर बनाया जाता है:

import org.apache.kafka.clients.consumer.ConsumerRecords
import org.apache.kafka.clients.consumer.KafkaConsumer
import java.time.Duration
import java.util.Properties

fun main() {
    val props = Properties().apply {
        put("bootstrap.servers", "localhost:9092")
        put("group.id", "my-group")
        put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
        put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
        put("auto.offset.reset", "earliest")
    }

    val consumer = KafkaConsumer<String, String>(props)
    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. Kafka Streams — डेटा की स्ट्रीम प्रोसेसिंग

Kafka Streams एक लाइब्रेरी है जो Kafka के आधार पर स्ट्रीम प्रोसेसिंग एप्लिकेशन बनाने के लिए उपयोग होती है। यह क्लाइंट साइड पर डेटा स्ट्रीम के ट्रांसफॉर्मेशन, एग्रीगेशन और जॉइन करने की सुविधा देती है। इसे जोड़ने का तरीका:

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

एक सरल एप्लिकेशन का उदाहरण जो टॉपिक से संदेश पढ़ता है, उन्हें अपरकेस में बदलता है और दूसरे टॉपिक में लिखता है:

import org.apache.kafka.streams.KafkaStreams
import org.apache.kafka.streams.StreamsBuilder
import org.apache.kafka.streams.StreamsConfig
import org.apache.kafka.streams.kstream.KStream
import java.util.Properties

fun main() {
    val props = Properties().apply {
        put(StreamsConfig.APPLICATION_ID_CONFIG, "uppercase-app")
        put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
        put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().