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

1. Создание сообщений:
   Core API предлагает методы публикации сообщений в темах Kafka. Вот пример использования класса ProducerRecordдля создания и отправки сообщения:

import org.apache.kafka.clients.producer.*;
public class KafkaProducerExample {
    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");
        Producer<String, String> producer = new KafkaProducer<>(props);
        String topic = "my_topic";
        String key = "my_key";
        String value = "Hello, Kafka!";
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
        producer.send(record);
        producer.close();
    }
}

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

import org.apache.kafka.clients.consumer.*;
public class KafkaConsumerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("group.id", "my_consumer_group");
        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("my_topic"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Received message: " + record.value());
            }
        }
    }
}

3. Настройка свойств Kafka:
   Core API предоставляет методы для настройки различных свойств Kafka, таких как брокеры, сериализаторы, десериализаторы и группы потребителей. Вот пример настройки производителя с настраиваемыми свойствами:

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");
Producer<String, String> producer = new KafkaProducer<>(props);

4. Обработка ошибок и повторных попыток.
   Core API поддерживает механизмы обработки ошибок и повторных попыток. Например, для обработки ошибок во время создания сообщения вы можете использовать интерфейс Callback:

producer.send(record, new Callback() {
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (exception != null) {
            System.err.println("Error sending message: " + exception.getMessage());
        } else {
            System.out.println("Message sent successfully!");
        }
    }
});

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