我在 java 中有一个函数，我试图在其中获取未读的消息。例如，如果我在代理中有 offSet 0,1,2 的消息，这些消息已经被消费者读取，并且如果我关闭我的消费者一个小时。那时我生成偏移量为 3、4、5 的消息。之后，当我的消费者启动时，它应该从偏移量 3 而不是 0 读取消息。但是，它要么读取所有消息，要么读取启动 Kafka 消费者后产生的那些消息。我想阅读那些未读或未提交的消息
我尝试了 "auto.offset.reset"= "latest" 和 "earliest"。以及 "enable.auto.commit"= "true" 和 "false"。在调用 close() 方法之前，我还尝试了 commitSync() 和 commitAsync() 但没有成功。
public static KafkaConsumer createConsumer() {
Properties properties = new Properties();
properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, Constants.KAFKA_BROKER);
properties.put(ConsumerConfig.GROUP_ID_CONFIG, "testGroup");
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "50");
properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1");
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
consumer.subscribe(Collections.singleton(Constants.TOPIC));
return consumer;
}
public static void main(String[] args) {
System.out.println("");
System.out.println("----------------");
System.out.println("");
System.out.println("KAFKA CONSUMER EXAMPLE");
System.out.println("");
System.out.println("----------------");
System.out.println("");
OffsetAndMetadata offsetAndMetadataInitial = createConsumer().committed(new TopicPartition(Constants.TOPIC, 0));
System.out.println("");
System.out.println("Offset And MetaData Initial : ");
System.out.println(offsetAndMetadataInitial);
System.out.println("");
ConsumerRecords<String, String> consumerRecords = createConsumer().poll(Duration.ofSeconds(2L));
System.out.println("");
System.out.println("Count Consumer Records : " + consumerRecords.count());
System.out.println("");
Iterator<ConsumerRecord<String, String>> itr = consumerRecords.iterator();
Map<TopicPartition, OffsetAndMetadata> partationOffsetMap = new HashMap<>(4);
while (itr.hasNext()) {
ConsumerRecord record = itr.next();
System.out.println("OffSet : " + record.offset());
System.out.println("");
System.out.println("Key : " + record.key());
System.out.println("Value : " + record.value());
System.out.println("Partition : " + record.partition());
System.out.println("--------------------------");
System.out.println("");
}
createConsumer().close();
}
我只想在 kafka Consumer 中获取未读消息。如果我在某个地方错了，请纠正我。提前致谢
最佳答案
代码中的主要问题是您没有关闭用于轮询消息的消费者；这是因为每次调用 createConsumer() 都会创建一个新的 KafkaConsumer。由于您没有关闭消费者，并且只调用 poll() 一次，因此您永远不会提交已阅读的消息。
(使用自动提交，提交在自动提交间隔后的 poll() 中调用，并在 close() 中调用)
一旦您更正它应该使用以下设置:
- auto-commit=true(否则您也可以手动提交，但自动提交更简单)。
- offset-reset= earliest(这只在你第一次使用给定的 group-id 时有效，以判断你是想从主题的开头消费还是只在你开始消费后产生的消息。一旦你已经开始使用给定的 group-id 消费，您将始终从您提交的最新偏移量继续消费。)
- group-id 不得在重新启动之间更改，否则您将根据您的偏移量重置设置从头或尾重新开始。
希望对你有帮助
关于java - 如何获取kafka中未提交的消息，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/56631614/