Понимание идентификатора группы потребителей

Я сделал новую установку Apache Kafka 0.10.1.0.

Мне удалось отправить/получить сообщения в командной строке.

При использовании примера "Продюсер/потребитель Java" я не могу узнать параметр group.id в примере пользователя.

Сообщите мне, как исправить эту проблему.

Ниже приведен пример потребителя, который я использовал:

public static void main(String[] args) {
             Properties props = new Properties();
             props.put("bootstrap.servers", "localhost:9092");
             props.put("group.id", "my-topic");
             props.put("enable.auto.commit", "true");
             props.put("auto.commit.interval.ms", "1000");
             props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
             props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
             KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
             try {
                 consumer.subscribe(Arrays.asList("my-topic"));

                     ConsumerRecords<String, String> records = consumer.poll(100);
                     System.err.println("records size=>"+records.count());
                     for (ConsumerRecord<String, String> record : records) 
                         System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());



              }
             catch (Exception ex){
                 ex.printStackTrace();
             }
            finally {
                 consumer.close();
            }
        }

После запуска команды для пользователя я могу видеть сообщения (на консоли), отправленные производителем. Но не удалось увидеть сообщения из java-программы

bin\windows\kafka-console-consumer.bat --bootstrap-server localhost: 9092 --topic my-topic --from-begin

Ответ 1

Потребители обозначают себя именем группы потребителей, и каждая запись опубликованный в теме, доставляется одному экземпляру пользователя в каждом подписной группы потребителей. Потребительские экземпляры могут быть раздельными процессов или на отдельных машинах.

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

Если все экземпляры потребителя имеют разные группы потребителей, то каждая запись будет передаваться всем потребительским процессам.

Группа .id - это строка, которая однозначно идентифицирует группу потребительских процессов, к которой принадлежит этот потребитель.

(Вступление Кафки)

Ответ 2

Присвойте произвольное значение идентификатору группы. Это не важно

props.put("group.id", "Any Random Value");

Ответ 3

В коде, который вы предоставили, вы просто ждете данных один раз за 100 мс. Вы должны получать данные в цикле или ждать более длительного периода времени (в этом случае вы получите только одну часть данных). Что касается "group.id", это случай, когда вы запускаете пользователя из консоли, он получает случайный "group.id".

Ответ 4

Поскольку никакого смещения не было предоставлено, клиент java будет ждать новых сообщений, но не будет показывать существующие сообщения - это так, как ожидалось. Если вы хотите прочитать все сообщения уже в теме, вы можете использовать этот фрагмент кода:

if (READ_FROM_BEGINNING) {
    //consume all the messages from the topic from the beginning.
    //this doesn't work reliably if it consumer.poll(..) is not called first 
    //probably because of lazy-loading issues            
    consumer.poll(10);
    consumer.seekToBeginning(consumer.assignment()); //if intending to 
    //read from the beginning or call below to read from a predefined offset.
    //consumer.seek(consumer.assignment().iterator().next(), READ_FROM_OFFSET);
}

Ответ 5

Ниже приведены некоторые результаты тестирования для раздела и потребительского свойства group.id

 Properties props = new Properties();
  //set all other properties as required
  props.put("group.id", "ConsumerGroup1");
  props.put("max.poll.records", "1");
  KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

user.group id - загрузить баланс полученных данных (если group.id отличается для каждого потребителя, каждый потребитель получит копию данных)

если раздел = 1 и общее количество потребителей = 2, только один из двух активных потребителей получит данные

если раздел = 2 и общее количество потребителей = 2, каждый из двух активных потребителей равномерно получает данные

если раздел = 3 и общее количество потребителей = 2, каждый из двух активных потребителей получит данные. один потребитель получает данные из 2 разделов, а другой получает данные из 1 раздела.

если раздел = 3 и общее количество потребителей = 3, каждый из трех активных потребителей равномерно получает данные.

Ответ 6

Группа потребителей - это группа потребителей, которая должна быть определена в файле Kafka consumer.properties.

Добавьте группу my-topic в группу потребителей, и она должна работать следующим образом:

# consumer group id
group.id=my-topic-consumer-group