Уголок сельского джависта

Отправка сообщений в Kafka

Alexander Kosarev·3 Dec 2024

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

Зависимости

Для работы с Kafka в проекте на языке программирования для платформы Java нам потребуется клиент, который мы можем добавить в зависимости проекта:

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

Настройка отправителя

Первым делом для отправки сообщений нужно сконфигурировать отправителя в проекте.

Минимальная настройка отправителя

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

class KafkaSender {

  public static void main(String[] args){
    var properties = new Properties();

    // Адреса для подключения
    properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
        "localhost:29092,localhost:39092,localhost:49092");

    // Сериализатор ключей
    properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
        StringSerializer.class.getName());

    // Сериализатор значений
    properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
        StringSerializer.class.getName());

    try (var producer = new KafkaProducer&lt;String, String>(producerProperties)) {
      // Здесь можно отправлять сообщения
    }
  }
}

Клас KafkaProducer является типизированным, K определяет тип ключей сообщений, а V — их содержимого.

Клиент Kafka предоставляет следующие сериализаторы:

  • BooleanSerializer
  • ByteArraySerializer
  • ByteBufferSerializer
  • BytesSerializer
  • DoubleSerializer
  • FloatSerializer
  • IntegerSerializer
  • ListSerializer
  • LongSerializer
  • ShortSerializer
  • StringSerializer
  • UUIDSerializer
  • VoidSerializer

Все сериализаторы сериализуют данные исходного типа в массив байтов. Вы можете создать и использовать собственный сериализатор, реализующий интерфейс Serializer.

Настройка транзакционного отправителя

Для использования транзакций при отправке сообщений необходимо указать в настройках транзакционный идентификатор отправителя (transactional.id):

class KafkaSender {

  public static void main(String[] args){
    var properties = new Properties();
    properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
        "localhost:29092,localhost:39092,localhost:49092");
    properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
        StringSerializer.class.getName());
    properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
        StringSerializer.class.getName());

    // Транзакционный идентификатор
    properties.setProperty(ProducerConfig.TRANSACTIONAL_ID_CONFIG,
        UUID.randomUUID().toString());

    try (var producer = new KafkaProducer&lt;String, String>(producerProperties)) {
      // Здесь можно отправлять сообщения
    }
  }
}

Транзакции могут использовать только идемпонентные отправители.

Настройка неидемпонентного отправителя

Начиная с Kafka 3.0.0 все отправители по умолчанию являются идемпонентными и ожидают подтверждения о доставки сообщений от всех реплик. Идемпонентные отправители гарантируют уникальность отправляемых сообщений в партиции в рамках сеанса работы отправителя.

Если вам не требуется идемпонентность, то её можно отключить параметром enable.idempotence=false, количество подтверждений указать при помощи параметра acks, который может принимать следующие значения:

  • all — ожидание подтверждения доставки сообщения всем репликам (по умолчанию)
  • 0 — не ожидать подтверждений вообще
  • 1 — ожидать подтверждения от лидирующей реплики
  • Любое валидное количество реплик
class KafkaSender {

  public static void main(String[] args){
    var properties = new Properties();
    properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
        "localhost:29092,localhost:39092,localhost:49092");
    properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
        StringSerializer.class.getName());
    properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
        StringSerializer.class.getName());

    // Отключение ожидания подтверждения при отправке сообщений
    properties.setProperty(ProducerConfig.ACKS_CONFIG, "0");

    // Отключение идемпотентности
    // Можно не делать явно, т.к. она отключается при acks отличном от all
    properties.setProperty(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,
        Boolean.FALSE.toString());

    try (var producer = new KafkaProducer<String, String>(producerProperties)) {
      // Здесь можно отправлять сообщения
    }
  }
}

Отправка сообщений

Для отправки сообщений в KafkaProducer существуют две версии метода send:

  • Future<RecordMetadata> send(ProducerRecord<K, V> record)
  • Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback)

Отправка сообщений в Kafka асинхронная, как видно из Future, возвращаемого методом send.

Класс ProducerRecord содержит свойства отправляемого сообщения:

  • topic — название целевого топика
  • partition — целевая партиция топика
  • headers - заголовки сообщения
  • key — ключ сообщения
  • value — само сообщение
  • timestamp — метка времени

Для удобства использования ProducerRecord предоставляет несколько перегруженных конструкторов, минимальный из которых принимает только название топика (topic) и сообщение (value), максимальный — все указанные свойства.

Методу send может быть передан экземпляр класса Callback, который содержит единственный метод void onCompletion(RecordMetadata metadata, Exception exception), вызываемый по окончании отправки сообщения.

В случае успеха отправки вы получите экземпляр класса RecordMetadata, который содержит метаданные добавленного сообщения:

  • offset — смещение добавленного сообщения
  • timestamp — метка времени сообщения
  • serializedKeySize — размер сериализованного ключа в байтах
  • serializedValueSize — размер сериализованного записанного сообщения в байтах
  • topicPartition — партиция, в которую было записано сообщение

Получить экземпляр класса RecordMetadata вы можете из возвращаемого Future или в Callback.

Давайте разберём несколько вариантов отправки сообщений:

class KafkaSender {

  public static void main(String[] args){
    try (var producer = getProducer()) {
      // Простейшая асинхронная отправка сообщения
      producer.send(new ProducerRecord<>("topic1", "Hello Kafka!"));

      // Отправка с ожиданием результата
      var recordMetadata = producer.send(new ProducerRecord<>("topic1", "Hello Kafka!"))
          .get();

      // Асинхронная отправка с использованием callback
      producer.send(new ProducerRecord<>("topic1", "Hello Kafka!"),
          (recordMetadata, exception) -> {
            if (exception != null) {
              LOGGER.error(exception.getMessage(), exception);
            } else {
              // обработка успешного результата
            }
          });

      // Отправка сообщения с точным указанием свойств
      producer.send(
          new ProducerRecord<>(
              // Топик
              "topic1",
              // Партиция
              1,
              // Метка времени
              System.currentTimeMillis(),
              // Ключ
              "key1",
              // Содержимое
              "Hello Kafka!",
              // Заголовки
              List.of(new RecordHeader("foo", "bar".getBytes()))
          )
      );
    } catch(ExecutionException executionException) {
    }
  }
}

Если нам требуется отправить несколько сообщений и нужны гарантии доставки всех сообщений, то мы можем сделать это при помощи транзакций:

public class KafkaSender {

  private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProducerApplication.class);

  public static void main(String[] args) throws InterruptedException {
    try (var producer = getProducer()) {
      try {
        // Инициализация транзакции
        producer.initTransactions();
        // Начало транзакции
        producer.beginTransaction();
        producer.send(new ProducerRecord<>("topic1", "message1"));
        producer.send(new ProducerRecord<>("topic2", "message3"));
        producer.send(new ProducerRecord<>("topic3", "message4"));
        producer.send(new ProducerRecord<>("topic1", "message2"));
        // Фиксация транзакции
        producer.commitTransaction();
      } catch (Exception exception) {
        LOGGER.error(exception.getMessage(), exception);
        // Откат транзакции в случае ошибки
        producer.abortTransaction();
      }
    }
  }
}
105 views·7 shares
105 views