时间戳

首页 > 解决方案 > Spring-Kafka 对 ConcurrentKafkaListenerContainerFactory 的使用不止一个 @Kafkalistener

java - Spring-Kafka 对 ConcurrentKafkaListenerContainerFactory 的使用不止一个 @Kafkalistener

问题描述

我正在使用 Spring-Kafka 框架实现对来自 Kafka 主题的消息的消费。我试图了解我为我的 Kafka 侦听器创建的 ConcurrentKafkaListenerContainerFactory 的一些用法。@KafkaListener 工作正常，如预期的那样，但是，在我的场景中，我有多个独立的侦听器，分别收听多个主题。我想知道我是否可以在所有侦听器中重用 ConcurrentKafkaListenerContainerFactory，还是必须为每个 @KafkaListener 创建一个 containerFactory。有没有一种可以在所有@Kafkalisteners 之间共享的通用 containerFactory 的方法

谢谢

标签： javaspring-bootapache-kafkaspring-kafka

解决方案

是的; 这就是重点——它是监听器容器的工厂；您通常只需要启动自动配置的一个工厂。

如果您需要侦听器的不同属性（例如反序列化器），最新版本（自 spring-kafka 2.2.4 起）允许您覆盖注释上的使用者属性。

要为单个侦听器覆盖其他属性，例如容器属性，请将侦听器容器定制器添加到工厂。

@Component
class ContainerFactoryCustomizer {

    ContainerFactoryCustomizer(AbstractKafkaListenerContainerFactory<?, ?, ?> factory) {
        factory.setContainerCustomizer(
                container -> {
                    String groupId = container.getContainerProperties().getGroupId();
                    if (groupId.equals("foo")) {
                        container.getContainerProperties().set...
                    }
                    else {
                        container.getContainerProperties().set...
                    }
                });
    }

如您所见，您可以通过访问groupId()容器属性来判断我们正在创建哪个容器。

如果您的侦听器具有非常不同的配置，您可能想要使用 2 个工厂，但是您会失去引导的自动配置功能（至少对于工厂而言）。

推荐阅读