我希望有一个Kafka Consumer,它从一个主题中的最新消息开始 .
这是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的值是 latest ,但是消费者会在2天前启动表单消息,然后赶上最新消息 .
我错过了什么?
2 回答
你之前使用相同的
group.id
运行相同的代码吗?仅当您的消费者尚未存储现有偏移量时,才使用auto.offset.reset
参数 . 因此,如果您之前运行过该示例,比如说两天前,然后再次运行它,它将从上次消耗的位置开始 .如果您想手动转到主题的末尾,请使用
seekToEnd()
.有关此问题的详细讨论,请参阅https://stackoverflow.com/a/32392174/1392894 .
如果要手动控制偏移的位置,则需要设置enable.auto.commit = false .
如果要将所有偏移定位到每个分区的末尾,则调用seekToEnd()
https://kafka.apache.org/0102/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#seekToEnd(java.util.Collection)