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

1. Распределенная архитектура:

- Брокер: кластер Kafka состоит из одного или нескольких брокеров, и каждый брокер отвечает за хранение и распространение сообщений на нем. Сообщения организованы в виде тем, и каждая тема может быть разделена на несколько разделов.

2. Темы и разделы:

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

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

3. Производители и потребители:

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

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

4. Механизм сохранения и копирования сообщений:

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

- Репликация: каждый раздел имеет несколько реплик, распределенных по разным брокерам, одна из которых является главной репликой (лидер), а остальные — подчиненными репликами (ведомый). Первичная реплика отвечает за обработку запросов на чтение и запись, а ведомая реплика пассивно синхронизирует данные из первичной реплики. Такая конструкция обеспечивает отказоустойчивость. В случае сбоя первичной реплики кластер может автоматически выбрать новую первичную реплику для продолжения обслуживания.

5. Управление ходом компенсации и потребления:

- Смещение: каждая группа потребителей сохраняет прогресс потребления (смещение) для каждого раздела, указывая положение использованных сообщений. Потребители фиксируют прогресс своего потребления, отправляя Offset в Kafka или во внешнее хранилище (например, ZooKeeper). Это позволяет потребителю продолжать потребление с того места, где он остановился после перезапуска, обеспечивая восстановление после сбоя и семантику доставки сообщения хотя бы один раз.

6. Координация и управление метаданными:

- ZooKeeper (ранняя версия): ранняя версия Kafka использовала ZooKeeper для координации кластера и управления метаданными, включая регистрацию брокера, отслеживание статуса копии раздела, координацию групп потребителей и т. д.

- KRaft (новая версия): начиная с версии Kafka 2.8 протокол KRaft был представлен как дополнительный режим управления кластером, который больше не зависит от ZooKeeper, что упрощает эксплуатацию и обслуживание, а также повышает производительность и доступность.

Сценарии приложений Kafka

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

1. Сбор и анализ журналов. Будучи центральной системой хранения журналов, она собирает данные журналов из различных распределенных систем, а затем предоставляет инструменты анализа журналов (такие как Elasticsearch, Splunk, Hadoop и т. д.) для анализа в режиме реального времени или пакетного анализа.

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

3. Потоковая обработка. Являясь источником входных и выходных данных платформы потоковой обработки, она тесно интегрирована с такими платформами потоковой обработки, как Spark Streaming, Flink и Storm, для выполнения таких операций, как фильтрация, агрегирование и оконный расчет реальных данных. потоки данных времени.

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

5. Мониторинг и подача сигналов тревоги. Собирайте данные мониторинга системы (например, использование ЦП, использование памяти, сетевой трафик и т. д.) для мониторинга состояния системы в реальном времени, запуска сигналов тревоги или дальнейших автоматизированных операций.

6. Синхронизация данных: синхронизируйте данные между несколькими системами, например, сценарии базы данных CDC (Change Data Capture), синхронизируйте события изменения базы данных с Kafka, а затем подпишитесь и используйте их нижестоящими системами для достижения обновлений хранилища данных в реальном времени или перекрестного обмена данными. -согласованность данных системы.

7. Отслеживание поведения пользователей: собирайте данные о поведении пользователей, такие как клики, просмотр и покупки на веб-сайте или в приложении, для получения рекомендаций в режиме реального времени, построения портретов пользователей, анализа маркетинговой активности и т. д.

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

Интеграция Kafka в проект Spring Boot

1. Добавьте зависимости:

Добавьте зависимость Spring Kafka в файл pom.xml (проект Maven) или файл build.gradle (проект Gradle) проекта Spring Boot. Например, для Maven:

<dependencies>

<dependency>

<groupId>org.springframework.kafka</groupId>

<artifactId>spring-kafka</artifactId>

</dependency>

</dependencies>

Для Градла:

dependencies {

implementation 'org.springframework.kafka:spring-kafka'

}

2. Настройте соединение Kafka:

Настройте адрес, тему и другую информацию сервера Kafka в файле application.properties или application.yml:

properties

spring.kafka.bootstrap-servers=localhost:9092

Если требуются другие расширенные настройки, такие как идентификатор группы потребителей, сериализатор и т. д., они также задаются здесь.

3. Создайте продюсера Kafka:

Создайте класс @Configuration и определите bean-компонент KafkaTemplate. KafkaTemplate — удобный инструмент Spring для отправки сообщений в темы Kafka.

@Configuration

public class KafkaConfig {

@Bean

public ProducerFactory<String, String> producerFactory() {

Map<String, Object> config = new HashMap<>();

config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

return new DefaultKafkaProducerFactory<>(config);

}

@Bean

public KafkaTemplate<String, String> kafkaTemplate() {

return new KafkaTemplate<>(producerFactory());

}

}

```

Или, если свойства Spring.kafka.*` настроены, вы можете напрямую внедрить KafkaTemplate:

@Autowired

private KafkaTemplate<String, String> kafkaTemplate;

4. Используйте производитель Kafka для отправки сообщений:

Внедрите KafkaTemplate в службу или контроллер, которому необходимо отправлять сообщения, и вызовите его метод send():

@Service

public class MessageService {

@Autowired

private KafkaTemplate<String, String> kafkaTemplate;

public void sendMessage(String topicName, String message) {

kafkaTemplate.send(topicName, message);

}

}

5. Создайте потребителя Kafka:

Используйте аннотацию @KafkaListener, чтобы отметить метод, который будет автоматически прослушивать сообщения по указанной теме:

@Service

public class MessageConsumer {

@KafkaListener(topics = "my-topic")

public void consume(String message) {

System.out.println("Received message: " + message);

}

}

Выше приведены основные шаги Springboot для интеграции Kafka. При реальном использовании вам также может потребоваться рассмотреть возможность обработки исключений, подтверждения сообщений (acknowledgement), управления группами потребителей, конвертера сообщений (MessageConverter) и т. д., в зависимости от потребностей вашего бизнеса.