我正在使用下面的示例来使用 spring Kafka 消费者读取消息。我的用例要求每次生成一条消息时,听众每次都从头开始阅读。
@KafkaListener(
id = "grouplistener",
topicPartitions = {
@TopicPartition(
topic = "mycompactedtopic", partitionOffsets = @PartitionOffset(partition = "0", initialOffset = "0")
)
}
)
public void onReceiving(
String payload, @Header(KafkaHeaders.OFFSET) Integer offset,
@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
@Header(KafkaHeaders.RECEIVED_TOPIC) String topic
) {
log.info(
"Processing topic = {}, partition = {}, offset = {}, payload= {}",
topic, partition, offset, payload
);
}
我似乎只能在应用程序启动时从头开始读取它,然后它通常只会继续使用消息。
有没有办法让它每次都从头开始寻找?
使用具有 1 个分区的压缩主题来保存配置列表。然后需要由休息端点调用,它应该显示完整的唯一配置列表
你应该实现它的方式是使用 Kafka Streams 和 KTable 并在你的 REST 层后面设置 交互式查询。不是需要倒带自身以获得系统最新状态的标准消费者。
Kafka Connect框架中已经有这样的例子,它有一个配置主题,你只能访问最新的值
GET /connectors/name/config
,只有重启它或扩展到更多实例,它才会消耗所有消息再次。 Schema Registry 也是一个例子,它存储了 _schemas
主题中所有模式的内部 Hashmap,并有一个用于读取、插入、删除的 REST API
本质上,当您获得给定键的新配置时,您可以用一个全新的值“替换”给定键的旧值,或者您可以以某种方式将旧值与新数据“合并” .
这里我将如何实现它。您需要实现
ConsumerSeekAware
接口并对onPartitionsAssigned
方法做一些实现。如果您在重新启动应用程序时发送环境变量,也可以按需进行 seekToBegining。我还没有实现它!
@Service
@EnableKafka
public class Service implements ConsumerSeekAware {
@KafkaListener(topics = "${topicName}", groupId = "${groupId}")
public void listen(@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
@Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts,
@Payload List<String> messageBatch
) {
//do a bunch of stuff
}
@Override
public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
String topic= Optional.ofNullable(System.getProperty(TOPIC_NAME)).orElseThrow(()->new RuntimeException("topicName needs to be set"));
assignments.keySet().stream().filter(partition->topic.equals(partition.topic()))
.forEach(partition -> callback.seekToBeginning(topic, partition.partition()));
}
@Override
public void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {}
@Override
public void registerSeekCallback(ConsumerSeekCallback callback) {}
}
@KafkaListener(topicPartitions
= @TopicPartition(topic = "test", partitionOffsets = {
@PartitionOffset(partition = "0", initialOffset = "0")}),groupId = "foo",
containerFactory = "kafkaListenerContainerFactory")
public void listenAllMsg(@Payload String message,@Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
System.out.println(" all msg Received Messasge in group 'foo': " + message+"RECEIVED_PARTITION_ID - "+partition);
}
在卡夫卡 2.3.1
我认为你应该尝试编写 ConsumerSeekAwareListener,并在每次阅读消息时寻求 0 偏移量。听起来像是疯狂的解决方法,但它可能会有所帮助。希望这会对你有所帮助:-)
class Listener implements ConsumerSeekAware {
private final ThreadLocal<ConsumerSeekCallback> seekCallBack = new ThreadLocal<>();
----Override all methods that are needed----
@KafkaListener(...)
public void listen(@Payload String message) {
this.seekCallBack.get().seek(topic, partition, 0);
}
}
}
@Nimo1981 所以这是一个使用纯 Java 的实现。我不确定它是否满足您的需求。所以基本上我提交的偏移量为 0,(意思是,即使我从 Kafka 主题中读到,我也会回到开头的偏移量。)我不确定你是否考虑过这个实现,但请让我知道如果这个是你要找的
省略 CommitCountObj。那不是你需要的。 所以默认情况下 offsetMap 将有下一个这样的偏移量记录,
offsetMap.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1, "some commit success message"));
但是对于你的用例,我做了一些修改,当消费者没有重启时它工作得很好
offsetMap.put(new TopicPartition(record.topic(), record.partition()), 新的 OffsetAndMetadata(0,"没有提交完成"));
public class KafkaConsumerClass {
private static final org.apache.log4j.Logger log = org.apache.log4j.Logger.getLogger(KafkaConsumerClass.class);
private CommitCountClass commitCountobj = new CommitCountClass();
public Consumer<String, List<FeedBackConsumerClass>> createConsumer() {
Map<String, Object> consumerProps = new HashMap<String, Object>();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:7070,localhost:7072");
consumerProps.put(ConsumerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG, 50000);
consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "first_group-client1");
// consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "first_group");
// consumerProps.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, KafkaConsumerInterceptor.class);
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
consumerProps.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 15000);
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
consumerProps.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1500);
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
return new KafkaConsumer<String, List<FeedBackConsumerClass>>(consumerProps);
}
public void consumeRecord() {
log.info("Coming inside consumer consumer");
ArrayList<String> topicList = new ArrayList<String>();
topicList.add("topic1");
commitCountobj.setCount(0);
Consumer<String, List<FeedBackConsumerClass>> kafkaConsumer = createConsumer();
kafkaConsumer.subscribe(topicList);
log.info("after subscribing");
Map<TopicPartition, OffsetAndMetadata> offsetMap = new HashMap<>();
while (true) {
ConsumerRecords<String, List<FeedBackConsumerClass>> recordList = kafkaConsumer.poll(Long.MAX_VALUE);
// kafkaConsumer.seekToBeginning(kafkaConsumer.assignment());
log.info("Inside while loop:" + recordList);
if (!recordList.isEmpty()) {
recordList.forEach(record -> {
int i = 0;
System.out.println(record.toString());
// we can make the call to the API here
// call the db here or any API and process the record
// then call the code to commit
// since the commit is switched off, it becomes a developers responsibility to do the auto commit
offsetMap.put(new TopicPartition(record.topic(), record.partition()),
new OffsetAndMetadata(0, "no metadata/offset commited"));
// here we are incrementing the offsetMap so that we are making sure we are storing the
// next set of offsets in the map
if (commitCountobj.getCount() % 1000 == 0) {
kafkaConsumer.commitAsync(offsetMap, new OffsetCommitCallback() {
@Override
public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets,
Exception exception) {
// TODO Auto-generated method stub
if (exception != null) {
// retry it now with a sync
// possibility of error occuring here as well
// so capture the exception and exit the consumer gracefully
kafkaConsumer.commitSync();
log.error(exception.getMessage());
}
}
});
}
commitCountobj.setCount(i++);
});
}
}
}
}