使用 AggregatingReplyingKafkaTemplate 处理多个响应答案

【问题标题】：Handle multiple responses with AggregatingReplyingKafkaTemplate使用 AggregatingReplyingKafkaTemplate 处理多个响应
【发布时间】：2020-03-29 10:16:45
【问题描述】：

我有一个问题场景需要解决，类似于 Handle multiple responses with ReplyingKafkaTemplate

我了解 AggregatingReplyingKafkaTemplate（使用 spring-kafka-2.3.7）可以帮助解决这个问题，但是，我正在努力寻找正确的 bean 配置并获得可以聚合来自多个消费者的结果的场景。以下是我的 bean 配置，但是，我收到编译错误说无法解析 GenericMessageListenerContainer 的构造函数

    @Bean
    public AggregatingReplyingKafkaTemplate<String, Model, Model> replyKafkaTemplate(ProducerFactory<String, Model> pf,
                                                                                     KafkaMessageListenerContainer<String, Collection<ConsumerRecord<String, Model>>> replyContainer,
                                                                                     BiPredicate<List<ConsumerRecord<String, Model>>, Boolean> releaseStrategy) {
        return new AggregatingReplyingKafkaTemplate<>(pf, replyContainer, releaseStrategy);
    }
@Bean
    public GenericMessageListenerContainer<String, Collection<ConsumerRecord<String, Model>>>  replyContainer(ConsumerFactory<String, Model> cf) {
        ContainerProperties containerProperties = new ContainerProperties(requestReplyTopic);
        return new KafkaMessageListenerContainer<String, Collection<ConsumerRecord<String, Model>>>(cf,
                containerProperties);
    }

任何如何使用 AggregatingReplyingKafkaTemplate 的简单示例都会有所帮助。

非常感谢，

【问题讨论】：

标签： spring spring-kafka

【解决方案1】：

得到编译错误说无法解析GenericMessageListenerContainer的构造函数

在哪里？我没有看到你试图构建一个。

你的replyContainer bean 应该返回KafkaMessageListenerContainer，因为它是replyKafkaTemplate 工厂方法中声明的类型。

例如，请参阅我对this question 的回答。

使用ConcurrentKafkaListenerContainerFactory 来创建容器是最简单的。如果你使用 Spring Boot，它会自动为你配置一个；如果没有，您需要将其配置为@Bean

【讨论】：