在 Kotlin 中使用 Kafka：库、配置与示例

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

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.ktsdependencies {    implementation("org.apache.kafka:kafka-clients:3.6.0")}

以下是用 Kotlin 创建生产者（producer）的示例：

import org.apache.kafka.clients.producer.KafkaProducerimport org.apache.kafka.clients.producer.ProducerRecordimport 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", "Hello from Kotlin!")

    producer.send(record) { metadata, exception ->        if (exception != null) {            println("发送错误：${exception.message}")        } else {            println("已发送到分区 ${metadata.partition()}，偏移量 ${metadata.offset()}")        }    }

    producer.close()}

类似地，创建消费者（consumer）来读取消息：

import org.apache.kafka.clients.consumer.ConsumerRecordsimport org.apache.kafka.clients.consumer.KafkaConsumerimport java.time.Durationimport 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("收到：键 = ${record.key()}，值 = ${record.value()}，偏移量 = ${record.offset()}")        }    }}

2. Kafka Streams —— 流式数据处理

Kafka Streams 是一个用于构建基于 Kafka 的流处理应用程序的库。它允许在客户端直接执行数据流的转换、聚合和连接。引入方式：

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

以下是一个简单应用程序的示例，它从主题中读取消息，将其转换为大写，然后写入另一个主题：

import org.apache.kafka.streams.KafkaStreamsimport org.apache.kafka.streams.StreamsBuilderimport org.apache.kafka.streams.StreamsConfigimport org.apache.kafka.streams.kstream.KStreamimport java.util.Properties

fun main() {    val props = Properties().apply {        put(StreamsConfig.APPLICATION_ID_CONFIG, "uppercase-app")        put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "local