是否可以分别使用来自kafka主题的每条消息？

Question

好的，目标是：我有一个应该发送邮件的服务，如果失败，我的Kafka制作人将把这封邮件发送到Kafka主题。第二个程序每两分钟查看一次主题，应该只使用一条消息（最早的一条），然后重试发送，如果失败，程序应该将此消息返回主题。

我已经有了一个消费者，但问题是，它会消耗我直到现在还没有使用消费者的所有消息。但我希望他只吃最老的，他以前从未吃过。

这是我的实际消费者：

“CustMessage”是我为测试而创建的一个类，它只是一个具有不同属性的对象，如日期、消息和其他东西。为了测试，代码有点修改，所以这个消费者正在运行endles。它也只显示消息而不是处理它，所以只有“System.out.println（message.displayMessage（））；”。

公共级消费电子{

public static void main(String[] args) {

    Properties properties = new Properties();
    properties.put("bootstrap.servers", "localhost:9092");
    properties.put("group.id", "happyConsumer");
    properties.put("auto.offset.reset", "earliest");
    properties.put("enable.auto.commit", "false");
    properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

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

    consumer.subscribe(Arrays.asList("mymailtopic"));

    while (true) {
        ObjectMapper om = new ObjectMapper();
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            customMessage message=null;
            try {
                 message = om.readValue(record.value(), customMessage.class);
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
            System.out.println(message.displayMessage());
        }
    }


}

}

对于实际的代码，服务也可以工作，但我至少可以尝试它是否也可以像我在第一段中提到的那样工作。所以，我的问题是：

1. 甚至可以只使用该主题中的一条消息吗

我在用Java

Answer 1

您可以使用max.poll.records属性逐条使用消息，在配置中将此属性设置为1

对poll（）的单个调用中返回的最大记录数。

properties.put("max.poll.records", 1);