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

【问题标题】：Is it posible to consume each message from a kafka topic separately?是否可以分别使用来自 kafka 主题的每条消息？
【发布时间】：2021-03-27 09:00:06
【问题描述】：

好的，这是目标：我有一个应该发送邮件的服务，如果失败，我的 Kafka 生产者会将此邮件发送到 Kafka 主题。第二个程序每两分钟查看一次主题，并且应该只使用一条消息（最旧的一条）并重试发送它，如果失败，程序应该将此消息带回主题。

我已经有一个消费者，但问题是，它消费了我到现在为止还没有消费的所有消息。但我希望他只吃最老的，他以前没有吃过。

这是我的实际消费者：

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

公共类 ConsumerOne {

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());
        }
    }


}

}

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

是否甚至可以只使用来自该主题的一条消息？
如果是，我该怎么做？

我在整个服务中使用 Java 和 quarkus（我认为是最新版本的 Kafka）。

【问题讨论】：

你的意思是一条一条地消费消息？还是使用特定的消息？
@Deadpool 是的，只有一个

标签： java apache-kafka kafka-consumer-api quarkus

【解决方案1】：

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

在一次 poll() 调用中返回的最大记录数。

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

【讨论】：