使用Spring Kafka实现Apache Kafka头的自定义转换

Question

我正在使用Spring Boot 2.3.0和Spring Kafka 2.5.0，在我的KafkaListener中，我试图将MessageHeaders映射到一个自定义类。下面的代码可以工作，但给出了byte[]格式的头，然后我必须将其转换为侦听器内部的类（并对每个侦听器重复此操作），这是我希望避免的。

    @Slf4j
    @Component
    @KafkaListener(topics = {"${spring.kafka.topics.simple}"}, groupId = "consumerGroup", 
    containerFactory = "kafkaListenerContainerFactory")
    public class RequestConsumer {

       @KafkaHandler
       public void listen(@Payload CustomerDetails customerDetails, @Header("sec") byte[] 
       principle, @Headers MessageHeaders messageHeaders) {
           log.info("Received a CustomerDetails");
       }

       @KafkaHandler(isDefault = true)
       public void listen(@Payload(required = false) GenericRecord object, @Headers 
       MessageHeaders messageHeaders) {
           log.info("Received an unexpected html" target="_blank">object");
       }
    }

当我将代码更改为：

    @KafkaHandler
    public void listen(@Payload CustomerDetails customerDetails, @Header("sec") PreAuthenticatedAuthenticationToken principle, @Headers MessageHeaders messageHeaders) {
        log.info("Received a CustomerDetails");
    }

我查看了https://docs.spring.io/spring-kafka/reference/html/#Headers文档，其中讨论了从header到MessageHeaders的头映射，反之亦然，但没有转换这些头的示例。这个错误似乎暗示我可以注册一个转换器，但没有找到如何，尽管许多谷歌搜索和尝试逐步通过Spring Kafka的代码。

在这方面提供帮助将不胜感激。

干杯,奥斯卡

Answer 1

我一直坚持下去，我找到了一个适当的方法来实现这一点。虽然我没有找到转换头的方法，但可以编写一个handlerMethodArgumentResolver，允许我们在侦听器中使用自定义对象。在这段代码中，很容易将头转换为PreAuthenticatedAuthenticationToken对象，然后我可以直接在侦听器中使用该对象。

更多详细信息:https://docs.spring.io/spring-kafka/docs/2.5.3.release/reference/html/#adding-custom-handlermethodargumentresolver-to-kafkalistener

已验证正在工作。

干杯。