Работа с Apache Kafka в Java: полное руководство для разработчика

Работа с Apache Kafka в Java: от основ до продвинутого использования

Apache Kafka — это распределенная платформа для потоковой обработки данных, которая стала стандартом де-факто для построения event-driven архитектур. В этом руководстве мы разберем, как эффективно работать с Kafka на Java: от установки необходимых библиотек до написания продюсера и консьюмера с обработкой ошибок.

Почему Java и Kafka — идеальная пара?

Java — один из основных языков для работы с Kafka по нескольким причинам:

• Официальный клиент Kafka написан на Java и предоставляет максимальную производительность
• Богатая экосистема библиотек (Spring Kafka, Kafka Streams, Akka Streams)
• Строгая типизация и возможность тонкой настройки через конфигурационные объекты

Установка и подключение зависимостей

Для работы с Kafka в Java вам потребуется добавить зависимость в pom.xml (Maven) или build.gradle (Gradle).

Maven

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.6.0</version>
</dependency>

Gradle

implementation 'org.apache.kafka:kafka-clients:3.6.0'

Создание Kafka Producer (Продюсера)

Продюсер отвечает за отправку сообщений в топики Kafka. Вот базовый пример настройки:

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

public class SimpleProducer {
    public static void main(String[] args) {
        // 1. Настройка свойств продюсера
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        
        // 2. Создание продюсера
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        
        // 3. Отправка сообщения
        ProducerRecord<String, String> record = new ProducerRecord<>(
            "my-topic", 
            "key-1", 
            "Hello, Kafka from Java!"
        );
        
        // 4. Асинхронная отправка с callback
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                System.out.println("Сообщение отправлено: " + 
                    "topic=" + metadata.topic() + 
                    ", partition=" + metadata.partition() + 
                    ", offset=" + metadata.offset());
            } else {
                System.err.println("Ошибка отправки: " + exception.getMessage());
            }
        });
        
        // 5. Закрытие продюсера
        producer.close();
    }
}

Создание Kafka Consumer (Консьюмера)

Консьюмер читает сообщения из топиков. Важно правильно настроить группу консьюмеров для балансировки нагрузки.

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class SimpleConsumer {
    public static void main(String[] args) {
        // 1. Настройка свойств консьюмера
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // читаем с начала
        
        // 2. Создание консьюмера
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        
        // 3. Подписка на топик
        consumer.subscribe(Collections.singletonList("my-topic"));
        
        // 4. Цикл чтения сообщений
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Получено сообщение: key=%s, value=%s, partition=%d, offset=%d%n",
                        record.key(), record.value(), record.partition(), record.offset());
                    
                    // Здесь можно добавить бизнес-логику обработки
                    processMessage(record.value());
                }
                // Важно: коммит offset'ов после успешной обработки
                consumer.commitSync();
            }
        } catch (Exception e) {
            System.err.println("Ошибка при чтении: " + e.getMessage());
        } finally {
            consumer.close();
        }
    }
    
    private static void processMessage(String message) {
        // Пример обработки
        System.out.println("Обработка сообщения: " + message);
    }
}

П