Apache Kafka - пример группы потребителей

Группа потребителей - это многопоточное или многопоточное потребление из тем Kafka.

Потребительская группа

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

Ребалансировка потребителя

Добавление дополнительных процессов / потоков приведет к повторной балансировке Kafka. Если какой-либо потребитель или брокер не может отправить контрольный сигнал в ZooKeeper, его можно перенастроить через кластер Kafka. Во время этой перебалансировки Kafka назначит доступные разделы доступным потокам, возможно, переместив раздел в другой процесс.

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 ConsumerGroup {
   public static void main(String[] args) throws Exception {
      if(args.length < 2){
         System.out.println("Usage: consumer <topic> <groupname>");
         return;
      }
      
      String topic = args[0].toString();
      String group = args[1].toString();
      Properties props = new Properties();
      props.put("bootstrap.servers", "localhost:9092");
      props.put("group.id", group);
      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.serialization.ByteArraySerializer");
      props.put("value.deserializer", 
         "org.apache.kafka.common.serializa-tion.StringDeserializer");
      KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
      
      consumer.subscribe(Arrays.asList(topic));
      System.out.println("Subscribed to topic " + topic);
      int i = 0;
         
      while (true) {
         ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records)
               System.out.printf("offset = %d, key = %s, value = %s\n", 
               record.offset(), record.key(), record.value());
      }     
   }  
}

Компиляция

javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*" ConsumerGroup.java

Исполнение

>>java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*":. 
ConsumerGroup <topic-name> my-group
>>java -cp "/home/bala/Workspace/kafka/kafka_2.11-0.9.0.0/libs/*":. 
ConsumerGroup <topic-name> my-group

Здесь мы создали образец названия группы как my-group с двумя потребителями. Точно так же вы можете создать свою группу и количество потребителей в группе.

Ввод

Откройте CLI производителя и отправьте несколько сообщений вроде -

Test consumer group 01
Test consumer group 02

Выход первого процесса

Subscribed to topic Hello-kafka
offset = 3, key = null, value = Test consumer group 01

Выход второго процесса

Subscribed to topic Hello-kafka
offset = 3, key = null, value = Test consumer group 02

Надеюсь, вы бы поняли SimpleConsumer и ConsumeGroup, используя демонстрацию клиента Java. Теперь у вас есть представление о том, как отправлять и получать сообщения с помощью клиента Java. Давайте продолжим интеграцию Kafka с технологиями больших данных в следующей главе.