Apache Kafka® — это распределенная платформа потоковой передачи. В предыдущем уроке мы обсуждали, как реализовать потребителей и производителей Kafka с помощью Spring .
В этом руководстве мы узнаем, как использовать коннекторы Kafka.
Мы рассмотрим:
В предыдущей статье у нас было краткое введение в Kafka Connect, включая различные типы соединителей, основные функции Connect, а также REST API.
В этом руководстве мы будем использовать коннекторы Kafka для создания более «реального» примера.
Мы будем использовать коннектор для сбора данных через MQTT и записывать собранные данные в MongoDB.
В этом руководстве мы рассмотрим, как Kafka обеспечивает однократную доставку между приложениями-производителями и приложениями-потребителями с помощью недавно представленного API транзакций.
Кроме того, мы будем использовать этот API для реализации транзакционных производителей и потребителей для достижения сквозной однократной доставки в примере WordCount.
Из-за различных сбоев системы обмена сообщениями не могут гарантировать доставку сообщений между приложениями производителя и потребителя. В зависимости от того, как клиентские приложения взаимодействуют с такими системами, возможна следующая семантика сообщений:
В этом руководстве мы рассмотрим MockConsumer , одну из реализаций Kafka Consumer .
Во-первых, мы обсудим, что необходимо учитывать при тестировании Kafka Consumer . Затем мы увидим, как мы можем использовать MockConsumer для реализации тестов.
потребителя KafkaИспользование данных из Kafka состоит из двух основных этапов. Во-первых, мы должны подписываться на темы или назначать разделы тем вручную. Во-вторых, мы опрашиваем пакеты записей методом опроса .
Apache Flink — это платформа обработки потоков, которую можно легко использовать с Java. Apache Kafka — это распределенная система обработки потоков, поддерживающая высокую отказоустойчивость.
В этом руководстве мы рассмотрим, как построить конвейер данных с использованием этих двух технологий.
Чтобы установить и настроить Apache Kafka, обратитесь к официальному руководству . После установки мы можем использовать следующие команды для создания новых тем с именами flink_input и flink_output:
Kafka — это система обработки сообщений, построенная на основе распределенной очереди сообщений. Он предоставляет библиотеку Java, чтобы приложения могли записывать данные или считывать данные из темы Kafka.
Теперь, поскольку большая часть логики бизнес-домена проверяется с помощью модульных тестов, приложения обычно имитируют все операции ввода-вывода в JUnit. Kafka также предоставляет MockProducer для имитации приложения производителя.
В этом руководстве мы сначала реализуем приложение производителя Kafka. Позже мы реализуем модульный тест для проверки общих операций производителя с помощью MockProducer .
В этом руководстве мы кратко представим Apache Kafka , а затем посмотрим, как программно создавать и настраивать разделы в кластере Kafka.
Apache Kafka — это мощная, высокопроизводительная распределенная платформа для потоковой передачи событий.
Как правило, приложения-производители публикуют события в Kafka, а потребители подписываются на эти события, чтобы читать и обрабатывать их. Kafka использует темы для хранения и классификации этих событий, например, в приложении электронной коммерции может быть тема «заказы».
Мониторинг управляемой событиями системы, использующей кластер Apache Kafka, часто требует от нас получения списка активных брокеров. В этом руководстве мы рассмотрим несколько команд оболочки, чтобы получить список активных брокеров в работающем кластере.
Для целей этой статьи давайте воспользуемся приведенным ниже файлом docker-compose.yml для настройки двухузлового кластера Kafka : ``
$ cat docker-compose.yml
---
version: '2'
services:
zookeeper-1:
image: confluentinc/cp-zookeeper:latest
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ZOOKEEPER_TICK_TIME: 2000
ports:
- 2181:2181
kafka-1:
image: confluentinc/cp-kafka:latest
depends_on:
- zookeeper-1
ports:
- 29092:29092
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper-1:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-1:9092,PLAINTEXT_HOST://localhost:29092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
kafka-2:
image: confluentinc/cp-kafka:latest
depends_on:
- zookeeper-1
ports:
- 39092:39092
environment:
KAFKA_BROKER_ID: 2
KAFKA_ZOOKEEPER_CONNECT: zookeeper-1:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-2:9092,PLAINTEXT_HOST://localhost:39092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
Apache Kafka — это мощная распределенная отказоустойчивая платформа для потоковой передачи событий с открытым исходным кодом. Однако, когда мы используем Kafka для отправки сообщений, размер которых превышает установленный лимит, возникает ошибка.
Мы показали , как работать со Spring и Kafka в предыдущем уроке. В этом уроке мы рассмотрим способ отправки больших сообщений с помощью Kafka.
Конфигурация Kafka ограничивает размер сообщений, которые разрешено отправлять. По умолчанию это ограничение составляет 1 МБ. Однако, если требуется отправлять большие сообщения, нам нужно настроить эти конфигурации в соответствии с нашими требованиями.
Чтобы лучше понять, как работает RabbitMQ, нам нужно погрузиться в его основные компоненты.
В этой статье мы рассмотрим обмены, очереди и привязки, а также то, как мы можем объявить их программно в приложении Java.
Как обычно, мы будем использовать клиент Java и официальный клиент для сервера RabbitMQ.