无法获取Kafka主题中的邮件数

Question

我对Kafka还很陌生。我已经用Java创建了一个示例生产者和消费者。使用producer,我可以将数据发送到kafka主题,但我无法使用以下消费者代码获取主题中的记录数。

    public class ConsumerTests {
        public static void main(String[] args) throws Exception {
            BasicConfigurator.configure();

            String topicName = "MobileData";
            String groupId = "TestGroup";
            Properties properties = new Properties();
            properties.put("bootstrap.servers", "localhost:9092");
            properties.put("group.id", groupId);
            properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

            KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
            kafkaConsumer.subscribe(Arrays.asList(topicName));


try {
  while (true) {
    ConsumerRecords<String, String> consumerRecords = consumer.poll(100);
    System.out.println("Record count is " + records.count());
  }
} catch (WakeupException e) {
  // ignore for shutdown
} finally {
  consumer.close();
}

}
}

Answer 1

轮询(...)调用通常应处于循环中。在分区分配过程中，初始轮询(...)始终可能不返回数据（取决于超时时间）。这里有一个例子：

try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    System.out.println("Record count is " + records.count());
  }
} catch (WakeupException e) {
  // ignore for shutdown
} finally {
  consumer.close();
}

欲知更多信息，请参阅相关文章：