Spring-Kafka использует ConcurrentKafkaListenerContainerFactory для более чем одного @Kafkalistener

Question

Я работаю над реализацией потребления сообщений из тем Kafka с использованием среды Spring-Kafka. Я пытаюсь понять, как я использую ConcurrentKafkaListenerContainerFactory, которую я создаю для моего слушателя Kafka. @KafkaListener работает нормально, и, как и ожидалось, в моем сценарии у меня более одного независимого слушателя, слушающего более одного Topi c соответственно. Я хотел бы знать, могу ли я повторно использовать ConcurrentKafkaListenerContainerFactory среди всех моих слушателей, или мне нужно создать один containerFactory для @KafkaListener. Есть ли способ иметь обобщенную c containerFactory, которая может быть использована всеми @ Kafkalisteners

Спасибо вам

Answer 1

Да; в этом весь смысл - это фабрика для слушающих контейнеров; Обычно вам нужна только одна фабрика, которая автоматически конфигурируется при загрузке.

Если вам нужны разные свойства (например, десериализаторы) для прослушивателя, последние версии (начиная с 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 фабрики, если ваш слушатели имеют совершенно другую конфигурацию, но тогда вы теряете возможности автоматической конфигурации загрузки (по крайней мере, для фабрики).