我有一个主题，我可以从中接收不同类型的json。但是，当消费者试图阅读消息时，我似乎遇到了一个异常。我试图添加额外的bean名称，但没有成功。似乎是在尝试从主题中阅读，并尝试转换为从主题中阅读的所有类型。是否有一种方法可以指定只应为特定输入类型启用特定工厂。有没有别的办法解决这个问题。
错误
原因：org.springframework.messaging.converter.messageconversionexception:无法将genericmessage[payload=com.lte.assessment.assessment.attemptrequest]从[com.lte.assessment.analytics.data.sitelevelanalyticsrequest]转换为[com.lte.assessment.assessments]。assessmentattemptrequest@68eb637f，标题={kafka\u offset=22，kafka\u consumer=org.apache.kafka.clients.consumer。kafkaconsumer@252d8ffb，kafka\u timestamptype=create\u time，kafka\u receivedmessagekey=null，kafka\u receivedpartitionid=0，kafka\u receivedtopic=ltetopic，kafka\u receivedtimestamp=1546117529267}
配置

@EnableKafka
@Configuration
public class KafkaConfig {
    static Map<String, Object> config = new HashMap();

    static {
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        config.put(ConsumerConfig.GROUP_ID_CONFIG, "group_id");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
    }

    @Bean
    public ConsumerFactory<String, AssessmentAttemptRequest> assessmentAttemptDetailsEntityConsumerFactory() {
        JsonDeserializer<AssessmentAttemptRequest> deserializer = new JsonDeserializer<>();
        deserializer.addTrustedPackages("com.lte.assessment.assessments");
        return new DefaultKafkaConsumerFactory(config, new StringDeserializer(), deserializer);
    }

    @Bean(name="aaKafkaListenerFactory")
    public ConcurrentKafkaListenerContainerFactory aaKafkaListenerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, AssessmentAttemptRequest> factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(assessmentAttemptDetailsEntityConsumerFactory());
        return factory;
    }

    @Bean
    public ConsumerFactory<String, AssessmentQuestionAnalyticsEntity> assessmentQuestionAnalyticssEntityConsumerFactory() {
        JsonDeserializer<AssessmentQuestionAnalyticsEntity> deserializer = new JsonDeserializer<>();
        deserializer.addTrustedPackages("com.lte.assessment.assessments");
        return new DefaultKafkaConsumerFactory(config, new StringDeserializer(), deserializer);
    }

    @Bean(name="aqKafkaListenerFactory")
    public ConcurrentKafkaListenerContainerFactory aqKafkaListenerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, AssessmentQuestionAnalyticsEntity> factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(assessmentQuestionAnalyticssEntityConsumerFactory());
        return factory;
    }

    @Bean
    public ConsumerFactory<String, SiteLevelAnalyticsEntity> siteLevelAnalyticsEntityConsumerFactory() {
        JsonDeserializer<SiteLevelAnalyticsEntity> deserializer = new JsonDeserializer<>();
        deserializer.addTrustedPackages("com.lte.assessment.assessments");
        return new DefaultKafkaConsumerFactory(config, new StringDeserializer(), deserializer);
    }

    @Bean("slaKafkaListenerFactory")
    public ConcurrentKafkaListenerContainerFactory slaKafkaListenerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, SiteLevelAnalyticsEntity> factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(siteLevelAnalyticsEntityConsumerFactory());
        return factory;
    }
}

服务

@Service
public class TopicObserver implements
        ConsumerSeekAware.ConsumerSeekCallback,ConsumerSeekAware{

    @Autowired
    private AssessmentAttemptService assessmentAttemptService;

    @Autowired
    private AssessmentQuestionService assessmentQuestionService;

    @Autowired
    private SiteLevelAnalyticsService siteLevelAnalyticsService;

    private final ThreadLocal<ConsumerSeekCallback> seekCallBack = new ThreadLocal<>();

    @KafkaListener(topics = "ltetopic", groupId = "group_id", containerFactory = "aaKafkaListenerFactory")
    public void consumeAttemptDetails(AssessmentAttemptRequest request) {
        assessmentAttemptService.storeAttempDetails(request);
    }

    @KafkaListener(topics = "ltetopic", groupId = "group_id", containerFactory = "aqKafkaListenerFactory")
    public void setAssessmentQeustionAnalytics(AssessmentQuestionRequest request) {
        assessmentQuestionService.storeQuestionDetails(request);
    }

    @KafkaListener(topics = "ltetopic", groupId = "group_id", containerFactory = "slaKafkaListenerFactory")
    public void siteLevelAnalytics(SiteLevelAnalyticsRequest request) {
        siteLevelAnalyticsService.storeSiteLevelDetailsDetails(request);
    }
}