Kafka Client для Java: полное руководство с примерами

Введение в Apache Kafka Client для Java

Apache Kafka — это распределенная платформа потоковой передачи событий, которая позволяет публиковать, хранить и обрабатывать потоки данных в реальном времени. Библиотека Kafka Client для Java является официальным клиентом, предоставляющим API для взаимодействия с брокерами Kafka. Она позволяет Java-приложениям выступать в роли продюсеров (отправлять сообщения) и консьюмеров (читать сообщения), а также управлять метаданными и административными задачами.

Эта библиотека незаменима при построении микросервисных архитектур, систем обработки логов, ETL-пайплайнов, систем аналитики в реальном времени и любых других сценариев, где требуется надежная, масштабируемая и отказоустойчивая передача данных между компонентами.

Установка Kafka Client для Java

Kafka Client для Java распространяется через Maven Central Repository. Для добавления библиотеки в проект необходимо указать зависимость в файле 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'

После добавления зависимости и обновления проекта все классы библиотеки будут доступны для импорта.

Быстрый старт: Producer и Consumer

Рассмотрим минимальный пример отправки и получения сообщения.

Producer (Отправитель)

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

public class SimpleProducer {
    public static void main(String[] args) {
        // 1. Настройка свойств
        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");

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

        // 3. Отправка сообщения
        ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "key1", "Hello, Kafka!");
        producer.send(record);

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

Consumer (Получатель)

import org.apache.kafka.clients.consumer.*;
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("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");

        // 2. Создание консьюмера
        Consumer<String, String> consumer = new KafkaConsumer<>(props);

        // 3. Подписка на топик
        consumer.subscribe(Collections.singletonList("my-topic"));

        // 4. Чтение сообщений
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }
    }
}

Основные методы и функции библиотеки

Рассмотрим ключевые методы, которые используются при работе с Kafka Client.

1. KafkaProducer.send()

Что делает: Отправляет запись (ProducerRecord) в кластер Kafka. Это асинхронный метод, который немедленно возвращает Future.

Параметры:

• ProducerRecord<K, V> record — запись, содержащая топик, ключ и значение.
• Callback callback (опционально) — колбэк, вызываемый после подтверждения отправки.

Пример:

producer.send(new ProducerRecord<>("orders", "order-123", "{\"item\": \"laptop\"}"), 
    (metadata, exception) -> {
        if (exception != null) {
            System.err.println("Ошибка отправки: " + exception.getMessage());
        } else {
            System.out.println("Отправлено в partition " + metadata.partition() + " с offset " + metadata.offset());
        }
    });

2. KafkaProducer.flush()

Что делает: Принудительно отправляет все буферизированные записи в брокер. Блокирует выполнение до завершения отправки.

Параметры: Нет.

Пример:

producer.send(new ProducerRecord<>("logs", "error", "Critical failure"));
producer.flush(); // Гарантирует, что сообщение отправлено перед закрытием

3. KafkaConsumer.subscribe()

Что делает: Подписывает консьюмера на один или несколько топиков. Консьюмер будет получать сообщения из всех партиций этих топиков.

Параметры:

• Collection<String> topics — список топиков для подписки.
• ConsumerRebalanceListener listener (опционально) — слушатель событий перебалансировки.

Пример:

consumer.subscribe(Arrays.asList("topic1", "topic2"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        System.out.println("Потеряны партиции: " + partitions);
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        System.out.println("Назначены партиции: " + partitions);
    }
});

4. KafkaConsumer.poll()

Что делает: Извлекает записи из подписанных топиков. Блокирует выполнение на указанное время, если данных нет.

Параметры:

• Duration timeout — максимальное время ожидания новых записей.

Пример:

ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
for (ConsumerRecord<String, String> record : records) {
    // Обработка записи
}

5. ConsumerRecords.iterator()

Что делает: Возвращает итератор по всем записям, полученным в результате вызова poll(). Позволяет последовательно обойти все записи.

Параметры: Нет.

Пример:

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
Iterator<ConsumerRecord<String, String>> iter = records.iterator();
while (iter.hasNext()) {
    ConsumerRecord<String, String> record = iter.next();
    System.out.println(record.value());
}

6. KafkaConsumer.commitSync()

Что делает: Синхронно фиксирует текущие смещения (offsets) для всех назначенных партиций. Блокирует выполнение до получения подтверждения от брокера.

Параметры:

• Map<TopicPartition, OffsetAndMetadata> offsets (опционально) — конкретные смещения для фиксации.

Пример:

try {
    consumer.commitSync();
} catch (CommitFailedException e) {
    System.err.println("Не удалось зафиксировать offset: " + e.getMessage());
}

7. KafkaConsumer.commitAsync()

Что делает: Асинхронно фиксирует смещения. Не блокирует выполнение, можно указать колбэк для обработки результата.

Параметры:

• OffsetCommitCallback callback (опционально) — колбэк, вызываемый после завершения фиксации.

Пример:

consumer.commitAsync((offsets, exception) -> {
    if (exception != null) {
        System.err.println("Ошибка асинхронного коммита: " + exception.getMessage());
    }
});

8. KafkaConsumer.seek()

Что делает: Перемещает указатель чтения консьюмера на конкретное смещение в указанной партиции. Позволяет повторно читать сообщения.

Параметры:

• TopicPartition partition — партиция.
• long offset — новое смещение.

Пример:

TopicPartition partition = new TopicPartition("my-topic", 0);
consumer.assign(Arrays.asList(partition));
consumer.seek(partition, 10); // Начать чтение с offset 10

9. KafkaConsumer.wakeup()

Что делает: Прерывает текущий или следующий вызов poll(). Используется для безопасного завершения работы консьюмера из другого потока.

Параметры: Нет.

Пример:

// В главном потоке
consumer.wakeup();

// В потоке консьюмера
try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        // обработка
    }
} catch (WakeupException e) {
    // Игнорируем, завершаем цикл
} finally {
    consumer.close();
}

10. KafkaConsumer.partitionsFor()

Что делает: Получает метаданные о партициях указанного топика: количество партиций, лидеры, реплики.

Параметры:

• String topic — имя топика.

Пример:

List<PartitionInfo> partitions = consumer.partitionsFor("my-topic");
for (PartitionInfo p : partitions) {
    System.out.println("Partition: " + p.partition() + ", Leader: " + p.leader());
}

Реальный практический пример: Система обработки заказов

Создадим систему, которая принимает заказы от пользователей, отправляет их в Kafka, а затем обрабатывает (например, сохраняет в базу данных).

Producer: Отправка заказов

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

public class OrderProducer {
    private final Producer<String, String> producer;

    public OrderProducer(String bootstrapServers) {
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("acks", "all"); // Ожидать подтверждения от всех реплик
        this.producer = new KafkaProducer<>(props);
    }

    public void sendOrder(String orderId, String orderData) {
        ProducerRecord<String, String> record = new ProducerRecord<>("orders", orderId, orderData);
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                System.out.println("Заказ " + orderId + " отправлен в партицию " + metadata.partition());
            } else {
                System.err.println("Ошибка отправки заказа " + orderId + ": " + exception.getMessage());
            }
        });
    }

    public void close() {
        producer.close();
    }

    public static void main(String[] args) {
        OrderProducer producer = new OrderProducer("localhost:9092");
        producer.sendOrder("order-001", "