在 Kotlin 中使用 Kafka：库与示例

在 Kotlin 中使用 Kafka：库概览与实践示例

Apache Kafka 是一个分布式流数据处理平台，已成为微服务架构中的事实标准。Kotlin 作为一种现代 JVM 语言，与 Java 生态系统完全兼容，因此可以使用所有流行的 Kafka 库。在本文中，我们将探讨三种主要方法：官方 Java Kafka 客户端、Spring Kafka 和轻量级库 Kafkakit。所有示例均使用 Kotlin 编写。

1. 官方 Kafka Client（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("已发送到分区 ${metadata.partition()} 偏移量 ${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