Как динамически создавать несколько потребителей в Spring Kafka - PullRequest
Новая
       21

Как динамически создавать несколько потребителей в Spring Kafka

1 голос
/ 23 марта 2019

У меня есть два кластера Kafka, IP-адреса для которых я динамически выбираю из базы данных. Я использую @KafkaListener для создания слушателей. Теперь я хочу создать несколько слушателей Kafka во время выполнения в зависимости от атрибута сервера начальной загрузки (значения, разделенные запятыми), каждый из которых слушает кластер. Подскажите, пожалуйста, как мне этого добиться?

Spring-boot: 2.1.3. ВЫПУСК Кафка-2.0.1 Java-8

1 Ответ

2 голосов
/ 23 марта 2019

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

@SpringBootApplication
@EnableConfigurationProperties(ClusterProperties.class)
public class So55311070Application {

    public static void main(String[] args) {
        SpringApplication.run(So55311070Application.class, args);
    }

    private final Map<String, MyListener> listeners = new HashMap<>();

    @Bean
    public ApplicationRunner runner(ClusterProperties props, ConsumerFactory<Object, Object> cf,
            ConcurrentKafkaListenerContainerFactory<Object, Object> containerFactory,
            ApplicationContext context, KafkaListenerEndpointRegistry registry) {

        return args -> {
            AtomicInteger instance = new AtomicInteger();
            Arrays.stream(props.getClusters()).forEach(cluster -> {
                Map<String, Object> consumerProps = new HashMap<>(cf.getConfigurationProperties());
                consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster);
                String groupId = "group" + instance.getAndIncrement();
                consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
                containerFactory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerProps));
                this.listeners.put(groupId, context.getBean("listener", MyListener.class));
            });
            registry.getListenerContainers().forEach(c -> System.out.println(c.getGroupId())); // 2.2.5 snapshot only
        };
    }

    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public MyListener listener() {
        return new MyListener();
    }

}

class MyListener {

    @KafkaListener(topics = "so55311070")
    public void listen(String in) {
        System.out.println(in);
    }

}


@ConfigurationProperties(prefix = "kafka")
public class ClusterProperties {

    private String[] clusters;

    public String[] getClusters() {
        return this.clusters;
    }

    public void setClusters(String[] clusters) {
        this.clusters = clusters;
    }

}

kafka.clusters=localhost:9092,localhost:9093

spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=false

Результат 

group0
group1
...
2019-03-23 11:43:25.993  INFO 74869 --- [ntainer#0-0-C-1] o.s.k.l.KafkaMessageListenerContainer    
    : partitions assigned: [so55311070-0]
2019-03-23 11:43:25.994  INFO 74869 --- [ntainer#1-0-C-1] o.s.k.l.KafkaMessageListenerContainer    
    : partitions assigned: [so55311070-0]

РЕДАКТИРОВАТЬ

Добавить код для повторной попытки запуска сбойных контейнеров.

Оказывается, нам не нужна локальная карта слушателей, в реестре есть карта всех контейнеров, включая те, которые не удалось запустить. 

@SpringBootApplication
@EnableConfigurationProperties(ClusterProperties.class)
public class So55311070Application {

    public static void main(String[] args) {
        SpringApplication.run(So55311070Application.class, args);
    }

    private boolean atLeastOneFailure;

    private ScheduledFuture<?> restartTask;

    @Bean
    public ApplicationRunner runner(ClusterProperties props, ConsumerFactory<Object, Object> cf,
            ConcurrentKafkaListenerContainerFactory<Object, Object> containerFactory,
            ApplicationContext context, KafkaListenerEndpointRegistry registry, TaskScheduler scheduler) {

        return args -> {
            AtomicInteger instance = new AtomicInteger();
            Arrays.stream(props.getClusters()).forEach(cluster -> {
                Map<String, Object> consumerProps = new HashMap<>(cf.getConfigurationProperties());
                consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, cluster);
                String groupId = "group" + instance.getAndIncrement();
                consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
                attemptStart(containerFactory, context, consumerProps, groupId);
            });
            registry.getListenerContainers().forEach(c -> System.out.println(c.getGroupId())); // 2.2.5 snapshot only
            if (this.atLeastOneFailure) {
                Runnable rescheduleTask = () -> {
                    registry.getListenerContainers().forEach(c -> {
                        this.atLeastOneFailure = false;
                        if (!c.isRunning()) {
                            System.out.println("Attempting restart of " + c.getGroupId());
                            try {
                                c.start();
                            }
                            catch (Exception e) {
                                System.out.println("Failed to start " + e.getMessage());
                                this.atLeastOneFailure = true;
                            }
                        }
                    });
                    if (!this.atLeastOneFailure) {
                        this.restartTask.cancel(false);
                    }
                };
                this.restartTask = scheduler.scheduleAtFixedRate(rescheduleTask,
                        Instant.now().plusSeconds(60),
                        Duration.ofSeconds(60));
            }
        };
    }

    private void attemptStart(ConcurrentKafkaListenerContainerFactory<Object, Object> containerFactory,
            ApplicationContext context, Map<String, Object> consumerProps, String groupId) {

        containerFactory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerProps));
        try {
            context.getBean("listener", MyListener.class);
        }
        catch (BeanCreationException e) {
            this.atLeastOneFailure = true;
        }
    }

    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public MyListener listener() {
        return new MyListener();
    }

    @Bean
    public TaskScheduler scheduler() {
        return new ThreadPoolTaskScheduler();
    }

}

class MyListener {

    @KafkaListener(topics = "so55311070")
    public void listen(String in) {
        System.out.println(in);
    }

}
...