我想要一个从主题中的最新消息开始的 Kafka 消费者。
这是Java代码:
private static Properties properties = new Properties();
private static KafkaConsumer<String, String> consumer;
static
{
properties.setProperty("bootstrap.servers","localhost");
properties.setProperty("enable.auto.commit", "true");
properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.setProperty("group.id", "test");
properties.setProperty("auto.offset.reset", "latest");
consumer = new KafkaConsumer<>(properties);
consumer.subscribe(Collections.singletonList("mytopic"));
}
@Override
public StreamHandler call() throws Exception
{
while (true)
{
ConsumerRecords<String, String> consumerRecords = consumer.poll(200);
Iterable<ConsumerRecord<String, String>> records = consumerRecords.records("mytopic");
for(ConsumerRecord<String, String> rec : records)
{
System.out.println(rec.value());
}
}
}
虽然 auto.offset.reset 的值是 最新 ,但消费者开始形成属于 2 天前的消息,然后它会 catch 最新的消息。
我错过了什么?
最佳答案
您之前是否使用相同的 group.id
运行过相同的代码? ? auto.offset.reset
参数仅在没有为您的消费者存储的现有偏移量时才使用。所以如果你之前运行过这个例子,比如说两天前,然后你再次运行它,它将从最后一个消耗的位置开始。
使用 seekToEnd()
如果您想手动转到主题的末尾。
见 https://stackoverflow.com/a/32392174/1392894对此进行更彻底的讨论。
关于apache-kafka - Kafka 消费者不是从最新消息开始,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/43945927/