Spring 云流反序列化不同的pojo

Question

我正在使用 Kstream 来使用 kafka 消息并将其保存到我的数据库中。这些消息属于不同的 pojo，目前我正在使用对象映射器来创建对象，然后将它们保存在数据库中。我读过可以使用 Jsondeserialzerserde，但我不确定如何使用它映射到不同的 pojo。为每个 pojo 自定义 serde 是没有意义的。请帮忙。提前致谢。

这是我的代码：

    public Consumer<KStream<String, String>> process() {

        return input ->
                inpu.foreach((key, value) -> {
                    ObjectMapper mapper = new ObjectMapper();
                    try {

                        if(value.contains("Teacher"))
                        {
                            Teacher teacher= mapper.readValue(value,Teacher.class);
                            teacherRepository.save(teacher);
                        }
                        else if(value.contains("Student"))
                        {
                            Student student= mapper.readValue(value,Student.class);
                            studentRepository.save(student)
                        }
                        else  if(value.contains("Principal"))
                        {
                            Principal principal= mapper.readValue(value,Principal.class);
                            principalRepository.save(Principal);
                        }
                    } catch (JsonProcessingException e) {
                        e.printStackTrace();
                    }

                });    }

}

Answer 1

Spring Cloud Stream 中的 Kafka Streams 活页夹不直接提供任何机制来执行您的要求。您的消费者的类型签名表示您正在将其作为 String 使用，因此活页夹可以推断出该信息，而无需您明确提供任何 Serde。但是，如果您想使用 jackson 将 String 进一步转换为其他类型，您需要在业务逻辑中自行完成，因为您已经拥有它。如果您只有几个有限类型，我认为这样做没有问题。

Spring 云流反序列化不同的pojo

Spring Cloud stream deserialize different pojos

java

spring

apache-kafka

spring-cloud-stream

apache-kafka-streams