如何在SpringKafka上创建自定义解串器？

Question

我已经创建了一个消费者来接收来自Kafka主题的消息，但是我正在手动解析消息，有没有一种方法可以使用自定义的反序列化器来自动解析这个主题消费者？

// I've tried to replace the second "String" with "MyMessage" but it didn't work
// e.g: onMessage(message: ConsumerRecord<String, MyMessage>)
fun onMessage(message: ConsumerRecord<String, String>) {
    val myMessage = try {
        jacksonObjectMapper.readValue<MyMessage>(message.value())
    } catch (ex: Exception) {
        logger.error(ex.message)
        null
    }

    myMessage?.let{ // do stuff }
}

预期结果是：

fun onMessage(message: ConsumerRecord<String, MyMessage>) {
    // do something with message.value() parameter that will be for sure a MyMessage instance.
}

如果出现解析错误，我可以设置回调来处理此错误吗？

Answer 1

对只需实现您自己的并将其添加到消费者配置值中。反序列化器。

https://kafka.apache.org/documentation/#consumerconfigs_value.deserializer

如果您使用Spring Boot:

https://docs.spring.io/spring-boot/docs/current/reference/html/application-properties.html#application-属性.集成.Spring.kafka.耗电元件.值分解器

如果这样配置，它不受Spring管理；如果希望Spring管理它，可以通过构造函数或setter将它添加到< code > DefaultKafkaProducerFactory 中。

Spring for Apache Kafka提供了一个JSON反序列化程序：

https://docs.spring.io/spring-kafka/docs/current/reference/html/#json-serde