Учебники

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

Ноябрь 6, 2018

Потребительская группа — это многопоточное или многопользовательское потребление по темам Kafka.

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

  • Потребители могут присоединиться к группе, используя тот же group.id.

  • Максимальный параллелизм группы заключается в том, что количество потребителей в группе ← нет разделов.

  • Kafka назначает разделы темы потребителю в группе, так что каждый раздел потребляется ровно одним потребителем в группе.

  • Кафка гарантирует, что сообщение будет прочитано только одним потребителем в группе.

  • Потребители могут видеть сообщение в том порядке, в котором они были сохранены в журнале.

Потребители могут присоединиться к группе, используя тот же group.id.

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

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 = con-sumer.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 с технологиями больших данных в следующей главе.

71
Share: