Spring Kafka JsonDesirialization MessageConversionException无法解析类名Class not found

Question

我有两个服务应通过Kafka进行通信。我们将第一个服务称为WriteService，将第二个服务称为QueryService。

在WriteService端，我为生产者提供了以下配置。

@Configuration
public class KafkaProducerConfiguration {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        // list of host:port pairs used for establishing the initial connections to the Kakfa cluster
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                JsonSerializer.class);

        return props;
    }

    @Bean
    public ProducerFactory<String, Object> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<String, Object> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

我正在尝试发送com.example.project.web.routes.dto.RouteDto类的对象

在QueryService端，使用者配置定义如下。

@Configuration
@EnableKafka
public class KafkaConsumerConfiguration {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.groupid}")
    private String serviceGroupId;

    @Value("${spring.kafka.consumer.trusted-packages}")
    private String trustedPackage;

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                bootstrapServers);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                JsonDeserializer.class);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, serviceGroupId);
        props.put(JsonDeserializer.TRUSTED_PACKAGES, trustedPackage);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        return props;
    }

    @Bean
    public ConsumerFactory<String, Object> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, Object> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());

        return factory;
    }

}

侦听器具有以下定义。有效负载类具有完全限定的名称-com.example.project.clientqueryview.module.routes.messaging.kafka.RouteDto

@KafkaListener(topics = "${spring.kafka.topics.routes}",
            containerFactory = "kafkaListenerContainerFactory")
    public void listenForRoute(ConsumerRecord<String, RouteDto> cr,
                               @Payload RouteDto payload) {
        logger.info("Logger 1 [JSON] received key {}: Type [{}] | Payload: {} | Record: {}", cr.key(),
                typeIdHeader(cr.headers()), payload, cr.toString());
    }

    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");
    }

发送消息时，出现以下错误

原因：org.springframework.messaging.converter.MessageConversionException：无法解析类名。找不到课程[com.example.project.web.routes.dto.RouteDto];嵌套异常为java.lang.ClassNotFoundException：com.example.project.web.routes.dto.RouteDto

错误很明显。但是我不明白为什么默认情况下它具有此行为。我不希望在不同的服务中使用相同的程序包，这根本没有任何意义。

我还没有找到一种方法来禁用此功能并使用提供给侦听器的类，并用@Payload注释]

如何解决此问题，而无需手动配置映射器？

我有两个应该通过Kafka进行通信的服务。让我们将第一个服务称为WriteService，将第二个服务称为QueryService。在WriteService端，我对...

Answer 1

[如果您使用的是spring-kafka-2.2.x，则可以通过JsonDeserializer docs的重载构造函数禁用默认标头>

[从版本2.2开始，您可以使用具有boolean useHeadersIfPresent（默认为true）的重载构造函数之一，将反序列化器显式配置为使用提供的目标类型并忽略标头中的类型信息。以下示例显示了如何执行此操作：

Answer 2

@@ Deadpool，如何获得将不同的JSON对象路由到不同的KafkaHandlers的方法，我已经关注了您的帖子，但我无法使其正常工作，LinkedHashMap有效负载始终只能使用一种方法，那不是我想要的方法，您可以请帮助