SprintBoot kafka值序列化程序

Question

我有一个使用apachekafka(开源流处理软件(的SpringBoot项目我有这个听众

@KafkaListener(topics = "test")
public String consume(Hostel hostel) throws IOException {
}

这个序列化程序

public class HostelSerializer implements Deserializer<Hostel> {
private final ObjectMapper objectMapper;
public InputRequestMessageSerializer(ObjectMapper objectMapper) {
this.objectMapper = objectMapper;
}
@Override
public void configure(Map<String, ?> map, boolean b) {
this.configure(map,b);
}
@SneakyThrows
@Override
public Hostel deserialize(String s, byte[] bytes) {
return objectMapper.readValue(bytes, Hostel.class);
}
@Override
public void close() {
this.close();
}
}

和属性：

spring:
kafka:
consumer:
bootstrap-servers: localhost:9092
group-id: group_id
topics: test
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: com.kafka.config.HostelSerializer

当我收到消息时，我有这个错误

Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot convert from [java.lang.String] 
to [com.message.Hostel] for GenericMessage [.....}]

查看日志时，它不会从属性中获取值：

2020-04-09 16:41:30,840 INFO  :  gid: trace= span= [main] o.a.k.c.consumer.ConsumerConfig ConsumerConfig values: 
key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer

Answer 1

混合反序列化。由于您配置了使用者，因此只需要使用正确的反序列化接口和实现：

kafka:
consumer:
bootstrap-servers: localhost:9092
group-id: group_id
topics: test
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: com.kafka.config.HostelDeserializer

和

public class HostelDeserializer implements Deserializer<Hostel> { .. }