Распределённые Kafka-консьюмеры на Java

Введение

Цель этого урока — написать полноценный Java-консьюмер с использованием библиотеки kafka-clients, запустить несколько его экземпляров и на собственном опыте убедиться, как работает механизм распределения партиций.

Вы увидите, как код, который вы напишете, реализует концепции, которые мы обсуждали:

• Один консьюмер в группе обрабатывает все партиции.
• При добавлении новых консьюмеров в ту же группу партиции перераспределяются между ними.

Это практическое упражнение закрепит ваше понимание одного из самых мощных механизмов Kafka.

1. Анатомия Java-консьюмера

Прежде чем писать код, давайте разберемся с основными компонентами kafka-clients API для создания консьюмера.

Для начала, посмотрите короткий обзор от Confluent, который познакомит вас с ключевыми классами и методами.

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

Шаг 1: Зависимости

Для работы с Kafka в Java-приложении нам понадобится библиотека kafka-clients. Если вы используете Maven, зависимость выглядит так:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.7.0</version> <!-- Рекомендуется использовать актуальную версию -->
</dependency>

Шаг 2: Конфигурация

Для создания экземпляра KafkaConsumer необходимо передать ему набор конфигурационных параметров. Обычно это делается через объект java.util.Properties.

Вот ключевые параметры для нашего консьюмера:

• bootstrap.servers: Адрес одного или нескольких брокеров Kafka (например, "localhost:9092").
• group.id: Самый важный параметр для этого урока. Это строковый идентификатор, который объединяет несколько экземпляров консьюмера в одну группу. Все консьюмеры с одинаковым group.id будут работать совместно.
• key.deserializer и value.deserializer: Kafka хранит все сообщения в виде байтовых массивов. Эти параметры указывают, какие классы использовать для преобразования (десериализации) байтов обратно в нужные нам типы данных (например, String или Integer). Мы будем использовать StringDeserializer.
• auto.offset.reset: Что делать консьюмеру, если он впервые подключается к топику (или его сохраненная позиция устарела)?
  • earliest: Начать читать с самого начала топика.
  • latest: Начать читать только новые сообщения, которые появятся после его подключения.
    Для нашего эксперимента мы выберем earliest, чтобы обработать все сообщения, которые уже есть в топике.

Шаг 3: Создание и подписка

С готовыми Properties мы можем создать экземпляр KafkaConsumer, а затем подписать его на один или несколько топиков.

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-java-consumer-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");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("demo-topic")); // Подписываемся на топик 'demo-topic'

Шаг 4: Цикл опроса (Poll Loop)

Сердце любого консьюмера — это бесконечный цикл, в котором он опрашивает Kafka на предмет новых сообщений. Это делается с помощью метода poll().

while (true) {
    // Запрашиваем сообщения, ожидая до 100 мс
    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());
    }
}

Важно понимать, что poll() — это не просто запрос данных. Во время вызова этого метода консьюмер также отправляет "сигналы жизни" (heartbeats) координатору группы и участвует в процессе ребалансировки, если это необходимо. Метод возвращает ConsumerRecords — коллекцию, которая может содержать от нуля до множества сообщений.

2. Практическое упражнение: Запускаем консьюмеры

Теперь давайте соберем все вместе и проведем эксперимент.

Подготовка

1. Убедитесь, что Kafka запущена. Если нет, запустите Zookeeper и Kafka Server, как мы делали в первом уроке.

2. Создайте топик с несколькими партициями. Для нашего эксперимента идеально подойдет топик с 3 партициями. Выполните в терминале:

   kafka-topics.sh --bootstrap-server localhost:9092 --create --topic demo-topic --partitions 3 --replication-factor 1
   

   Если топик demo-topic уже существует, удалите его (--delete) и создайте заново.

3. Отправьте сообщения в топик. Используйте Java-продюсер из урока 3, чтобы отправить 10-20 сообщений в demo-topic. Часть сообщений отправьте с ключом, часть — без, чтобы они распределились по разным партициям.

Код консьюмера

Создайте новый Java-класс SimpleConsumer и вставьте в него следующий код. Он основан на принципах, которые мы только что обсудили.

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class SimpleConsumer {

    private static final Logger log = LoggerFactory.getLogger(SimpleConsumer.class);

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        // Обратите внимание на group.id - он у всех будет одинаковый
        props.put("group.id", "my-java-consumer-group");
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList("demo-topic"));

            log.info("Консьюмер запущен. Ожидание сообщений...");

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

                for (ConsumerRecord<String, String> record : records) {
                    log.info("Сообщение получено: key={}, value={}, partition={}, offset={}",
                            record.key(), record.value(), record.partition(), record.offset());
                }
            }
        } catch (Exception e) {
            log.error("Произошла ошибка в консьюмере", e);
        }
    }
}

Не забудьте добавить зависимость для SLF4J, например slf4j-simple, чтобы логирование работало.

Эксперимент

Теперь самое интересное. Мы запустим несколько экземпляров этого приложения.

Шаг 1: Запуск одного консьюмера

• Действие: Запустите SimpleConsumer в вашей IDE.
• Наблюдение: В консоли вы увидите логи о получении сообщений. Обратите внимание на поле partition. Вы увидите, что этот единственный экземпляр получает сообщения из всех трех партиций (0, 1 и 2). Это подтверждает правило: если в группе один консьюмер, он забирает всю работу на себя.

Шаг 2: Запуск второго консьюмера

• Действие: Не останавливая первый экземпляр, запустите SimpleConsumer еще раз. В IntelliJ IDEA для этого можно кликнуть правой кнопкой мыши на main и выбрать "Run 'SimpleConsumer.main()'". IDE предложит выбор: "Stop and Rerun" или запустить еще один экземпляр. Выберите запуск нового экземпляра. Или отредактируйте конфигурацию запуска, поставив галочку "Allow multiple instances".
• Наблюдение: Посмотрите на логи обоих экземпляров. Вы увидите, что Kafka инициировала ребалансировку. Теперь партиции распределены между двумя консьюмерами. Например, первый экземпляр может получать сообщения из партиций 0 и 2, а второй — только из партиции 1. Распределение может быть и другим (например, 1 и 2, 0), но главное, что каждая партиция теперь назначена только одному консьюмеру.

Шаг 3: Запуск третьего консьюмера

• Действие: Запустите третий экземпляр SimpleConsumer.
• Наблюдение: После очередной ребалансировки вы увидите, что теперь каждый из трех консьюмеров обрабатывает ровно одну партицию. Мы достигли максимального параллелизма для нашего топика с тремя партициями.

Шаг 4 (бонусный): Запуск четвертого консьюмера

• Действие: Запустите четвертый экземпляр.
• Наблюдение: Посмотрите на логи четвертого консьюмера. Он запустится, но не будет получать никаких сообщений. Почему? Потому что все три партиции уже распределены между первыми тремя консьюмерами. Этот четвертый экземпляр останется в режиме ожидания (idle), готовый взять на себя работу, если один из активных консьюмеров выйдет из строя.

Чтобы визуально закрепить наблюдаемый эффект, посмотрите следующий фрагмент. Хотя код в видео использует Spring Boot, демонстрация распределения нагрузки абсолютно идентична тому, что вы только что проделали.

Заключение

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

Ключевые выводы из урока:

1. Для создания Java-консьюмера используется класс KafkaConsumer из библиотеки kafka-clients.
2. Ключевые конфигурации включают bootstrap.servers, group.id и десериализаторы.
3. Основная логика консьюмера заключена в бесконечном цикле poll().
4. Запуская несколько экземпляров приложения с одинаковым group.id, мы можем горизонтально масштабировать обработку сообщений.
5. Kafka автоматически распределяет партиции топика между активными участниками группы, обеспечивая параллелизм и отказоустойчивость.

Что дальше?

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