英文:
Limit number of messages from Kafka in Spring Batch
问题
我如何限制队列大小为100万条？每当队列被轮询并且有空闲空间时，它应该重新开始监听。
英文:
I am storing the data from KafkaListener in a ConcurrentLinkedQueue to be processed. Currently it consumes as many data as it can and completely fills up RAM. How do I limit the number of messages in the queue so that when it reaches the limit the KafkaListener pauses.
ConcurrentLinkedQueue<Message> queue = new ConcurrentLinkedQueue<>();@KafkaListener(topics = "topic",id = "topic-kafka-listener",groupId = "batch-processor",containerFactory = "kafkaListenerContainerFactory")public void receive(@NotNull @Payload List<Message> messages) {queue.addAll(messages);}
How do I limit the queue size to say 1 million?
Whenever the queue is polled and there is free space it should start listening again.
OR
How do I limit the rate at which Kafka consumes messages to say 100,000 messages per second?
答案1
得分: 0
以下是您要翻译的内容：
"Instead of using annotation I used the KafkaConsumer object to poll for data manually. With this there is more control."
Map<String, Object> consumerConfig = Map.of("bootstrap.servers", "localhost:9092","key.deserializer", StringDeserializer.class,"value.deserializer", StringDeserializer.class,"group.id", "batch-processor","max.poll.records", 480000);KafkaConsumer<String, Message> kafkaConsumer = new KafkaConsumer<>(consumerConfig);kafkaConsumer.subscribe(List.of("topic"));
public void receive() {ConsumerRecords<String, Message> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000));consumerRecords.forEach(record -> queue.add(record.value()));}
英文:
Instead of using annotation I used the KafkaConsumer object to poll for data manually. With this there is more control.
Map<String, Object> consumerConfig = Map.of("bootstrap. Servers", "localhost:9092","key.deserializer", StringDeserializer.class,"value.deserializer", StringDeserializer.class,"group.id", "batch-processor","max.poll.records", 480000);KafkaConsumer<String, Message> kafkaConsumer = new KafkaConsumer<>(consumerConfig);kafkaConsumer.subscribe(List.of("topic"));
public void receive() {ConsumerRecords<String, Message> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000));consumerRecords.forEach(record -> queue. Add(record. Value()));}
答案2
得分: 0
Kafka Consumer 可以通过 API 暂停和恢复。有关详细信息，请查看以下方法：
英文:
Kafka Consumer can be paused and resumed by API. Check this methods for details:
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库，让每个人都能够通过互相帮助和分享经验来进步。
评论