将Kafka有效载荷转换为对象

Question

有效负载：{“post”：{“postid”：“5e22fac7f7356803e8784172”，“tags”：[“a”，“lovve”，“asldkjfsbajfdkjlnzx”，“z”]}，“date”：“2020-01-18t16:12:50.833423”，“user”：{“userid”：“5dfcfd77367c690edd91b2d9”}，“reactiontype”：“unreoved”}

我在Kafka中有这样的配置

@Bean
public Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>(
            kafkaProperties.buildConsumerProperties());

    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
            StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
            JsonDeserializer.class);
    props.put(ConsumerConfig.GROUP_ID_CONFIG,
            "magpie-trending");
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

    return props;
}

我有一个类，有相同的有效负载代码，我如何在任何时候在对象中转换它？

大家好，我有一个微服务接收这个字符串并且正在工作，但是我需要将这个字符串转换为一个特定的对象，当我使用ObjectMapper转换应用程序时，会返回这个异常：

抛出异常；嵌套异常是com.fasterxml.jackson.databind.exc.InvalidDefinitionException：无法构造com.avenuecode.magpie.trending.component.kafka.message.PostMessage的实例（没有像默认构造一样的创建者存在）：无法从对象值反序列化（没有基于委托或属性的创建者）

所以我的目标是：

public class ReactionMessage {
 private PostMessage post;
 private String date;
 private UserMessage user;
 private String reactionType;

 @JsonCreator
 public ReactionMessage(@JsonProperty("post") PostMessage post,
                       @JsonProperty("user") UserMessage user,
                       @JsonProperty("date") String date,
                       @JsonProperty("reactionType") String reactionType) {
    this.post = post;
    this.date = date;
    this.user = user;
    this.reactionType = reactionType;
 }

 @JsonCreator
 public ReactionMessage() {
 }
}

  @KafkaListener(topics = "reaction-topic", clientIdPrefix = "string", groupId = "magpie-trending")
  public void listenAsObject(ConsumerRecord<String, String> cr,
                           @Payload String payload) throws IOException {
    logger.info("Logger 1 [JSON] received key {}: Type [{}] | Payload: {} | Record: {}", cr.key(),
            typeIdHeader(cr.headers()), payload, cr.toString());

    ObjectMapper mapper = new ObjectMapper();
    ReactionMessage message = mapper.readValue(payload, ReactionMessage.class);

}

   private static String typeIdHeader(Headers headers) {
    return StreamSupport.stream(headers.spliterator(), false)
            .filter(header -> header.key().equals("__TypeId__"))
            .findFirst().map(header -> new String(header.value())).orElse("N/A");
}

Answer 1

您需要使JSONDeserializer返回您期望的对象类型

例如，Kafka内置的JSON反序列化器只返回jsonnode。

Spring JSON反序列化器有额外的属性来传递类名