Kafka в Kotlin: библиотеки и примеры кода

Работа с Kafka в Kotlin: обзор библиотек и практические примеры

Apache Kafka — это распределённая платформа для потоковой обработки данных, ставшая стандартом де-факто в микросервисной архитектуре. Kotlin, будучи современным JVM-языком, полностью совместим с Java-экосистемой, что позволяет использовать все популярные библиотеки для Kafka. В этой статье мы рассмотрим три основных подхода: официальный Java-клиент Kafka, Spring Kafka и легковесную библиотеку Kafkakit. Все примеры написаны на Kotlin.

1. Официальный Kafka Client (Java-клиент)

Библиотека kafka-clients от Apache — это основа, на которой строятся все остальные обёртки. Она предоставляет низкоуровневый API для продюсеров и консьюмеров. Для Kotlin достаточно добавить зависимость в build.gradle.kts:

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 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.ConsumerRecords
import org.apache.kafka.clients.consumer.KafkaConsumer
import java.time.Duration
import 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. Spring Kafka с Kotlin

Если вы используете 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.springframework.kafka.core.KafkaTemplate
import org.springframework.stereotype.Component

@Component
class MessageProducer(private val kafkaTemplate: KafkaTemplate<String, String>) {

    fun send(topic: String, key: String, value: String) {
        kafkaTemplate.send(topic, key, value).whenComplete { _, ex ->
            if (ex != null) println("Error: ${ex.message}") else println("Sent $value")
        }
    }
}

Консьюмер с аннотацией @KafkaListener

import org.springframework.kafka.annotation.KafkaListener
import org.springframework.stereotype.Component

@Component
class MessageConsumer {

    @KafkaListener(topics = ["my-topic"], groupId = "spring-kotlin-group")
    fun consume(message: String) {
        println("Consumed message: $message")
    }
}

Spring Kafka также поддерживает @KafkaHandler для обработки разных типов сообщений в одном листнере и транзакционность.

3. Kafkakit — лёгкая Kotlin-обёртка

Kafkakit — это минималистичная библиотека, написанная на Kotlin, которая упрощает создание продюсеров и консьюмеров за счёт функций высшего порядка и безопас