Данная статья будет полезной тем, кто только начал знакомиться с микросервисной архитектурой и с сервисом Apache Kafka. Материал не претендует на подробный туториал, но поможет быстро начать работу с данной технологией. Я расскажу о том, как установить и настроить Kafka на Windows 10. Также мы создадим проект, используя Intellij IDEA и Spring Boot.
Зачем?
Трудности в понимании тех или иных инструментов часто связаны с тем, что разработчик никогда не сталкивался с ситуациями, в которых эти инструменты могут понадобиться. С Kafka всё обстоит точно также. Опишем ситуацию, в которой данная технология будет полезной. Если у вас монолитная архитектура приложения, то разумеется, никакая Kafka вам не нужна. Всё меняется с переходом на микросервисы. По сути, каждый микросервис – это отдельная программа, выполняющая ту или иную функцию, и которая может быть запущена независимо от других микросервисов. Микросервисы можно сравнить с сотрудниками в офисе, которые сидят за отдельными столами и независимо от коллег решают свою задачу. Работа такого распределённого коллектива немыслима без централизованной координации. Сотрудники должны иметь возможность обмениваться сообщениями и результатами своей работы между собой. Именно эту проблему и призвана решить Apache Kafka для микросервисов.
Apache Kafka является брокером сообщений. С его помощью микросервисы могут взаимодействовать друг с другом, посылая и получая важную информацию. Возникает вопрос, почему не использовать для этих целей обычный POST – reqest, в теле которого можно передать нужные данные и таким же образом получить ответ? У такого подхода есть ряд очевидных минусов. Например, продюсер (сервис, отправляющий сообщение) может отправить данные только в виде response’а в ответ на запрос консьюмера (сервиса, получающего данные). Допустим, консьюмер отправляет POST – запрос, и продюсер отвечает на него. В это время консьюмер по каким-то причинам не может принять полученный ответ. Что будет с данными? Они будут потеряны. Консьюмеру снова придётся отправлять запрос и надеяться, что данные, которые он хотел получить, за это время не изменились, и продюсер всё ещё готов принять request.
Apache Kafka решает эту и многие другие проблемы, возникающие при обмене сообщениями между микросервисами. Не лишним будет напомнить, что бесперебойный и удобный обмен данными – одна из ключевых проблем, которую необходимо решить для обеспечения устойчивой работы микросервисной архитектуры.
Установка и настройка ZooKeeper и Apache Kafka на Windows 10
Первое, что надо знать для начала работы — это то, что Apache Kafka работает поверх сервиса ZooKeeper. ZooKeeper — это распределенный сервис конфигурирования и синхронизации, и это всё, что нам нужно знать о нём в данном контексте. Мы должны скачать, настроить и запустить его перед тем, как начать работу с Kafka. Прежде чем начать работу с ZooKeeper, убедитесь, что у вас установлен и настроен JRE.
Скачать свежею версию ZooKeeper можно с официального сайта.
Извлекаем из скаченного архива ZooKeeper`а файлы в какую-нибудь папку на диске.
В папке zookeeper с номером версии, находим папку conf и в ней файл “zoo_sample.cfg”.
Копируем его и меняем название копии на “zoo.cfg”. Открываем файл-копию и находим в нём строчку dataDir=/tmp/zookeeper. Прописываем в данной строчке полный путь к нашей папке zookeeper-х.х.х. У меня это выглядит так: dataDir=C:\ZooKeeper\zookeeper-3.6.0
Теперь добавим системную переменную среды: ZOOKEEPER_HOME = C: ZooKeeper zookeeper-3.4.9 и в конце системной переменной Path добавим запись: ;%ZOOKEEPER_HOME%bin;
Запускаем командную строку и пишем команду:
zkserver
Если всё сделано правильно, вы увидите примерно следующее.
Это означает, что ZooKeeper стартанул нормально. Переходим непосредственно к установке и настройке сервера Apache Kafka. Скачиваем свежую версию с официального сайта и извлекаем содержимое архива: kafka.apache.org/downloads
В папке с Kafka находим папку config, в ней находим файл server.properties и открываем его.
Находим строку log.dirs= /tmp/kafka-logs и указываем в ней путь, куда Kafka будет сохранять логи: log.dirs=c:/kafka/kafka-logs.
В этой же папке редактируем файл zookeeper.properties. Строчку dataDir=/tmp/zookeeper меняем на dataDir=c:/kafka/zookeeper-data, не забывая при этом, после имени диска указывать путь к своей папке с Kafka. Если вы всё сделали правильно, можно запускать ZooKeeper и Kafka.
Для кого-то может оказаться неприятной неожиданностью, что никакого GUI для управления Kafka нет. Возможно, это потому, что сервис рассчитан на суровых нёрдов, работающих исключительно с консолью. Так или иначе, для запуска кафки нам потребуется командная строка.
Сначала надо запустить ZooKeeper. В папке с кафкой находим папку bin/windows, в ней находим файл для запуска сервиса zookeeper-server-start.bat, кликаем по нему. Ничего не происходит? Так и должно быть. Открываем в этой папке консоль и пишем:
start zookeeper-server-start.bat
Опять не работает? Это норма. Всё потому что zookeeper-server-start.bat для своей работы требует параметры, прописанные в файле zookeeper.properties, который, как мы помним, лежит в папке config. Пишем в консоль:
start zookeeper-server-start.bat c:kafkaconfigzookeeper.properties
Теперь всё должно стартануть нормально.
Ещё раз открываем консоль в этой папке (ZooKeeper не закрывать!) и запускаем kafka:
start kafka-server-start.bat c:kafkaconfigserver.properties
Для того, чтобы не писать каждый раз команды в командной строке, можно воспользоваться старым проверенным способом и создать батник со следующим содержимым:
start C:kafkabinwindowszookeeper-server-start.bat C:kafkaconfigzookeeper.properties
timeout 10
start C:kafkabinwindowskafka-server-start.bat C:kafkaconfigserver.properties
Строка timeout 10 нужна для того, чтобы задать паузу между запуском zookeeper и kafka. Если вы всё сделали правильно, при клике на батник должны открыться две консоли с запущенным zookeeper и kafka.Теперь мы можем прямо из командной строки создать продюсера сообщений и консьюмера с нужными параметрами. Но, на практике это может понадобиться разве что для тестирования сервиса. Гораздо больше нас будет интересовать, как работать с kafka из IDEA.
Работа с kafka из IDEA
Мы напишем максимально простое приложение, которое одновременно будет и продюсером и консьюмером сообщения, а затем добавим в него полезные фичи. Создадим новый спринг-проект. Удобнее всего делать это с помощью спринг-инициалайзера. Добавляем зависимости org.springframework.kafka и spring-boot-starter-web
В итоге файл pom.xml должен выглядеть так:
Для того, чтобы отправлять сообщения, нам потребуется объект KafkaTemplate<K, V>. Как мы видим объект является типизированным. Первый параметр – это тип ключа, второй – самого сообщения. Пока оба параметра мы укажем как String. Объект будем создавать в классе-рестконтроллере. Объявим KafkaTemplate и попросим Spring инициализировать его, поставив аннотацию Autowired.
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
В принципе, наш продюсер готов. Всё что осталось сделать – это вызвать у него метод send(). Имеется несколько перегруженных вариантов данного метода. Мы используем в нашем проекте вариант с 3 параметрами — send(String topic, K key, V data). Так как KafkaTemplate типизирован String-ом, то ключ и данные в методе send будут являться строкой. Первым параметром указывается топик, то есть тема, в которую будут отправляться сообщения, и на которую могут подписываться консьюмеры, чтобы их получать. Если топик, указанный в методе send не существует, он будет создан автоматически. Полный текст класса выглядит так.
@RestController
@RequestMapping("msg")
public class MsgController {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
@PostMapping
public void sendOrder(String msgId, String msg){
kafkaTemplate.send("msg", msgId, msg);
}
}
Контроллер мапится на localhost:8080/msg, в теле запроса передаётся ключ и само сообщений.
Отправитель сообщений готов, теперь создадим слушателя. Spring так же позволяет cделать это без особых усилий. Достаточно создать метод и пометить его аннотацией @KafkaListener, в параметрах которой можно указать только топик, который будет слушаться. В нашем случае это выглядит так.
@KafkaListener(topics="msg")
У самого метода, помеченного аннотацией, можно указать один принимаемый параметр, имеющий тип сообщения, передаваемого продюсером.
Класс, в котором будет создаваться консьюмер необходимо пометить аннотацией @EnableKafka.
@EnableKafka
@SpringBootApplication
public class SimpleKafkaExampleApplication {
@KafkaListener(topics="msg")
public void msgListener(String msg){
System.out.println(msg);
}
public static void main(String[] args) {
SpringApplication.run(SimpleKafkaExampleApplication.class, args);
}
}
Так же в файле настроек application.property необходимо указать параметр консьюмера groupe-id. Если этого не сделать, приложение не запустится. Параметр имеет тип String и может быть любым.
spring.kafka.consumer.group-id=app.1
Наш простейший кафка-проект готов. У нас есть отправитель и получатель сообщений. Осталось только запустить. Для начала запускаем ZooKeeper и Kafka с помощью батника, который мы написали ранее, затем запускаем наше приложение. Отправлять запрос удобнее всего с помощью Postman. В теле запроса не забываем указывать параметры msgId и msg.
Если мы видим в IDEA такую картину, значит всё работает: продюсер отправил сообщение, консьюмер получил его и вывел в консоль.
Усложняем проект
Реальные проекты с использованием Kafka конечно же сложнее, чем тот, который мы создали. Теперь, когда мы разобрались с базовыми функциями сервиса, рассмотрим, какие дополнительные возможности он предоставляет. Для начала усовершенствуем продюсера.
Если вы открывали метод send(), то могли заметить, что у всех его вариантов есть возвращаемое значение ListenableFuture<SendResult<K, V>>. Сейчас мы не будем подробно рассматривать возможности данного интерфейса. Здесь будет достаточно сказать, что он нужен для просмотра результата отправки сообщения.
@PostMapping
public void sendMsg(String msgId, String msg){
ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("msg", msgId, msg);
future.addCallback(System.out::println, System.err::println);
kafkaTemplate.flush();
}
Метод addCallback() принимает два параметра – SuccessCallback и FailureCallback. Оба они являются функциональными интерфейсами. Из названия можно понять, что метод первого будет вызван в результате успешной отправки сообщения, второго – в результате ошибки.Теперь, если мы запустим проект, то увидим на консоли примерно следующее:
SendResult [producerRecord=ProducerRecord(topic=msg, partition=null, headers=RecordHeaders(headers = [], isReadOnly = true), key=1, value=Hello, world!, timestamp=null), recordMetadata=msg-0@6]
Посмотрим ещё раз внимательно на нашего продюсера. Интересно, что будет если в качестве ключа будет не String, а, допустим, Long, а в качестве передаваемого сообщения и того хуже – какая-нибудь сложная DTO? Попробуем для начала изменить ключ на числовое значение…
Если мы укажем в продюсере в качестве ключа Long, то приложение нормально запуститься, но при попытке отправить сообщение будет выброшен ClassCastException и будет сообщено, что класс Long не может быть приведён к классу String.
Если мы попробуем вручную создать объект KafkaTemplate, то увидим, что в конструктор в качестве параметра передаётся объект интерфейса ProducerFactory<K, V>, например DefaultKafkaProducerFactory<>. Для того, чтобы создать DefaultKafkaProducerFactory, нам нужно в его конструктор передать Map, содержащий настройки продюсера. Весь код по конфигурации и созданию продюсера вынесем в отдельный класс. Для этого создадим пакет config и в нём класс KafkaProducerConfig.
@Configuration
public class KafkaProducerConfig {
private String kafkaServer="localhost:9092";
@Bean
public Map<String, Object> producerConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
kafkaServer);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
LongSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class);
return props;
}
@Bean
public ProducerFactory<Long, String> producerFactory() {
return new DefaultKafkaProducerFactory<>(producerConfigs());
}
@Bean
public KafkaTemplate<Long, String> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
}
В методе producerConfigs() создаём мапу с конфигурациями и в качестве сериализатора для ключа указываем LongSerializer.class. Запускаем, отправляем запрос из Postman и видим, что теперь всё работает, как надо: продюсер отправляет сообщение, а консьюмер принимает его.
Теперь изменим тип передаваемого значения. Что если у нас не стандартный класс из библиотеки Java, а какой-нибудь кастомный DTO. Допустим такой.
@Data
public class UserDto {
private Long age;
private String name;
private Address address;
}
@Data
@AllArgsConstructor
public class Address {
private String country;
private String city;
private String street;
private Long homeNumber;
private Long flatNumber;
}
Для отправки DTO в качестве сообщения, нужно внести некоторые изменения в конфигурацию продюсера. В качестве сериализатора значения сообщения укажем JsonSerializer.class и не забудем везде изменить тип String на UserDto.
@Configuration
public class KafkaProducerConfig {
private String kafkaServer="localhost:9092";
@Bean
public Map<String, Object> producerConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
kafkaServer);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
LongSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
JsonSerializer.class);
return props;
}
@Bean
public ProducerFactory<Long, UserDto> producerFactory() {
return new DefaultKafkaProducerFactory<>(producerConfigs());
}
@Bean
public KafkaTemplate<Long, UserDto> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
}
Отправим сообщение. В консоль будет выведена следующая строка:
Теперь займёмся усложнением консьюмера. До этого наш метод public void msgListener(String msg), помеченный аннотацией @KafkaListener(topics=«msg») в качестве параметра принимал String и выводил его на консоль. Как быть, если мы хотим получить другие параметры передаваемого сообщения, например, ключ или партицию? В этом случае тип передаваемого значения необходимо изменить.
@KafkaListener(topics="msg")
public void orderListener(ConsumerRecord<Long, UserDto> record){
System.out.println(record.partition());
System.out.println(record.key());
System.out.println(record.value());
}
Из объекта ConsumerRecord мы можем получить все интересующие нас параметры.
Мы видим, что вместо ключа на консоль выводятся какие-то кракозябры. Это потому, что для десериализации ключа по умолчанию используется StringDeserializer, и если мы хотим, чтобы ключ в целочисленном формате корректно отображался, мы должны изменить его на LongDeserializer. Для настройки консьюмера в пакете config создадим класс KafkaConsumerConfig.
@Configuration
public class KafkaConsumerConfig {
@Value("${spring.kafka.bootstrap-servers}")
private String kafkaServer;
@Value("${spring.kafka.consumer.group-id}")
private String kafkaGroupId;
@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaGroupId);
return props;
}
@Bean
public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Long, UserDto> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
}
@Bean
public ConsumerFactory<Long, UserDto> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
}
Класс KafkaConsumerConfig очень похож на KafkaProducerConfig, который мы создавали ранее. Здесь так же присутствует Map, содержащий необходимые конфигурации, например, такие как десериализатор для ключа и значения. Созданная мапа используется при создании ConsumerFactory<>, которая в свою очередь, нужна для создания KafkaListenerContainerFactory<?>. Важная деталь: метод возвращающий KafkaListenerContainerFactory<?> должен называться kafkaListenerContainerFactory(), иначе Spring не сможет найти нужного бина и проект не скомпилируется. Запускаем.
Видим, что теперь ключ отображается как надо, а это значит, что всё работает. Конечно, возможности Apache Kafka далеко выходят за пределы тех, что описаны в данной статье, однако, надеюсь, прочитав её, вы составите представление о данном сервисе и, самое главное, сможете начать работу с ним.
Мойте руки чаще, носите маски, не выходите без необходимости на улицу, и будьте здоровы.
Kafka is a distributed event streaming platform that can be used for high-performance streaming analytics, asynchronous event processing and reliable applications. This article provides step-by-step guidance about installing Kafka on Windows 10 for test and learn purposes.
Install Git Bash
Download Git Bash from https://git-scm.com/downloads and then install it. We will use it to unzip Kafka binary package. If you have 7-zip of other unzip software, this is then not required.
Install Java JDK
Java JDK is required to run Kafka. If you have not installed Java JDK, please install it.
1) You can install JDK 8 from the following page:
https://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html
2) Setup environment variable
Let’s configure JAVA_HOME environment variable.
First, we need to find out the location of Java SDK. In my system, the path is: D:Javajdk1.8.0_161.
Your location can be different depends on where you install your JDK.
And then run the following command in PowerShell window:
SETX JAVA_HOME "D:Javajdk1.8.0_161"
Remember to quote the path if you have spaces in your JDK path.
3) Add Java bin folder into PATH system variable.
4) Verify java command
Once you complete the installation, please run the following command in PowerShell or Git Bash to verify:
$ java -version java version "1.8.0_161" Java(TM) SE Runtime Environment (build 1.8.0_161-b12) Java HotSpot(TM) 64-Bit Server VM (build 25.161-b12, mixed mode)
Download Kafka binary package
1) Go to Kafka download portal and select a version. For this tutorial, version Scala 2.13 — kafka_2.13-2.6.0.tgz is downloaded.
2) Unzip the binary package to a installation folder.
Now we need to unpack the downloaded package using GUI tool (like 7 Zip) or command line. I will use git bash to unpack it.
Open git bash and change the directory to the destination folder:
cd F:/big-data
And then run the following command to unzip:
tar -xvzf kafka_2.13-2.6.0.tgz
Most of the scripts that we need to run in the following steps are located in bin/windows folder as the screenshot shows:
3) Setup Kafka environment variable.
Let’s add a environment variable KAFKA_HOME so that we can easily reference it in the following steps.
Remember to change variable value accordingly based on your environment setup.
Start Kafka environment
1) Open Command Prompt and start ZooKeeper services by running this command:
%KAFKA_HOME%/bin/windows/zookeeper-server-start.bat %KAFKA_HOME%/config/zookeeper.properties
In this version, ZooKeeper is still required.
2) Start Kafka server
Open another Command Prompt window and run the following command:
%KAFKA_HOME%/bin/windows/kafka-server-start.bat %KAFKA_HOME%/config/server.properties
Once all the services are launched, you will have a Kafka environment ready to use.
You can verify by running jps commands (if you have Hadoop installed in your environment):
Let’s run some tests about Kafka environment.
Create a Kafka topic
Open another Command Prompt window and run the following command:
%KAFKA_HOME%/bin/windows/kafka-topics.bat --create --topic kontext-events --bootstrap-server localhost:9092
The command will create a topic named kontext-events as the above screenshot shows.
Describe Kafka topic
Run the following command to describe the created topic.
%KAFKA_HOME%/bin/windows/kafka-topics.bat --describe --topic kontext-events --bootstrap-server localhost:9092
The output looks like the following:
Topic: kontext-events PartitionCount: 1 ReplicationFactor: 1 Configs: segment.bytes=1073741824 Topic: kontext-events Partition: 0 Leader: 0 Replicas: 0 Isr: 0
Write some events into the topic
Let’s start to write some events into the topic by running the following command:
%KAFKA_HOME%/bin/windows/kafka-console-producer.bat --topic kontext-events --bootstrap-server localhost:9092
Each time represents an event. Let’s type into some events:
Press Ctrl + C to terminate this Console producer client.
Read the events in the topic
Let’s read the events by running the following command:
%KAFKA_HOME%/bin/windows/kafka-console-consumer.bat --topic kontext-events --from-beginning --bootstrap-server localhost:9092
The Console consumer client prints out the following events:
Press Ctrl + C to terminate the consumer client.
Shutdown Kafka services
After finish practices, you can turn off the series by running the following commands:
%KAFKA_HOME%/bin/windows/kafka-server-stop.bat %KAFKA_HOME%/config/server.properties %KAFKA_HOME%/bin/windows/zookeeper-server-stop.bat %KAFKA_HOME%/config/zookeeper.properties
The output looks like the following screenshot:
References
- Apache Kafka Quick Start
As you can see, it is very easy to configure and run Kafka on Windows 10. Stay tuned and more articles will be published about streaming analytics with Kafka in this Column.
Apache Kafka is an open-source application used for real-time streams for data in huge amount. Apache Kafka is a publish-subscribe messaging system. A messaging system lets you send messages between processes, applications, and servers. Broadly Speaking, Apache Kafka is software where topics can be defined and further processed.
Downloading and Installation
Apache Kafka can be downloaded from its official site kafka.apache.org
For the installation process, follow the steps given below:
Step 1: Go to the Downloads folder and select the downloaded Binary file.
Step 2: Extract the file and move the extracted folder to the directory where you wish to keep the files.
Step 3: Copy the path of the Kafka folder. Now go to config inside kafka folder and open zookeeper.properties file. Copy the path against the field dataDir and add /zookeeper-data to the path.
For example if the path is c:/kafka
Step 4: Now in the same folder config open server.properties and scroll down to log.dirs and paste the path. To the path add /kafka-logs
Step 5: This completes the configuration of zookeeper and kafka server. Now open command prompt and change the directory to the kafka folder. First start zookeeper using the command given below:
.binwindowszookeeper-server-start.bat .configzookeeper.properties
Step 6: Now open another command prompt and change the directory to the kafka folder. Run kafka server using the command:
.binwindowskafka-server-start.bat .configserver.properties
Now kafka is running and ready to stream data.
Apache Kafka is an open-source application used for real-time streams for data in huge amount. Apache Kafka is a publish-subscribe messaging system. A messaging system lets you send messages between processes, applications, and servers. Broadly Speaking, Apache Kafka is software where topics can be defined and further processed.
Downloading and Installation
Apache Kafka can be downloaded from its official site kafka.apache.org
For the installation process, follow the steps given below:
Step 1: Go to the Downloads folder and select the downloaded Binary file.
Step 2: Extract the file and move the extracted folder to the directory where you wish to keep the files.
Step 3: Copy the path of the Kafka folder. Now go to config inside kafka folder and open zookeeper.properties file. Copy the path against the field dataDir and add /zookeeper-data to the path.
For example if the path is c:/kafka
Step 4: Now in the same folder config open server.properties and scroll down to log.dirs and paste the path. To the path add /kafka-logs
Step 5: This completes the configuration of zookeeper and kafka server. Now open command prompt and change the directory to the kafka folder. First start zookeeper using the command given below:
.binwindowszookeeper-server-start.bat .configzookeeper.properties
Step 6: Now open another command prompt and change the directory to the kafka folder. Run kafka server using the command:
.binwindowskafka-server-start.bat .configserver.properties
Now kafka is running and ready to stream data.
Это вводная статья по использованию Apache Kafka со Spring Boot. О том, как установить и запустить Apache Kafka на Windows, создать топик и отправить туда первое событие из одного Spring Boot приложения. А получить его из другого Spring Boot приложения.
Kafka запускается в ZooKeeper, но отдельно скачивать ZooKeeper не нужно, он идет вместе с Kafka.
Как запустить Kafka под Windows 10
- Сначала нужно загрузить и распаковать tgz-архив.
- Нужно переименовать папку вида kafka_2.13-2.7.0 — дать более короткое имя, например kafka, иначе возникнут проблемы при запуске из консоли
- Открыть Windows PowerShell (либо командную строку cmd.exe, неважно)
- Перейти в переименованную папку kafka (внутри будут папки bin, config и другие папки и файлы)
- Запустить ZooKeeper, набрав в консоли:
.binwindowszookeeper-server-start.bat .configzookeeper.properties
- Открыть вторую консоль, перейти в ту же переименованную папку kafka и запустить Kafka:
.binwindowskafka-server-start.bat .configserver.properties
Все, Apache Kafka запущен.
Теперь в нем можно создавать топики — как из консоли, так и из Spring Boot. Мы сделаем это из Spring Boot.
У нас будет два Spring Boot приложения. Они совершенно независимы друг от друга. Одно приложение будет отправлять событие в топик (это приложение Producer), а второе — получать их из топика (приложение Consumer).
Mаven-зависимость
Чтобы работать с Kafka, оба приложения (как Producer, так и Consumer) должны иметь зависимость:
<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency>
Spring Boot Producer
Выглядит приложение просто:
@SpringBootApplication public class ProducerApplication { public static void main(String[] args) { SpringApplication.run(ProducerApplication.class, args); } @Bean public NewTopic topic() { return TopicBuilder.name("topic1") .partitions(10) .replicas(1) .build(); } @Bean public ApplicationRunner runner(KafkaTemplate<String, String> template) { return args -> { template.send("topic1", "test"); }; } }
Бин topic типа NewTopic создает топик topic1, если его нет.
KafkaTemplate отправляет события в топик topic1. У нас это всего одна строка — test.
Перейдем к созданию второго приложения, которое получает события из Kafka.
Spring Boot Consumer
Это приложение слушает топик topic1:
@SpringBootApplication public class Application { public static void main(String[] args) { SpringApplication.run(Application.class, args); } @Bean public NewTopic topic() { return TopicBuilder.name("topic1") .partitions(10) .replicas(1) .build(); } @KafkaListener(id = "myId", topics = "topic1") public void listen(String in) { System.out.println(in); } }
Если в топик приходит событие, то оно выводится в консоль.
То есть в консоль будет выведена одна строка:
test
Приложение Spring Boot Producer завершилось быстро, а Spring Boot Consumer висит и слушает топик. Можно запустить Spring Boot Producer еще раз, тогда в консоли Spring Boot Consumer снова будет выведена строка:
test
Итоги
Код примеров доступен на GitHub.
Источники:
- Spring for Apache Kafka
- https://kafka.apache.org/intro
Вступление
В моей последней статье мы рассмотрели настройку и использование Hadoop. Эта статья посвящена настройке и запуску сервера Apache Kafka в ОС Windows. Это руководство также содержит инструкции по настройке Java и Apache ZooKeeper.
Apache Kafka — это быстрая и масштабируемая очередь обмена сообщениями, способная обрабатывать большие нагрузки в контексте чтения и записи, то есть вещей, связанных с IO. Вы можете найти больше информации о Кафке на http://kafka.apache.org/ . Для Apache Kafka требуется работающий экземпляр ZooKeeper, который используется для надежной распределенной координации. Пожалуйста, узнайте больше о ZooKeeper на https://zookeeper.apache.org/ .
Из этого видео вы можете получить помощь по настройке Kafka в Windows.
Авторский GitHub:
Я создал несколько утилит Spark-Scala по адресу https://github.com/gopal-tiwari , которые могут быть полезны в некоторых других случаях.
Скачивание необходимых файлов
- Загрузите серверную JRE в соответствии с вашей ОС и архитектурой процессора с http://www.oracle.com/technetwork/java/javase/downloads/jre8-downloads-2133155.html
- Загрузите и установите 7-zip с http://www.7-zip.org/download.html
- Загрузите и распакуйте ZooKeeper, используя 7-zip с http://zookeeper.apache.org/releases.html
- Загрузите и распакуйте Kafka, используя 7-zip с http://kafka.apache.org/downloads.html
В этом руководстве мы предполагаем, что ZooKeeper и Kafka разархивированы на диске C :, но вы можете разархивировать их в любом месте.
Здесь мы используем полноценный ZooKeeper, а не тот, который поставляется вместе с Kafka, потому что это будет одноузловой экземпляр ZooKeeper. Если вы хотите, вы можете запустить Kafka с упакованным ZooKeeper, расположенным в пакете Kafka в каталоге kafka bin windows .
Монтаж
A. Настройка JDK
1. Запустите установку JRE и установите флажок «Изменить папку назначения», затем нажмите «Установить».
2. Измените каталог установки на любой путь без пробелов в имени папки. Например, C: Java jre1.8.0_xx . (По умолчанию это будет C: Program Files Java jre1.8.0_xx), затем нажмите «Далее».
3. Теперь откройте диалоговое окно системных переменных среды, открыв Панель управления -> Система -> Расширенные настройки системы -> Переменные среды.
4. Нажмите кнопку «Новая переменная пользователя» в разделе «Пользовательские переменные», затем введите «JAVA_HOME» в поле «Имя переменной» и укажите свой путь jre в значении переменной. Это должно выглядеть как на картинке ниже:
(Путь и версия Java могут меняться в зависимости от используемой вами версии Kafka)
5. Теперь нажмите ОК.
6. Найдите переменную Path в разделе «Системная переменная» в диалоговом окне «Переменные среды», которое вы только что открыли.
7. Отредактируйте путь и введите «;% JAVA_HOME% bin» в конце уже написанного текста, как показано на рисунке ниже:
8. Чтобы подтвердить установку Java, просто откройте cmd и введите « java –version». Вы должны увидеть версию Java, которую вы только что установили.
Если ваша командная строка выглядит примерно так, как на картинке выше, вы готовы. В противном случае вам необходимо перепроверить, соответствует ли ваша установочная версия правильной архитектуре ОС (x86, x64) или путь переменных среды является правильным.
B. Установка ZooKeeper
1. Перейдите в каталог конфигурации ZooKeeper. Для меня это C: zookeeper-3.4.7 conf
2. Переименуйте файл «zoo_sample.cfg» в « zoo.cfg».
3. Откройте zoo.cfg в любом текстовом редакторе, например «Блокнот»; Я предпочитаю Notepad ++.
4. Найти и редактировать dataDir=/tmp/zookeeper to :zookeeper-3.4.7data
5. Добавьте запись в переменные системной среды, как мы это делали для Java.
а. Добавьте ZOOKEEPER_HOME = C:zookeeper-3.4.7
к системным переменным.
б. Отредактируйте системную переменную с именем «Path» и добавьте ;%ZOOKEEPER_HOME%bin;
6. Вы можете изменить порт Zookeeper по умолчанию в файле zoo.cfg (Порт по умолчанию 2181).
7. Запустите ZooKeeper, открыв новый cmd и набрав zkserver
.
8. Вы увидите командную строку с некоторыми подробностями, как на картинке ниже:
Поздравляем, ваш ZooKeeper запущен и работает на порту 2181!
C. Настройка Кафки
1. Перейдите в каталог настроек Kafka. Для меня это C: kafka_2.11-0.9.0.0 config
2. Отредактируйте файл «server.properties».
3. Найдите и отредактируйте строку . log.dirs=/tmp/kafka-logs” to “log.dir= C:kafka_2.11-0.9.0.0kafka-logs
4. Если ваш ZooKeeper работает на другом компьютере или кластере, вы можете отредактировать «zookeeper. подключите : 2181 ” к вашему собственному IP и порту. Для этой демонстрации мы используем одну и ту же машину, поэтому нет необходимости менять. Также порт Kafka и broker.id настраиваются в этом файле. Оставьте другие настройки как есть.
5. Ваш Kafka будет работать по умолчанию через порт 9092 и подключаться к порту по умолчанию ZooKeeper, 2181.
D. Запуск сервера Kafka
Важно: Пожалуйста, убедитесь, что ваш экземпляр ZooKeeper запущен и работает перед запуском сервера Kafka.
1. Перейдите в каталог установки Kafka: C: kafka_2.11-0.9.0.0
2. Откройте здесь командную строку, нажав Shift + щелчок правой кнопкой мыши и выберите «Открыть окно командной строки здесь»).
3. Теперь введите .binwindowskafka-server-start.bat .configserver.properties
и нажмите Enter.
.binwindowskafka-server-start.bat .configserver.properties
4. Если все прошло нормально, ваша командная строка будет выглядеть так:
5. Теперь ваш Kafka Server запущен и вы можете создавать темы для хранения сообщений. Кроме того, мы можем создавать или потреблять данные из кода Java или Scala или непосредственно из командной строки.
E. Создание тем
1. Теперь создайте тему с именем «test» и коэффициентом репликации 1, поскольку у нас работает только один сервер Kafka. Если у вас есть кластер с несколькими запущенными серверами Kafka, вы можете соответственно увеличить коэффициент репликации, что повысит доступность данных и будет действовать как отказоустойчивая система.
2. Откройте новую командную строку в расположении C: kafka_2.11-0.9.0.0 bin windows.
3. Введите следующую команду и нажмите Enter:
kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
F. Создание производителя и потребителя для тестирования сервера
1. Откройте новую командную строку в расположении C: kafka_2.11-0.9.0.0 bin windows
2. Чтобы запустить продюсера, введите следующую команду:
kafka-console-producer.bat --broker-list localhost:9092 --topic test
3. Снова откройте новую командную строку в том же месте, что и C: kafka_2.11-0.9.0.0 bin windows
4. Теперь запустите потребителя, введя следующую команду:
До кафки версии 2.0 (<2.0):
kafka-console-consumer.bat --zookeeper localhost:2181 --topic test
После кафки версии 2.0 (> = 2.0):
kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test
5. Теперь у вас будет две командные строки, как на картинке ниже:
6. Теперь введите что-нибудь в командной строке производителя и нажмите Enter, и вы сможете увидеть сообщение в командной строке другого потребителя.
7. Если вы можете просмотреть и просмотреть свои сообщения на стороне потребителя, вы завершили настройку Kafka.
Некоторые другие полезные команды
- Список тем:
kafka-topics.bat --list --zookeeper localhost:2181
- Опишите тему:
kafka-topics.bat --describe --zookeeper localhost:2181 --topic [Topic Name]
- Читать сообщения с самого начала
- До версии <2.0:
kafka-console-consumer.bat --zookeeper localhost:2181 --topic [Topic Name] --from-beginning
- После версии> 2.0:
kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic [Topic Name] --from-beginn
- До версии <2.0:
- Удалить тему:
kafka-run-class.bat kafka.admin.TopicCommand --delete --topic [topic_to_delete] --zookeeper localhost:2181
Learn to install Apache Kafka on Windows 10 and executing start server and stop server scripts related to Kafka and Zookeeper. We will also verify the Kafka installation by creating a topic, producing few messages to it and then use a consumer to read the messages written in Kafka.
1. Prerequisites
- Java 8 is required to run latest downloads from Kafka site.
- Zookeeper (to store metadata about the Kafka cluster) is also mandatory. Kafka comes with inbuilt Zookeeper to start with. But it is recommended to install zookeeper separately in production environment. Download it from its official site.
- Kafka can be run on any operating system. The linux is recommended OS. With Windows, Kafka have some knows bugs.
This tutorial is for beginners and does not use separate zookeeper instance – to keep things simple and focused towards Kafka only.
2. Download and install Kafka
- Download Kafka from official site. I download the latest version on today which is 2.5.0 and file name is “kafka_2.12-2.5.0.tgz“.
- Copy the downloaded file to some folder and extract it using
tar
command.> tar -xzf kafka_2.12-2.5.0.tgz
- Copy the extracted folder in desired location. I have put it on location “
E:devsetupbigdatakafka2.5
“.Installation is pretty much done !!
3. Startup and Shutdown
To start Kafka, we need to first start Zookeeper and then Kafka. I am writing small batch files which move to Kafka installation directory first and then execute the command in new command prompt window.
3.1. Start Zookeeper
To start zookeeper, we need to run zookeeper-server-start.bat
script and pass zookeeper configuration file path.
cd E:devsetupbigdatakafka2.5 start cmd /k binwindowszookeeper-server-start.bat configzookeeper.properties
3.2. Start Kafka
To start Kafka, we need to run kafka-server-start.bat
script and pass broker configuration file path.
cd E:devsetupbigdatakafka2.5 start cmd /k binwindowskafka-server-start.bat configserver.properties
3.3. Shutdown Kafka
To stop Kafka, we need to run kafka-server-stop.bat
script.
cd E:devsetupbigdatakafka2.5 start cmd /k binwindowskafka-server-stop.bat
3.4. Shutdown Zookeeper
To stop Zookeeper, we need to run zookeeper-server-stop.bat
script.
cd E:devsetupbigdatakafka2.5 start cmd /k binwindowszookeeper-server-stop.bat
Do not stop zookeeper and kafka using
CTRL+C
commands. Always use above.bat
files or commands. Otherwise the data corruption may occur.
4. Verify Kafka Installation
First, start Zookeeper and Kafka using above scripts.
- Open a new command prompt, and create new Kafka topic.
> binwindowskafka-topics.bat --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test //Output: Created topic test.
- Now list all the topics to verify the created topic is present in this list. At this step, we have only one topic.
> binwindowskafka-topics.bat --list --bootstrap-server localhost:9092 //Output: test
- Now list all the topics to verify the created topic is present in this list. At this step, we have only one topic.
> binwindowskafka-topics.bat --list --bootstrap-server localhost:9092 //Output: test
- Produce some messages and submit to
test
topic. I added two messages i.e. “Hello” and “Kafka”.> binwindowskafka-console-producer.bat --bootstrap-server localhost:9092 --topic test
- Consume the messages and submit to
test
topic.> binwindowskafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test --from-beginning //Output: Hello Kafka
5. Conclusion
In this tutorial, we learned to install Kafka along with Zookeeper. We learned to start and stop both servers. Additionally, we verified the installation by creating a topic, posting some messages and then consuming using console consumer script.
Important thing to note is that we should never stop the servers by kill process or CTRL+C
commands. Always use scripts to stop the servers.
Happy Learning !!
Reference: