Есть ли проблемы с этим способом запуска бесконечного цикла в приложении Spring Boot?

Question

У меня есть приложение Spring Boot, и оно должно обрабатывать некоторые потоковые данные Kafka. Я добавил бесконечный цикл в класс CommandLineRunner, который будет запускаться при запуске. Там есть потребитель Кафки, которого можно разбудить. Я добавил хук отключения с Runtime.getRuntime().addShutdownHook(new Thread(consumer::wakeup));. Буду ли я столкнуться с какими-либо проблемами? Есть ли более идиоматический способ сделать это весной? Должен ли я использовать @Scheduled вместо этого? В приведенном ниже коде отсутствуют конкретные элементы реализации Kafka, но в остальном он завершен.

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.errors.WakeupException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Properties;


    @Component
    public class InfiniteLoopStarter implements CommandLineRunner {

        private final Logger logger = LoggerFactory.getLogger(this.getClass());

        @Override
        public void run(String... args) {
            Consumer<AccountKey, Account> consumer = new KafkaConsumer<>(new Properties());
            Runtime.getRuntime().addShutdownHook(new Thread(consumer::wakeup));

            try {
                while (true) {
                    ConsumerRecords<AccountKey, Account> records = consumer.poll(Duration.ofSeconds(10L));
                    //process records
                }
            } catch (WakeupException e) {
                logger.info("Consumer woken up for exiting.");
            } finally {
                consumer.close();
                logger.info("Closed consumer, exiting.");
            }
        }
    }

Answer 1

Реализация выглядит нормально, но использование CommandLineRunner не предназначено для этого. CommandLineRunner используется для запуска некоторых задач при запуске только один раз. С точки зрения дизайна это не очень элегантно. Я бы предпочел использовать компонент адаптера пружинной интеграции с Kafka. Вы можете найти пример здесь https://github.com/raphaelbrugier/spring-integration-kafka-sample/blob/master/src/main/java/com/github/rbrugier/esb/consumer/Consumer.java.

Answer 2

Чтобы просто ответить на мой собственный вопрос, я взглянул на библиотеки интеграции Kafka, такие как Spring-Kafka и Spring Cloud Stream, но интеграция с реестром схем Confluent либо не завершена, либо мне не совсем понятна. Этого достаточно для примитивов, но нам это нужно для типизированных объектов Avro, которые проверяются реестром схемы. Теперь я внедрил решение, не зависящее от Кафки, основанное на ответе Spring Boot - Лучший способ запустить фоновый поток при развертывании

Окончательный код выглядит так:

@Component
public class AccountStreamConsumer implements DisposableBean, Runnable {

    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    private final AccountService accountService;
    private final KafkaProperties kafkaProperties;
    private final Consumer<AccountKey, Account> consumer;

    @Autowired
    public AccountStreamConsumer(AccountService accountService, KafkaProperties kafkaProperties,
                                 ConfluentProperties confluentProperties) {

        this.accountService = accountService;
        this.kafkaProperties = kafkaProperties;

        if (!kafkaProperties.getEnabled()) {
            consumer = null;
            return;
        }

        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
        props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, confluentProperties.getSchemaRegistryUrl());
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, kafkaProperties.getSecurityProtocolConfig());
        props.put(SaslConfigs.SASL_MECHANISM, kafkaProperties.getSaslMechanism());
        props.put(SaslConfigs.SASL_JAAS_CONFIG, PlainLoginModule.class.getName() + " required username=\"" + kafkaProperties.getUsername() + "\" password=\"" + kafkaProperties.getPassword() + "\";");
        props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getAccountConsumerGroupId());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);

        consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList(kafkaProperties.getAccountsTopicName()));

        Thread thread = new Thread(this);
        thread.start();
    }

    @Override
    public void run() {
        if (!kafkaProperties.getEnabled())
            return;

        logger.debug("Started account stream consumer");
        try {
            //noinspection InfiniteLoopStatement
            while (true) {
                ConsumerRecords<AccountKey, Account> records = consumer.poll(Duration.ofSeconds(10L));
                List<Account> accounts = new ArrayList<>();
                records.iterator().forEachRemaining(record -> accounts.add(record.value()));
                if (accounts.size() != 0)
                    accountService.store(accounts);
            }
        } catch (WakeupException e) {
            logger.info("Account stream consumer woken up for exiting.");
        } finally {
            consumer.close();
        }
    }

    @Override
    public void destroy() {
        if (consumer != null)
            consumer.wakeup();

        logger.info("Woke up account stream consumer, exiting.");
    }
}

Answer 3

Я не уверен, что у вас возникнут какие-либо проблемы, но это немного грязно - у Spring действительно хорошая встроенная поддержка работы с Kafka, так что я бы к этому склонился (в Интернете достаточно документации по этому вопросу). , но хороший это: https://www.baeldung.com/spring-kafka).

Вам понадобится следующая зависимость:

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.2.2.RELEASE</version>
</dependency>

Настроить так же просто, добавив аннотацию @EnableKafka в класс конфигурации, а затем настроить компоненты Listener и ConsumerFactory

После настройки вы можете легко настроить потребителя следующим образом:

@KafkaListener(topics = "topicName")
public void listenWithHeaders(
  @Payload String message, 
  @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
      System.out.println("Received Message: " + message"+ "from partition: " + partition);
}