Kafka Producer: Отправка сообщений с ключом и без ключа

Введение

В предыдущем уроке мы теоретически разобрали, как ключ сообщения (message key) влияет на его распределение по партициям. Мы узнали, что сообщения без ключа распределяются равномерно для балансировки нагрузки, в то время как сообщения с одинаковым ключом всегда попадают в одну и ту же партицию, что гарантирует порядок их обработки.

Сегодня мы перейдем от теории к практике.

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

Мы создадим простое Java-приложение, настроим его для отправки сообщений в Kafka и на практике убедимся в работе механизмов, которые обсуждали ранее. Этот практический навык — основа для создания любого сервиса, взаимодействующего с Kafka.

1. Анатомия Kafka-продюсера на Java

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

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

   <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>kafka-clients</artifactId>
       <version>3.6.1</version> <!-- или любая актуальная версия -->
   </dependency>
   <!-- Также нам понадобится SLF4J для логирования -->
   <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-api</artifactId>
       <version>2.0.7</version>
   </dependency>
   <dependency>
       <groupId>org.slf4j</groupId>
       <artifactId>slf4j-simple</artifactId>
       <version>2.0.7</version>
   </dependency>
   

2. Конфигурация продюсера (Producer Properties): Продюсеру нужно знать, куда подключаться и как преобразовывать ваши данные в байты. Основные настройки:

   • bootstrap.servers: Адрес одного или нескольких брокеров Kafka (например, localhost:9092).
   • key.serializer: Класс, который будет сериализовать ключ сообщения (например, StringSerializer для строковых ключей).
   • value.serializer: Класс для сериализации значения сообщения.

3. Создание объектов:

   • KafkaProducer: Основной объект, который управляет соединением и отправкой сообщений.
   • ProducerRecord: Объект, представляющий одно сообщение. Он содержит имя топика, а также опционально ключ, значение и партицию.

4. Отправка и завершение работы:

   • producer.send(record): Асинхронно отправляет сообщение. Метод не ждет подтверждения от брокера, а сразу возвращает управление.
   • producer.flush(): Принудительно отправляет все сообщения, которые могут находиться во внутреннем буфере продюсера.
   • producer.close(): Закрывает соединение с Kafka, предварительно вызвав flush().

Чтобы систематизировать эти шаги, изучите следующее руководство.

2. Практический эксперимент: Пишем код

Теперь давайте напишем код, который позволит нам провести два эксперимента: отправку сообщений без ключа и с ключом. Для проверки, в какую партицию попало сообщение, мы будем использовать Callback — механизм, позволяющий асинхронно получать метаданные об отправленном сообщении.

Подготовка окружения

1. Убедитесь, что Zookeeper и Kafka запущены.

2. Создайте топик с несколькими партициями. Назовем его demo_java и создадим с 3 партициями:

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

Код продюсера

Создайте Java-класс ProducerDemo.java и вставьте в него следующий код. Он объединяет оба наших эксперимента.

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class ProducerDemo {

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

    public static void main(String[] args) throws InterruptedException {
        log.info("I am a Kafka Producer!");

        // 1. Create Producer Properties
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 2. Create the Producer
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        // --- ЭКСПЕРИМЕНТ 1: Отправка сообщений БЕЗ ключа ---
        log.info("\n--- Sending messages without a key ---");
        for (int i = 0; i < 10; i++) {
            String topic = "demo_java";
            String value = "message without key " + i;

            ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, value);

            // Send data - asynchronous
            producer.send(producerRecord, (metadata, exception) -> {
                // Executes every time a record is successfully sent or an exception is thrown
                if (exception == null) {
                    // The record was successfully sent
                    log.info("Received new metadata. \n" +
                            "Topic: " + metadata.topic() + "\n" +
                            "Partition: " + metadata.partition() + "\n" +
                            "Offset: " + metadata.offset() + "\n" +
                            "Timestamp: " + metadata.timestamp());
                } else {
                    log.error("Error while producing", exception);
                }
            });
        }
        // Даем время на отправку батча
        producer.flush();
        Thread.sleep(1000);


        // --- ЭКСПЕРИМЕНТ 2: Отправка сообщений С ключом ---
        log.info("\n--- Sending messages with a key ---");
        for (int i = 0; i < 10; i++) {
            String topic = "demo_java";
            String value = "message with key " + i;
            // Используем несколько повторяющихся ключей
            String key = "id_" + (i % 3); // Keys will be id_0, id_1, id_2, id_0, id_1, ...

            ProducerRecord<String, String> producerRecord = new ProducerRecord<>(topic, key, value);

            producer.send(producerRecord, (metadata, exception) -> {
                if (exception == null) {
                    log.info("Received new metadata. \n" +
                            "Topic: " + metadata.topic() + "\n" +
                            "Key: " + producerRecord.key() + "\n" + // Логируем ключ
                            "Partition: " + metadata.partition() + "\n" +
                            "Offset: " + metadata.offset() + "\n" +
                            "Timestamp: " + metadata.timestamp());
                } else {
                    log.error("Error while producing", exception);
                }
            });
        }

        // 3. Flush and close the Producer
        // flush data - synchronous
        log.info("Flushing messages...");
        producer.flush();
        // flush and close producer
        producer.close();
        log.info("Producer closed.");
    }
}

Запуск и анализ

Запустите main метод. В консоли вы увидите логи отправки.

• Анализ эксперимента 1 (без ключа):

  • Обратите внимание, что первые несколько сообщений, скорее всего, отправятся в одну и ту же партицию. Это и есть "липкий" партиционировщик (Sticky Partitioner) в действии. Он делает это для оптимизации, заполняя один батч для одной партиции перед переключением.
  • Задание: Раскомментируйте или добавьте Thread.sleep(500); внутри первого цикла for. Запустите код снова. Теперь вы должны увидеть, что сообщения распределяются по разным партициям (0, 1, 2, 0, 1, ...). Пауза заставляет продюсер отправлять неполные батчи, и для каждого нового батча он выбирает новую партицию.

• Анализ эксперимента 2 (с ключом):

  • Посмотрите на логи для сообщений с ключами. Вы увидите, что все сообщения с ключом id_0 попали в одну и ту же партицию. Все сообщения с id_1 — в другую (но тоже всегда в одну и ту же), и так далее.
  • Запустите код несколько раз. Номер партиции для конкретного ключа (id_0 -> Partition X) не изменится. Это и есть детерминированное партиционирование, которое гарантирует порядок.

3. Закрепление концепций

Теперь, когда вы увидели поведение продюсера вживую, давайте вернемся к статьям, чтобы закрепить теоретическое объяснение этих явлений.

В качестве дополнительного подтверждения можно обратиться к статье на Baeldung. Она демонстрирует те же концепции, но с использованием TDD-подхода, что может быть вам близко, учитывая ваш опыт.

Заключение

Отлично! Сегодня вы сделали важный шаг от теории к практике, создав свой первый Kafka-продюсер на Java.

Ключевые выводы:

• Мы научились настраивать и создавать KafkaProducer с помощью kafka-clients.
• На практике увидели работу Sticky Partitioner для сообщений без ключа и как он оптимизирует отправку, группируя сообщения в батчи для одной партиции.
• Мы эмпирически подтвердили, что все сообщения с одинаковым ключом гарантированно отправляются в одну и ту же партицию.
• Использование Callback при отправке — удобный способ получить метаданные (включая партицию и офсет) и убедиться, что сообщение доставлено успешно.

Что дальше?

Мы научились надежно отправлять сообщения. В следующем уроке мы построим вторую половину системы: напишем Java-консьюмер, который будет эти сообщения читать. Мы разберем одну из ключевых идей Kafka — группы консьюмеров (Consumer Groups) — и посмотрим, как они позволяют распараллеливать обработку данных.