Arbeit mit Kafka in Kotlin: Bibliotheken und Beispiele

Arbeit mit Kafka in Kotlin: Überblick über Bibliotheken und praktische Beispiele

Apache Kafka ist eine verteilte Plattform für die Streaming-Datenverarbeitung, die in der Microservice-Architektur zum De-facto-Standard geworden ist. Kotlin, als moderne JVM-Sprache, ist vollständig kompatibel mit dem Java-Ökosystem, was die Nutzung aller gängigen Bibliotheken für Kafka ermöglicht. In diesem Artikel betrachten wir drei Hauptansätze: den offiziellen Java-Client für Kafka, Spring Kafka und die leichtgewichtige Bibliothek Kafkakit. Alle Beispiele sind in Kotlin geschrieben.

1. Offizieller Kafka-Client (Java-Client)

Die Bibliothek kafka-clients von Apache ist die Grundlage, auf der alle anderen Wrapper aufbauen. Sie bietet eine Low-Level-API für Producer und Consumer. Für Kotlin reicht es, die Abhängigkeit in build.gradle.kts hinzuzufügen:

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

Beispiel eines Producers in 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("Fehler beim Senden: ${exception.message}")        } else {            println("Gesendet an Partition ${metadata.partition()} Offset ${metadata.offset()}")        }    }    producer.close()}

Beispiel eines Consumers in 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("Empfangen: key=${record.key()}, value=${record.value()}, offset=${record.offset()}")        }    }}

2. Spring Kafka mit Kotlin

Wenn Sie Spring Boot verwenden, ist Spring Kafka die naheliegende Wahl. Es bietet einen deklarativen Ansatz über Annotationen und automatische Konfiguration. Fügen Sie die Abhängigkeit hinzu:

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

Konfiguration in 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

Producer mit KafkaTemplate

import org.springfram