Apache Kafka - пример простого производителя

Давайте создадим приложение для публикации и использования сообщений с помощью Java-клиента. Клиент производителя Kafka состоит из следующих API.

API KafkaProducer

Давайте разберемся с наиболее важным набором API производителя Kafka в этом разделе. Центральная часть API KafkaProducer - это класс KafkaProducer . Класс KafkaProducer предоставляет возможность подключить брокера Kafka в его конструкторе с помощью следующих методов.

• Класс KafkaProducer предоставляет метод send для асинхронной отправки сообщений в тему. Подпись send () выглядит следующим образом

producer.send(new ProducerRecord<byte[],byte[]>(topic, 
partition, key1, value1) , callback);

• ProducerRecord - Производитель управляет буфером записей, ожидающих отправки.

• Callback - Предоставляемый пользователем обратный вызов для выполнения, когда запись была подтверждена сервером (ноль означает отсутствие обратного вызова).

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

public void flush()

• Класс KafkaProducer предоставляет метод partitionFor, который помогает получить метаданные раздела для заданной темы. Это можно использовать для пользовательского разбиения. Сигнатура этого метода следующая -

public Map metrics()

Он возвращает карту внутренних показателей, поддерживаемую производителем.

• public void close () - класс KafkaProducer предоставляет блоки метода close до тех пор, пока не будут выполнены все ранее отправленные запросы.

API производителя

Центральной частью Producer API является класс Producer . Класс Producer предоставляет возможность подключить брокера Kafka в своем конструкторе следующими методами.

Продюсерский класс

Класс производителя предоставляет метод отправки для send сообщения в одну или несколько тем, используя следующие подписи.

public void send(KeyedMessaget<k,v> message) 
- sends the data to a single topic,par-titioned by key using either sync or async producer.
public void send(List<KeyedMessage<k,v>>messages)
- sends data to multiple topics.
Properties prop = new Properties();
prop.put(producer.type,”async”)
ProducerConfig config = new ProducerConfig(prop);

Есть два типа производителей - Sync а также Async.

Та же конфигурация API применима и к производителю Sync . Разница между ними заключается в том, что производитель синхронизации отправляет сообщения напрямую, но отправляет сообщения в фоновом режиме. Если вам нужна более высокая пропускная способность, предпочтительнее использовать асинхронный производитель. В предыдущих выпусках, таких как 0.8, производитель async не имел обратного вызова для send () для регистрации обработчиков ошибок. Это доступно только в текущей версии 0.9.

public void close ()

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

Настройки конфигурации

Основные настройки конфигурации Producer API перечислены в следующей таблице для лучшего понимания -

S.No	Параметры конфигурации и описание
1	client.id идентифицирует приложение производителя
2	producer.type либо синхронизировать, либо асинхронно
3	acks Конфигурация acks контролирует критерии, по которым запросы производителя считаются завершенными.
4	retries Если запрос производителя не удается, автоматически повторите попытку с определенным значением.
5	bootstrap.servers начальный список брокеров.
6	linger.ms если вы хотите уменьшить количество запросов, вы можете установить linger.ms на что-то большее, чем некоторое значение.
7	key.serializer Ключ для интерфейса сериализатора.
8	value.serializer значение для интерфейса сериализатора.
9	batch.size Размер буфера.
10	buffer.memory контролирует общий объем памяти, доступной производителю для буферизации.

ProducerRecord API

ProducerRecord - это пара ключ / значение, которая отправляется в кластер Kafka. Конструктор класса ProductRecord для создания записи с парами разделов, ключей и значений с использованием следующей подписи.

public ProducerRecord (string topic, int partition, k key, v value)

• Topic - определяемое пользователем название темы, которое будет добавлено к записи.

• Partition - количество разделов

• Key - Ключ, который будет включен в запись.

• Value - Запись содержимого

public ProducerRecord (string topic, k key, v value)

Конструктор класса ProducerRecord используется для создания записи с парами ключ, значение и без разделения.

• Topic - Создайте тему для присвоения записи.

• Key - ключ для записи.

• Value - записывать содержимое.

public ProducerRecord (string topic, v value)

Класс ProducerRecord создает запись без раздела и ключа.

• Topic - создайте тему.

• Value - записывать содержимое.

Методы класса ProducerRecord перечислены в следующей таблице -

S.No	Методы и описание класса
1	public string topic() Тема будет добавлена ​​к записи.
2	public K key() Ключ, который будет включен в запись. Если такого ключа нет, здесь будет повторно установлено значение null.
3	public V value() Запишите содержимое.
4	partition() Количество разделов для записи

Приложение SimpleProducer

Перед созданием приложения сначала запустите ZooKeeper и брокер Kafka, затем создайте свою собственную тему в брокере Kafka, используя команду create topic. После этого создайте класс java с именем Sim-pleProducer.java и введите следующий код .

//import util.properties packages
import java.util.Properties;

//import simple producer packages
import org.apache.kafka.clients.producer.Producer;

//import KafkaProducer packages
import org.apache.kafka.clients.producer.KafkaProducer;

//import ProducerRecord packages
import org.apache.kafka.clients.producer.ProducerRecord;

//Create java class named “SimpleProducer”
public class SimpleProducer {
   
   public static void main(String[] args) throws Exception{
      
      // Check arguments length value
      if(args.length == 0){
         System.out.println("Enter topic name”);
         return;
      }
      
      //Assign topicName to string variable
      String topicName = args[0].toString();
      
      // create instance for properties to access producer configs   
      Properties props = new Properties();
      
      //Assign localhost id
      props.put("bootstrap.servers", “localhost:9092");
      
      //Set acknowledgements for producer requests.      
      props.put("acks", “all");
      
      //If the request fails, the producer can automatically retry,
      props.put("retries", 0);
      
      //Specify buffer size in config
      props.put("batch.size", 16384);
      
      //Reduce the no of requests less than 0   
      props.put("linger.ms", 1);
      
      //The buffer.memory controls the total amount of memory available to the producer for buffering.   
      props.put("buffer.memory", 33554432);
      
      props.put("key.serializer", 
         "org.apache.kafka.common.serializa-tion.StringSerializer");
         
      props.put("value.serializer", 
         "org.apache.kafka.common.serializa-tion.StringSerializer");
      
      Producer<String, String> producer = new KafkaProducer
         <String, String>(props);
            
      for(int i = 0; i < 10; i++)
         producer.send(new ProducerRecord<String, String>(topicName, 
            Integer.toString(i), Integer.toString(i)));
               System.out.println(“Message sent successfully”);
               producer.close();
   }
}

Compilation - Приложение можно скомпилировать с помощью следующей команды.

javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*” *.java

Execution - Приложение можно запустить с помощью следующей команды.

java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*”:. SimpleProducer <topic-name>

Output

Message sent successfully
To check the above output open new terminal and type Consumer CLI command to receive messages.
>> bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic <topic-name> —from-beginning
1
2
3
4
5
6
7
8
9
10

Простой потребительский пример

На данный момент мы создали производителя для отправки сообщений в кластер Kafka. Теперь давайте создадим потребителя для приема сообщений из кластера Kafka. KafkaConsumer API используется для получения сообщений из кластера Kafka. Конструктор класса KafkaConsumer определен ниже.

public KafkaConsumer(java.util.Map<java.lang.String,java.lang.Object> configs)

configs - Вернуть карту потребительских конфигов.

Класс KafkaConsumer имеет следующие важные методы, которые перечислены в таблице ниже.

S.No	Метод и описание
1	public java.util.Set<TopicPar-tition> assignment() Получите набор разделов, назначенных в данный момент потребителем.
2	public string subscription() Подпишитесь на указанный список тем, чтобы получать разделы с динамической подписью.
3	public void sub-scribe(java.util.List<java.lang.String> topics, ConsumerRe-balanceListener listener) Подпишитесь на указанный список тем, чтобы получать разделы с динамической подписью.
4	public void unsubscribe() Отписаться от тем из данного списка разделов.
5	public void sub-scribe(java.util.List<java.lang.String> topics) Подпишитесь на указанный список тем, чтобы получать разделы с динамической подписью. Если данный список тем пуст, он обрабатывается так же, как unsubscribe ().
6	public void sub-scribe(java.util.regex.Pattern pattern, ConsumerRebalanceLis-tener listener) Шаблон аргумента относится к шаблону подписки в формате регулярного выражения, а аргумент слушателя получает уведомления от шаблона подписки.
7	public void as-sign(java.util.List<TopicParti-tion> partitions) Вручную назначьте заказчику список разделов.
8	poll() Получение данных для указанных тем или разделов с помощью одного из API подписки / назначения. Это вернет ошибку, если темы не подписаны до опроса данных.
9	public void commitSync() Смещения фиксации, возвращенные в последнем опросе () для всего подписанного списка тем и разделов. Та же операция применяется к commitAsyn ().
10	public void seek(TopicPartition partition, long offset) Получите текущее значение смещения, которое потребитель будет использовать в следующем методе poll ().
11	public void resume() Возобновить приостановленные разделы.
12	public void wakeup() Разбуди потребителя.

ConsumerRecord API

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

public ConsumerRecord(string topic,int partition, long offset,K key, V value)

• Topic - Название темы для потребительской записи, полученной из кластера Kafka.

• Partition - Раздел по теме.

• Key - Ключ записи, если ключ не существует, возвращается null.

• Value - Запись содержимого.

ConsumerRecords API

ConsumerRecords API действует как контейнер для ConsumerRecord. Этот API используется для хранения списка ConsumerRecord для каждого раздела по определенной теме. Его конструктор определен ниже.

public ConsumerRecords(java.util.Map<TopicPartition,java.util.List
<Consumer-Record>K,V>>> records)

• TopicPartition - Вернуть карту раздела по определенной теме.

• Records - Вернуть список ConsumerRecord.

В классе ConsumerRecords определены следующие методы.

S.No	Методы и описание
1	public int count() Количество записей по всем темам.
2	public Set partitions() Набор разделов с данными в этом наборе записей (если данные не были возвращены, набор пуст).
3	public Iterator iterator() Итератор позволяет циклически перебирать, получать или перемещать элементы.
4	public List records() Получить список записей для данного раздела.

Настройки конфигурации

Параметры конфигурации для основных параметров конфигурации Consumer client API перечислены ниже:

S.No	Настройки и описание
1	bootstrap.servers Загрузочный список брокеров.
2	group.id Включает отдельного потребителя в группу.
3	enable.auto.commit Включите автоматическую фиксацию для смещений, если значение истинно, в противном случае не зафиксировано.
4	auto.commit.interval.ms Возвращает, как часто обновленные потребляемые смещения записываются в ZooKeeper.
5	session.timeout.ms Указывает, сколько миллисекунд Kafka будет ждать, пока ZooKeeper ответит на запрос (чтение или запись), прежде чем отказаться и продолжить прием сообщений.

Приложение SimpleConsumer

Шаги приложения производителя здесь остаются прежними. Сначала запустите своего брокера ZooKeeper и Kafka. Затем создайте приложение SimpleConsumer с классом java с именем SimpleCon-sumer.java и введите следующий код.

import java.util.Properties;
import java.util.Arrays;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;

public class SimpleConsumer {
   public static void main(String[] args) throws Exception {
      if(args.length == 0){
         System.out.println("Enter topic name");
         return;
      }
      //Kafka consumer configuration settings
      String topicName = args[0].toString();
      Properties props = new Properties();
      
      props.put("bootstrap.servers", "localhost:9092");
      props.put("group.id", "test");
      props.put("enable.auto.commit", "true");
      props.put("auto.commit.interval.ms", "1000");
      props.put("session.timeout.ms", "30000");
      props.put("key.deserializer", 
         "org.apache.kafka.common.serializa-tion.StringDeserializer");
      props.put("value.deserializer", 
         "org.apache.kafka.common.serializa-tion.StringDeserializer");
      KafkaConsumer<String, String> consumer = new KafkaConsumer
         <String, String>(props);
      
      //Kafka Consumer subscribes list of topics here.
      consumer.subscribe(Arrays.asList(topicName))
      
      //print the topic name
      System.out.println("Subscribed to topic " + topicName);
      int i = 0;
      
      while (true) {
         ConsumerRecords<String, String> records = con-sumer.poll(100);
         for (ConsumerRecord<String, String> record : records)
         
         // print the offset,key and value for the consumer records.
         System.out.printf("offset = %d, key = %s, value = %s\n", 
            record.offset(), record.key(), record.value());
      }
   }
}

Compilation - Приложение можно скомпилировать с помощью следующей команды.

javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*” *.java

Execution − Приложение можно запустить с помощью следующей команды

java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*”:. SimpleConsumer <topic-name>

Input- Откройте интерфейс командной строки производителя и отправьте несколько сообщений в тему. Вы можете ввести небольшой ввод как «Hello Consumer».

Output - Ниже будет вывод.

Subscribed to topic Hello-Kafka
offset = 3, key = null, value = Hello Consumer