Перейти к основному содержимому

Руководство по Spring Cloud Stream с Kafka, Apache Avro и Confluent Schema Registry

· 9 мин. чтения

1. Введение​

Apache Kafka — это платформа для обмена сообщениями. С его помощью мы можем обмениваться данными между различными приложениями в любом масштабе.

Spring Cloud Stream — это платформа для создания приложений, управляемых сообщениями. Это может упростить интеграцию Kafka в наши сервисы.

Обычно Kafka используется с форматом сообщений Avro, поддерживаемым реестром схем. В этом руководстве мы будем использовать реестр Confluent Schema. Мы попробуем реализацию интеграции Spring с реестром Confluent Schema Registry, а также собственные библиотеки Confluent.

2. Реестр слитных схем​

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

Confluent Schema Registry предоставляет простой способ хранения, извлечения и управления схемами. Он предоставляет несколько полезных RESTful API .

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

Каждый производитель будет знать схему, которую он создает, и каждый потребитель должен иметь возможность либо потреблять данные в ЛЮБОМ формате, либо должен иметь определенную схему, в которой он предпочитает читать. Производитель консультируется с реестром, чтобы установить правильный идентификатор для использования при отправке. сообщение. Потребитель использует реестр для получения схемы отправителя.

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

3. Апач Авро​

Apache Avro — это система сериализации данных .

Он использует структуру JSON для определения схемы, обеспечивая сериализацию между байтами и структурированными данными.

Одной из сильных сторон Avro является поддержка преобразования сообщений, написанных в одной версии схемы, в формат, определяемый совместимой альтернативной схемой.

Набор инструментов Avro также может генерировать классы для представления структур данных этих схем, упрощая сериализацию в POJO и из них.

4. Настройка проекта​

Чтобы использовать реестр схем с Spring Cloud Stream , нам потребуются зависимости Spring Cloud Kafka Binder и реестра схем от Maven:

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-schema</artifactId>
</dependency>

Для сериализатора Confluent нам нужно:

<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-avro-serializer</artifactId>
    <version>4.0.0</version>
</dependency>

И сериализатор Confluent находится в их репо:

<repositories>
    <repository>
        <id>confluent</id>
        <url>https://packages.confluent.io/maven/</url>
    </repository>
</repositories>

Кроме того, давайте воспользуемся подключаемым модулем Maven для создания классов Avro:

<build>
    <plugins>
        <plugin>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro-maven-plugin</artifactId>
            <version>1.8.2</version>
            <executions>
                <execution>
                    <id>schemas</id>
                    <phase>generate-sources</phase>
                    <goals>
                        <goal>schema</goal>
                        <goal>protocol</goal>
                        <goal>idl-protocol</goal>
                    </goals>
                    <configuration>                        
                        <sourceDirectory>${project.basedir}/src/main/resources/</sourceDirectory>
                        <outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

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

5. Весенний облачный поток​

Теперь, когда мы настроили наш проект, давайте напишем производителя, используя Spring Cloud Stream . Он будет публиковать сведения о сотрудниках по теме.

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

5.1. Схема​

Во-первых, давайте определим схему для сведений о сотрудниках. Мы можем назвать его employee-schema.avsc .

Мы можем сохранить файл схемы в src/main/resources:

{
    "type": "record",
    "name": "Employee",
    "namespace": "com.foreach.schema",
    "fields": [
    {
        "name": "id",
        "type": "int"
    },
    {
        "name": "firstName",
        "type": "string"
    },
    {
        "name": "lastName",
        "type": "string"
    }]
}

После создания приведенной выше схемы нам нужно собрать проект. Затем генератор кода Apache Avro создаст POJO с именем Employee в пакете com.foreach.schema .

5.2. Режиссер​

Spring Cloud Stream предоставляет интерфейс процессора . Это дает нам выходной и входной канал.

Давайте воспользуемся этим, чтобы создать производителя, который отправляет объекты Employee в топик Kafka с информацией о сотрудниках :

@Autowired
private Processor processor;

public void produceEmployeeDetails(int empId, String firstName, String lastName) {

    // creating employee details
    Employee employee = new Employee();
    employee.setId(empId);
    employee.setFirstName(firstName);
    employee.setLastName(lastName);

    Message<Employee> message = MessageBuilder.withPayload(employee)
                .build();

    processor.output()
        .send(message);
}

5.2. Потребитель​

Теперь давайте напишем нашего потребителя:

@StreamListener(Processor.INPUT)
public void consumeEmployeeDetails(Employee employeeDetails) {
    logger.info("Let's process employee details: {}", employeeDetails);
}

Этот потребитель будет читать события, опубликованные в теме сведений о сотрудниках . Давайте направим его вывод в журнал, чтобы посмотреть, что он делает.

5.3. Привязки Кафки​

До сих пор мы работали только с входными и выходными каналами нашего объекта Processor . Эти каналы необходимо настроить с правильными пунктами назначения.

Давайте используем application.yml для предоставления привязок Kafka:

spring:
  cloud:
    stream: 
      bindings:
        input:
          destination: employee-details
          content-type: application/*+avro
        output:
          destination: employee-details
          content-type: application/*+avro

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

5.4. Точка входа​

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

@Autowired
private AvroProducer avroProducer;

@PostMapping("/employees/{id}/{firstName}/{lastName}")
public String producerAvroMessage(@PathVariable int id, @PathVariable String firstName, 
  @PathVariable String lastName) {
    avroProducer.produceEmployeeDetails(id, firstName, lastName);
    return "Sent employee details to consumer";
}

5.5. Включить реестр Confluent Schema и привязки​

Наконец, чтобы наше приложение применяло привязки реестра Kafka и схемы, нам нужно добавить @EnableBinding и @EnableSchemaRegistryClient в один из наших классов конфигурации:

@SpringBootApplication
@EnableBinding(Processor.class)
// The @EnableSchemaRegistryClient annotation needs to be uncommented to use the Spring native method.
// @EnableSchemaRegistryClient
public class AvroKafkaApplication {

    public static void main(String[] args) {
        SpringApplication.run(AvroKafkaApplication.class, args);
    }

}

И мы должны предоставить bean-компонент ConfluentSchemaRegistryClient :

@Value("${spring.cloud.stream.kafka.binder.producer-properties.schema.registry.url}")
private String endPoint;

@Bean
public SchemaRegistryClient schemaRegistryClient() {
    ConfluentSchemaRegistryClient client = new ConfluentSchemaRegistryClient();
    client.setEndpoint(endPoint);
    return client;
}

EndPoint — это URL-адрес реестра Confluent Schema.

5.6. Тестирование нашего сервиса​

Давайте протестируем сервис с помощью POST-запроса:

curl -X POST localhost:8080/employees/1001/Harry/Potter

Журналы говорят нам, что это сработало:

2019-06-11 18:45:45.343  INFO 17036 --- [container-0-C-1] com.foreach.consumer.AvroConsumer       : Let's process employee details: {"id": 1001, "firstName": "Harry", "lastName": "Potter"}

5.7. Что произошло во время обработки?​

Попробуем понять, что именно произошло с нашим примером приложения:

  1. Производитель создал сообщение Kafka, используя объект Employee .
  2. Производитель зарегистрировал схему сотрудника в реестре схем, чтобы получить идентификатор версии схемы, это либо создает новый идентификатор, либо повторно использует существующий для этой точной схемы.
  3. Avro сериализовал объект Employee , используя схему
  4. Spring Cloud поместил идентификатор схемы в заголовки сообщений.
  5. Сообщение опубликовано в теме
  6. Когда сообщение пришло к потребителю, он прочитал идентификатор схемы из заголовка.
  7. Потребитель использовал schema-id для получения схемы Employee из реестра.
  8. Потребитель нашел локальный класс, который мог бы представлять этот объект, и десериализовал в него сообщение.

6. Сериализация/десериализация с использованием собственных библиотек Kafka​

Spring Boot предоставляет несколько готовых конвертеров сообщений. По умолчанию Spring Boot использует заголовок Content-Type для выбора соответствующего преобразователя сообщений.

В нашем примере Content-Type — это application/*+avro, поэтому он использовал AvroSchemaMessageConverter для чтения и записи форматов Avro. Но Confluent рекомендует использовать KafkaAvroSerializer и KafkaAvroDeserializer для преобразования сообщений .

Хотя собственный формат Spring работает хорошо, у него есть некоторые недостатки с точки зрения разбиения на разделы, и он несовместим со стандартами Confluent, что может потребоваться некоторым службам, отличным от Spring, в нашем экземпляре Kafka.

Давайте обновим наш application.yml , чтобы использовать преобразователи Confluent:

spring:
  cloud:
    stream:
      default: 
        producer: 
          useNativeEncoding: true
        consumer:  
          useNativeEncoding: true     
      bindings:
        input:
          destination: employee-details
          content-type: application/*+avro
        output:
          destination: employee-details
          content-type: application/*+avro
      kafka:
         binder:        
           producer-properties:
             key.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
             value.serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
             schema.registry.url: http://localhost:8081 
           consumer-properties:
             key.deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
             value.deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
             schema.registry.url: http://localhost:8081
             specific.avro.reader: true

Мы включили useNativeEncoding . Это заставляет Spring Cloud Stream делегировать сериализацию предоставленным классам.

Мы также должны знать, как мы можем предоставить собственные свойства настроек для Kafka в Spring Cloud, используя kafka.binder.producer-properties и kafka.binder.consumer-properties.

7. Группы потребителей и разделы​

Группы потребителей — это набор потребителей, принадлежащих одному и тому же приложению . Потребители из одной и той же группы потребителей имеют одно и то же имя группы.

Давайте обновим application.yml , чтобы добавить имя группы потребителей:

spring:
  cloud:
    stream:
      // ...     
      bindings:
        input:
          destination: employee-details
          content-type: application/*+avro
          group: group-1
      // ...

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

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

7.1. Ключ раздела​

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

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

Мы можем предоставить ключ раздела при отправке сообщения в тему. Сообщения с одним и тем же ключом раздела всегда будут отправляться в один и тот же раздел . Если ключ раздела отсутствует, сообщения будут разбиты на разделы в циклическом режиме.

Попробуем понять это на примере. Представьте, что мы получаем несколько сообщений для сотрудника и хотим обработать все сообщения сотрудника в последовательности. Название отдела и идентификатор сотрудника могут однозначно идентифицировать сотрудника.

Итак, давайте определим ключ раздела с идентификатором сотрудника и названием отдела:

{
    "type": "record",
    "name": "EmployeeKey",
    "namespace": "com.foreach.schema",
    "fields": [
     {
        "name": "id",
        "type": "int"
    },
    {
        "name": "departmentName",
        "type": "string"
    }]
}

После сборки проекта POJO EmployeeKey будет сгенерирован в пакете com.foreach.schema .

Давайте обновим нашего производителя, чтобы использовать EmployeeKey в качестве ключа раздела:

public void produceEmployeeDetails(int empId, String firstName, String lastName) {

    // creating employee details
    Employee employee = new Employee();
    employee.setId(empId);
    // ...

    // creating partition key for kafka topic
    EmployeeKey employeeKey = new EmployeeKey();
    employeeKey.setId(empId);
    employeeKey.setDepartmentName("IT");

    Message<Employee> message = MessageBuilder.withPayload(employee)
        .setHeader(KafkaHeaders.MESSAGE_KEY, employeeKey)
        .build();

    processor.output()
        .send(message);
}

Здесь мы помещаем ключ раздела в заголовок сообщения.

Теперь тот же раздел будет получать сообщения с тем же идентификатором сотрудника и названием отдела.

7.2. Потребительский параллелизм​

Spring Cloud Stream позволяет нам установить параллелизм для потребителя в application.yml :

spring:
  cloud:
    stream:
      // ... 
      bindings:
        input:
          destination: employee-details
          content-type: application/*+avro
          group: group-1
          concurrency: 3

Теперь наши потребители будут читать три сообщения из топика одновременно. Другими словами, Spring создаст три разных потока для независимого использования.

8. Заключение​

В этой статье мы интегрировали производитель и потребитель против Apache Kafka со схемами Avro и Confluent Schema Registry .

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

Мы рассмотрели, как использовать реализацию клиента Avro и Schema Registry в Spring, а затем увидели, как переключиться на стандартную реализацию сериализации и десериализации Confluent для целей взаимодействия.

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

Полный код, использованный в этой статье, можно найти на GitHub .