【问题标题】：RequestReplyFuture<String, String, List<Product>> not mapped, Instead it mapped to ArrayList<LinkedHashMap>RequestReplyFuture<String, String, List<Product>> 未映射，而是映射到 ArrayList<LinkedHashMap>
【发布时间】：2020-07-26 09:14:07
【问题描述】：

我正在使用 RequestReplyFuture 将响应映射到 List，结果如下所示

    @Service
    public class ProductProducer implements IProductProducer{
      private final ReplyingKafkaTemplate<String, String, List<Product>> _replyTemplate;
        private static final Logger LOG = LoggerFactory.getLogger(ProductProducer.class);
        public ProductProducer(ReplyingKafkaTemplate<String, String, List<Product>> replyTemplate) {
            this._replyTemplate = replyTemplate;
        }
    
     @Override
        public List<ProductViewModel> GetProducts() throws InterruptedException, ExecutionException, TimeoutException {
            RequestReplyFuture<String, String, List<Product>> future =
                    this._replyTemplate.sendAndReceive(new ProducerRecord<>(ProductTopicConstants.GET_PRODUCTS, 0, null, null));
                LOG.info(future.getSendFuture().get(10, TimeUnit.SECONDS).getRecordMetadata().toString());
                List<Product> products = future.get(10, TimeUnit.SECONDS).value(); --> Property not mapped to Product
var productViewModels = products.stream().map(item -> new ProductViewModel(item.getId(),item.getName(),item.getPrice(), item.getDescription())).collect(Collectors.toList());
                return productViewModels;
        }
    }

卡夫卡配置

@Configuration
public class KafkaConfiguration {
    @Bean
    public ReplyingKafkaTemplate<String, String, List<Product>> replyer(ProducerFactory<String, String> pf,
                                                                        ConcurrentKafkaListenerContainerFactory<String, List<Product>> containerFactory) {

        containerFactory.setReplyTemplate(kafkaTemplate(pf));
        ConcurrentMessageListenerContainer<String, List<Product>> container = replyContainer(containerFactory);
        ReplyingKafkaTemplate<String, String, List<Product>> replyer = new ReplyingKafkaTemplate<>(pf, container);
        return replyer;
    }

    @Bean
    public ConcurrentMessageListenerContainer<String, List<Product>> replyContainer(
            ConcurrentKafkaListenerContainerFactory<String, List<Product>> containerFactory) {

        ConcurrentMessageListenerContainer<String, List<Product>> container =
                containerFactory.createContainer(ProductTopicConstants.GET_PRODUCTS_CONTAINER);
        container.getContainerProperties().setGroupId(ProductTopicConstants.GET_PRODUCTS_CONTAINER);
        container.setBatchErrorHandler(new BatchLoggingErrorHandler());
        return container;
    }

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



   @Bean
    public NewTopic GetProducts() {
        return TopicBuilder.name(ProductTopicConstants.GET_PRODUCTS).partitions(1).replicas(1).build();
    }
    @Bean
    public NewTopic GetProductsContainer() {
        return TopicBuilder.name(ProductTopicConstants.GET_PRODUCTS_CONTAINER).partitions(1).replicas(1).build();
    }
}

【问题讨论】：

标签： java spring spring-boot spring-kafka

【解决方案1】：

由于类型擦除，标题中的类型信息使杰克逊认为它是List<Object>。

您可以改用类型函数，为 Jackson 提供更多信息...

spring.kafka.consumer.properties.spring.json.value.type.method=com.example.demo.SomeClass.returnType

public static JavaType returnType(byte[] data, Headers headers) {
    return TypeFactory.defaultInstance()
            .constructCollectionLikeType(List.class, Product.class);
}

【讨论】：