Spring Boot Service 按需消费 kafka 消息答案

【问题标题】：Sping Boot Service consume kafka messages on demandSpring Boot Service 按需消费 kafka 消息
【发布时间】：2020-08-20 22:23:41
【问题描述】：

我有一个需求，需要一个 Spring Boot Rest 服务，客户端应用程序将每 30 分钟调用一次，并且服务将返回

最新消息的数量基于查询参数中指定的数量，例如http://messages.com/getNewMessages?number=10 在这种情况下应该返回 10 条消息
基于查询参数中指定的数量和偏移量的消息数，例如http://messages.com/getSpecificMessages?number=5&start=123 在这种情况下应该返回 5 条消息，从偏移量 123 开始。

我有一个简单的独立应用程序，它工作正常。这是我测试的内容，并希望将其纳入服务中。

public static void main(String[] args) {
        // create kafka consumer
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "my-first-consumer-group");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, args[0]);

        Consumer<String, String> consumer = new KafkaConsumer<>(properties);

        // subscribe to topic
        consumer.subscribe(Collections.singleton("test"));
        consumer.poll(0);

        //get to specific offset and get specified number of messages
        for (TopicPartition partition : consumer.assignment())
            consumer.seek(partition, args[1]);

        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(5000));

        System.out.println("Total Record Count ******* : " + records.count());      

        for (ConsumerRecord<String, String> record : records) {
            System.out.println("Message: " + record.value());
            System.out.println("Message offset: " + record.offset());           
            System.out.println("Message: " + record.timestamp());
            Date date = new Date(record.timestamp());
            Format format = new SimpleDateFormat("yyyy MM dd HH:mm:ss.SSS");
            System.out.println("Message date: " + format.format(date));
        }
        consumer.commitSync();

因为我的消费者会在 Spring Boot Service 中按需想知道我该如何实现这一点。如果我在 application.properties 中放入那些在启动时注入的属性，我在哪里指定属性，但我如何在运行时控制 MAX_POLL_RECORDS_CONFIG。任何帮助表示赞赏。

【问题讨论】：

标签： spring-boot rest apache-kafka kafka-consumer-api spring-kafka

【解决方案1】：

MAX_POLL_RECORDS_CONFIG 只会影响您的 kafka-client 将记录返回给您的 spring 服务，它永远不会减少消费者从 kafka-server 轮询的字节数

看上图，不管你的起始offset = 150还是190，kafka server都会从(offset=110, offset=190)返回整个数据，kafka server甚至不知道有多少条记录返回给consumer，他只知道字节大小 = (220 - 110)

所以我认为你可以自己控制记录数，目前它是由kafka客户端jar控制的，它们都占用你的jvm本地内存

【讨论】：

在spring服务中哪里可以配置配置。

【解决方案2】：

您的问题的答案是here，代码示例的答案是this answer。均由出色的 Gary Russell 撰写，他是 Spring Kafka 背后的主要人物或主要人物之一。

TL;DR:

如果您想在运行时任意回退分区，请让您的侦听器实现 ConsumerSeekAware 并获取对 ConsumerSeekCallback。

【讨论】：