默认的 kafkaListenerContainerFactory 是如何工作的答案

【问题标题】：How does default kafkaListenerContainerFactory work默认的 kafkaListenerContainerFactory 是如何工作的
【发布时间】：2020-06-29 07:17:10
【问题描述】：

我有一个 kafka 消费者 java 应用程序。我正在从该应用程序中阅读两个单独的 kafka 主题。

对于主题 1，有一个名为 kafkaListenerContainerFactory 的 KafkaListenerContainerFactory，下面是代码 sn-p。消息采用 avro 格式。 Pojo1 是 pojo 类使用 avro 架构构建。

 @Bean
public ConcurrentKafkaListenerContainerFactory<String, Pojo1> kafkaListenerContainerFactory() {

    ConcurrentKafkaListenerContainerFactory<String, Pojo1> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());

    return factory;
}

为了消费来自topic1的消息，我们有以下方法

@kafkaListener (topics="topic1") public boolean readTopic1(ConsumerRecord<String,Pojo1> record){ // logic }

现在这个应用程序也在读取来自 topic2 的消息，它也有 avro 消息，Pojo2 是使用相应的 avro 模式构建的 pojo 类。

要从topic2中读取，我们有以下方法

@kafkaListener (topics="topic2") public boolean readTopic2(ConsumerRecord<String,Pojo2> record){ // logic }

现在我不知道来自 topic2 的消费消息是如何按预期工作的。

kafkaListenerContainerFactory 配置了 Pojo1，所以如何从 Pojo2 的 topic2 读取消息。据我所知，kafka 仅在缺少名称为“kafkaListenerContainerFactory”的 bean 时才创建默认容器工厂，但在我的应用程序中我们已经为 topic1 构建了一个 kafkaListenerContainerFactory。

据此应该有另一个为 Pojo2 创建的 KafkaListenerContainerFactory 并且应该在使用 @kafkaListener (topics="topic2", containerFactory="factoryForTopic2") 时给出参考

谁能帮我理解 KafkaListenerContainerFactory 如何为具有不同消息架构的不同主题工作。

【问题讨论】：

标签： java apache-kafka spring-kafka

【解决方案1】：

这叫做类型擦除；泛型类型仅在编译时适用。在运行时，只要您的反序列化器创建了正确的类型，它就可以工作。

使用ConcurrentKafkaListenerContainerFactory<String, Object> 可能更简洁以避免任何混淆；尤其是在使用可以创建不同类型的 avro 或 json 反序列化器时。

也就是说，您的 bean 被称为 importDecisionMessageKafkaListenerContainerFactory。对于使用非标准工厂的监听器，它的名称必须在监听器的containerFactory 属性中指定；否则使用kafkaListenerContainerFactory。

【讨论】：

嗨，加里，感谢您的回复。我的 bean 实际上只是 kafkaListenerContainerFactory ，但有问题我打错了。将更新它。因此，如果我正确理解您的观点，假设我的应用程序中有一个 kafkaListenerContainerFactory bean，那么即使消息具有不同的 avro 模式，我也可以使用来自不同主题的消息，其余所有消费者属性都与反序列化器一样（对于我的场景来说，两者都会是 avro 反序列化器）

【解决方案2】：

这仅适用于您的情况，因为反序列化器能够将 String 反序列化为两个对象。但是假设您使用 Avro，并且您仍然拥有 Pojo1 和 Pojo2。在这种情况下，您将需要两个容器工厂。如果您没有为 Pojo2 提供一个异常，则会引发异常 - MessageConversionException: cannot convert String to Pojo2 for GenericMessage。解决方案是定义@kafkaListener (topics="topic2", containerFactory="factoryForTopic2")

【讨论】：