默认的kafkaListenerContainerFactory是如何工作的

Question

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

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

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

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

    return factory;
}

对于使用来自topic1的消息，我们有以下方法

@kafkaListener（topics=“topic1”）公共布尔readTopic1（ConsumerRecord

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

阅读主题2，我们有以下方法

@kafkaListener（topics=“topic2”）公共布尔readTopic2（ConsumerRecord

现在，我不明白topic2的消息是如何按预期工作的。

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

根据规定，应该为Pojo2创建另一个KafkaListenerContainerFactory，并且在使用@kafkaListener（topics=“topic2”，containerFactory=“factoryForTopic2”）

有人能帮我理解KafkaListenerContainerFactory是如何使用不同的消息模式为不同的主题工作的吗？




Answer 1

这只适用于您的情况，因为反序列化程序能够将字符串反序列化为两个对象。但是想象一下，你使用Avro，你仍然有Pojo1和Pojo2。在这种情况下，你需要两个集装箱工厂。如果不为Pojo2提供一个异常，将引发异常-MessageConversionException:无法将字符串转换为GenericMessage的Pojo2。解决方案是定义@kafkaListener（topics=“topic2”，containerFactory=“factoryForTopic2”）

Answer 2

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

它可能是更干净的使用ConAutomtKafkaListenerContainerFactory

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