spring boot 中消费者如何获取消息ID

在 Spring Boot 中，你可以使用 Spring Kafka 提供的 KafkaListenerErrorHandler 接口来获取消息的 ID。KafkaListenerErrorHandler 接口用于处理消费者在消费消息时发生的异常，可以通过实现该接口并重写其中的方法来获取消息的 ID。

首先，你需要在消费者类上添加 @KafkaListener 注解，指定要监听的 Topic 和消费者组。然后，你可以在自定义的 KafkaListenerErrorHandler 实现类中重写 handle 方法，该方法会在消费者消费消息时发生异常时被调用。在 handle 方法中，你可以通过 ConsumerRecord 对象的 recordMetadata() 方法获取到消息的 ID。

举个例子，假设你的消费者类如下所示：

@Component
@KafkaListener(topics = "my-topic", groupId = "my-group")
public class MyConsumer {

    @KafkaHandler
    public void handleMessage(String message) {
        // 消费消息
    }
}

你可以创建一个自定义的 KafkaListenerErrorHandler 实现类，如下所示：

@Component
public class MyErrorHandler implements KafkaListenerErrorHandler {

    @Override
    public Object handleError(Message<?> message, ListenerExecutionFailedException exception) throws Exception {
        if (message.getPayload() instanceof ConsumerRecord) {
            ConsumerRecord<?, ?> record = (ConsumerRecord<?, ?>) message.getPayload();
            String messageId = String.format("%s-%d-%d", record.topic(), record.partition(), record.offset());
            // 处理异常
        }
        return null;
    }
}

在 handle 方法中，我们首先判断消息的 Payload 是否为 ConsumerRecord 对象，如果是，则可以通过 ConsumerRecord 对象的 topic()、partition() 和 offset() 方法获取到消息的 ID。你可以根据自己的需求来格式化消息的 ID，上面的例子采用了 "topic-partition-offset" 的格式。

需要注意的是，如果消息的 Payload 不是 ConsumerRecord 对象，那么在 handle 方法中获取消息的 ID 将会失败。因此，在实际使用中，你需要根据自己的业务场景来决定是否需要处理这种情况。