代码之家 › 专栏 › 技术社区 › Yashdeep Sharma

如何将GenericMessageJacksonDeserializer与Spring Kafka和Spring集成在一起?

spring-kafka spring-integration

1

Yashdeep Sharma · 技术社区 · 8 年前

我想知道如果我的kafka消息是GenericMessage类型的,如何将GenericMessageJacksonDeserializer与spring kafka consumer factory一起使用?

@Bean
public ConsumerFactory<?, GenericMessage> consumerFactory(KafkaProperties kafkaProperties) {
    final Map<String, Object> consumerProperties = kafkaProperties.buildConsumerProperties();
    final DefaultKafkaConsumerFactory<?, GenericMessage> consumerFactory = new DefaultKafkaConsumerFactory<>(consumerProperties);
    consumerFactory.setValueDeserializer(GenericMessageJacksonDeserializer.class);
    return consumerFactory;
}

1 回复 | 直到 8 年前

1

Artem Bilan 8 年前

这个 consumerFactory.setValueDeserializer() 是用于 org.apache.kafka.common.serialization.Deserializer
我想你是说 org.springframework.integration.support.json.GenericMessageJacksonDeserializer 这正是 ObjectMapper JacksonJsonUtils.messagingAwareMapper() ,也就是说,呃。。。,真的意识到了这一点 GenericMessageJacksonDeserializer 已经
当你得到它的时候 ,您应该将其注入。。。风俗 Deserializer 消费者工厂。setValueDeserializer() . 这个 messagingAwareMapper org.springframework.kafka.support.serializer.JsonDeserializer 因为这确实需要一个特定的类型来进行反序列化。但是,我们可以尝试以下方法:
```
consumerFactory.setValueDeserializer(new JsonDeserializer(GenericMessage.class, JacksonJsonUtils.messagingAwareMapper()));
```

推荐文章

user25089203 · 替代Kafka 3.6.2中的@EnableBinding和@StreamListener

1 年前

springenthusiast · 如何将springboot中的重复配置作为jar并用作依赖项

2 年前

bohunn · Kafka Streams-无法搜索状态存储

2 年前

Andrey P. · 添加spring-cloud-starter-bus-Kafka依赖项后,Kafka消费者没有收到消息

2 年前

StevenPG · Spring Cloud Streams错误处理不起作用

2 年前

pankaj · Kafka中的电流偏移量怎么会这么高,而日志端偏移量太低,导致负Kafka滞后过高?

2 年前