Spring Boot приложение с использованием Kafka Streams

Привет, Хабр! В этой статье мы рассмотрим, как в МегаФоне производят потоковую обработку данных, и разработаем простое Spring Boot приложение с использованием Kafka Streams.


У нас в компании есть домен «Фабрика микросервисов» (MFactory) — подразделение, которое занимается исследованием и разработкой приложений на базе микросервисной архитектуры. Для того, чтобы микросервисы могли «общаться» между собой, мы используем брокеры сообщений RabbitMQ и Kafka, но сегодня мы поговорим подробнее о Kafka, так как именно этот брокер реализует очень мощный инструмент для потоковой обработки данных — Kafka Streams.

Рассмотрим стандартный подход к обработке данных: получаем поток данных, помещаем их в базу данных, после чего подключаемся к базе, выгружаем данные и обрабатываем. Но существует ряд задач, в которых мы можем увеличить производительность, обрабатывая поток данных в режиме реального времени, минуя базу данных. Для их решения используем интерфейс Kafka Streams. Но для начала, вспомним, что такое Kafka.

Kafka — это распределённый программный брокер сообщений. «Распределённый» значит, что брокер может работать синхронно на нескольких серверах, образуя кластер. Kafka состоит из генератора сообщений, потребителя сообщений и топика. Топик — это файл, в котором входящие записи добавляются в конец файла. Другими словами, топик — это упорядоченный по времени журнал сообщений.

Рассмотрим следующую задачу. Мы получаем поток данных с информацией об изменении баланса абонента в формате JSON — {«phone_number»:number,«balance»:balance} и записываем его в топик “src”. И нам нужно в реальном времени отфильтровать сообщения, в которых баланс меньше либо равен нулю.

Теперь рассмотрим, что же такое Kafka Streams. Apache Kafka Streams API — это клиентская библиотека для создания приложений и микросервисов, которые позволяют в режиме реального времени обрабатывать данные, хранящиеся в кластерах Kafka. Kafka Streams сочетает в себе простоту написания и развертывания стандартных приложений Java и Scala на стороне клиента с преимуществами серверной кластерной технологии Kafka.

Устройство Kafka Streams можно представить в виде графа (топологии обработки):

1. узел-обработчик, который подписывается на топик и вычитывает входящие сообщения;
2. узел-обработчик, который реализует бизнес-логику;
3. узел-обработчик, который отправляет результат обработки в топик.

Представим следующую задачу. Мы получаем поток информации об изменении баланса абонентов в формате JSON вида {«phone_number»:number,«balance»:balance} и записываем его в топик “src”. Нам нужно в реальном времени отфильтровать сообщения, в которых баланс меньше и равен нулю, и записать результат в топик “out”, на который подписан сервис для обработки данных сообщений. Сымитируем подобный поток, используя консольный генератор Kafka:

PS C:\Users\User> cd C:\kafka_2.12-2.5.0\ PS C:\kafka> bin\windows\kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic src Created topic src. PS C:\kafka> bin\windows\kafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic out Created topic out. PS C:\kafka> bin\windows\kafka-topics.bat --list --bootstrap-server localhost:9092 out src PS C:\kafka> bin\windows\kafka-console-producer.bat --bootstrap-server localhost:9092 --topic src >{"phone_number":79301,"balance":100} >{"phone_number":79302,"balance":0} >{"phone_number":79303,"balance":-50} >

Теперь разработаем простое Spring Boot приложение и посмотрим на Kafka Streams в действии. Для начала определим следующие зависимости:

<dependency>     <groupId>org.apache.kafka</groupId>     <artifactId>kafka-streams</artifactId> </dependency> <dependency>     <groupId>org.springframework.kafka</groupId>     <artifactId>spring-kafka</artifactId> </dependency>

Создадим объект User:

package ru.megafon.kafkastreamsdemo.model;  import com.fasterxml.jackson.annotation.JsonProperty; import lombok.Getter;  @Getter public class User {     @JsonProperty("phone_number")     private Long phoneNumber;     @JsonProperty("balance")     private Double balance; }

Напишем конфигурацию нашего приложения:

package ru.megafon.kafkastreamsdemo.config;  import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.kstream.Consumed; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Produced; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.annotation.EnableKafka; import org.springframework.kafka.annotation.EnableKafkaStreams; import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration; import org.springframework.kafka.config.KafkaStreamsConfiguration; import org.springframework.kafka.support.serializer.JsonDeserializer; import org.springframework.kafka.support.serializer.JsonSerializer; import ru.megafon.kafkastreamsdemo.model.User;  import java.util.HashMap; import java.util.Map;  @Slf4j @Configuration @EnableKafka @EnableKafkaStreams public class KafkaStreamsConfig {     @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)     public KafkaStreamsConfiguration kStreamsConfigs() {         Map<String, Object> props = new HashMap<>();         props.put(StreamsConfig.APPLICATION_ID_CONFIG, "id");         props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");         props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());         props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());         return new KafkaStreamsConfiguration(props);     }      @Bean     public Serde<User> userSerde() {         return Serdes.serdeFrom(new JsonSerializer<>(), new JsonDeserializer<>(User.class));     }      @Bean     public KStream<String, User> kStream(StreamsBuilder kStreamBuilder) {         KStream<String, String> stream = kStreamBuilder                 .stream("src", Consumed.with(Serdes.String(), Serdes.String()));         KStream<String, User> userStream = stream                 .mapValues(this::getUserFromString)                 .filter((key, value) -> value.getBalance() <= 0);         userStream.to("out", Produced.with(Serdes.String(), userSerde()));         return userStream;     }      @Bean     public ObjectMapper objectMapper() {         return new ObjectMapper();     }      User getUserFromString(String userString) {         User user = null;         try {             user = objectMapper().readValue(userString, User.class);         } catch (JsonProcessingException e) {             log.error(e.getMessage(), e);         }         return user;     } }

Рассмотрим подробнее каждый метод:

В методе kStreamsConfigs мы создаём минимальную конфигурацию для подключения нашего приложения к серверу Kafka: устанавливаем идентификатор нашего приложения и адрес сервера брокера, дефолтный сериализатор/десериализатор.
В методе userSerde мы определили собственный сериализатор/десериализатор для модели User.
В методе kStream мы создаём топологию нашего Kafka Streams приложения и возвращаем объект KStream<String, User>. Основным объектом для создания топологии является объект StreamsBuilder:

1. в методе stream мы подписываемся на топик “src” и вычитываем входящие сообщения;
2. в методе filter мы реализовали нашу бизнес-логику — фильтрацию;
3. в методе to мы отправляем результат обработки в результирующий топик “out”.

Запускаем приложение и подписываемся на топик “out”:

PS C:\Users\User> cd C:\kafka_2.12-2.5.0\ PS C:\kafka> bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic out --from-beginning {"phone_number":79302,"balance":0.0} {"phone_number":79303,"balance":-50.0} 

Мы видим, что данные успешно обработаны.

Преимущества Kafka Streams:

• простая и легковесная библиотека, которая может быть легко встроена в любое Java-приложение и интегрирована с инструментами упаковки, развертывания и эксплуатации;
• не имеет внешних зависимостей от систем кроме самого Apache Kafka;
• поддерживает отказоустойчивое локальное состояние, что обеспечивает очень быстрые и эффективные операции с сохранением состояния;
• поддерживает правило “обработка ровно один раз”, чтобы гарантировать, что каждая запись будет обработана один и только один раз, даже если в процессе обработки происходит сбой на клиентах Kafka Streams или брокерах Kafka;
• использует обработку “по одной записи за раз” для достижения миллисекундной задержки.