Работа с Kafka в Java: библиотеки, настройка и примеры кода

Работа с Kafka в Java: библиотеки, настройка и примеры кода

Apache Kafka — это распределённая платформа для потоковой передачи данных, которая широко используется для построения высоконагруженных систем обмена сообщениями, сбора логов, обработки событий в реальном времени и интеграции микросервисов. В экосистеме Java Kafka занимает особое место благодаря официальной библиотеке Kafka Clients и множеству дополнительных инструментов. В этой статье мы подробно разберём, как работать с Kafka в Java, какие библиотеки использовать, и приведём практические примеры кода.

Основные библиотеки для работы с Kafka в Java

Для интеграции Java-приложений с Apache Kafka существует несколько ключевых библиотек:

• Kafka Clients — официальная библиотека от Apache, предоставляющая API для продюсеров (Producer) и консюмеров (Consumer). Это минимальный и самый надёжный способ взаимодействия с брокерами.
• Spring Kafka — расширение для Spring Framework, которое упрощает настройку и использование Kafka через аннотации и шаблоны (KafkaTemplate, @KafkaListener).
• Kafka Streams — библиотека для построения потоковых приложений (stream processing) на Java, без необходимости в Spark или Flink. Позволяет выполнять фильтрацию, агрегацию и объединение потоков.
• Akka Streams Kafka (Alpakka) — реактивная библиотека для работы с Kafka на базе Akka Streams, подходит для асинхронных и неблокирующих приложений.

В данном руководстве мы сосредоточимся на официальной библиотеке Kafka Clients, так как она является основой для всех остальных инструментов.

Подключение зависимостей (Maven)

Для начала работы добавьте в pom.xml следующую зависимость:

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

Если вы используете Gradle, добавьте:

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

Настройка продюсера (Producer)

Продюсер отправляет сообщения в топики Kafka. Рассмотрим простейший пример создания и отправки сообщения.

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

public class SimpleProducer {
    public static void main(String[] args) {
        // Настройка свойств продюсера
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("acks", "all"); // Ожидание подтверждения от всех реплик
        props.put("retries", 3);  // Количество повторных попыток при ошибке

        // Создание продюсера
        Producer<String, String> producer = new KafkaProducer<>(props);

        // Отправка сообщения
        ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key1", "Hello, Kafka from Java!");
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                System.out.println("Сообщение отправлено в топик: " + metadata.topic() +
                                   ", партиция: " + metadata.partition() +
                                   ", offset: " + metadata.offset());
            } else {
                System.err.println("Ошибка отправки: " + exception.getMessage());
            }
        });

        // Закрытие продюсера
        producer.close();
    }
}

В этом примере мы задаём адрес брокера (localhost:9092), сериализаторы для ключа и значения (в данном случае строки), а также параметры надёжности (acks=all). Метод send() асинхронный, поэтому для обработки результата используется callback.

Настройка консюмера (Consumer)

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

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.*;

public class SimpleConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "my-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "earliest"); // Читать с самого начала

        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my-topic"));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration