带有Kafka使用者的Spring Boot作业调度程序

Question

我正在一个POC中工作，我想消费来自Kafka主题“用户”的消息。尝试实现一旦spring boot scheduler在预定时间或cron时间触发，消费者应该从Kafka主题读取消息，然后我们应该开始从Kafka主题逐个消费现有消息并处理这些消息，当所有的消息被消费，那么Kafka消费者应该停止。调度程序应在cron时间触发并再次启动进程。

我已经尝试了以下方法来实现这一点，尽管我正在努力确定如何从我的调度器方法schedularMsgConsumeKakfa调用consumer（字符串消息）方法，以及任何我们有更好的结构来在Spring Boot调度器中使用Kafka消息并在实现tasklet接口的类中编写Kafka消费者方法的html" target="_blank">示例，等等。

@Configuration
@Service
public class SchedularMsgConsumeKafkaController2 {

    @Autowired
    private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

    @KafkaListener(topics = "users", id = "full-part-id", containerGroup = "full-part-group", autoStartup = "false")
    public void consume(String message) throws IOException, InterruptedException {
        System.out.println(String.format("#### -> Consumed message -> %s", message));
        System.out.println(String.format("Really happy"));
        MessageListenerContainer listenerContainer = kafkaListenerEndpointRegistry.getListenerContainer("full-part-id");
        listenerContainer.stop();
    }

    @Scheduled(fixedDelay = 30000, initialDelay = 15000)
    public void schedularMsgConsumeKakfa() throws Exception {
        MessageListenerContainer listenerContainer = kafkaListenerEndpointRegistry.getListenerContainer("full-part-id");
        listenerContainer.start();
    }

}

Answer 1

首先，@schedule不是Spring Boot特性。它是Spring Framework所固有的:https://docs.Spring.io/spring-framework/docs/current/reference/html/integration.html#scheduling。

第二个:@kafkalistener是Spring for Apache Kafka项目的一部分:https://docs.Spring.io/spring-kafka/docs/current/reference/html/#kafka-listener-annotation。

因此，这两个特性都可以在Spring Boot Framework之外使用。我知道这与问题无关，但最好用它们的专有名称来称呼事物。

@kafkalistener&@schedule的混合不是很好的解决方案，因为您混合了许多不兼容的问题。

最好使用一些不同的解决方案，在这些方案中，您不需要处理start()/stop()，并且不会使用额外的线程来增加环境的开销。

考虑一下Spring集成及其kafkamessageSource实现:https://docs.Spring.io/spring-integration/reference/html/kafka.html#kafka-inbound-pollable。这对你来说可能是新的东西，但这是值得的，因为你不打算做太多的东西自己。

“您自己”是指使用ConsumerFactoryAPI和从@sedule手动调用KafkaConsumer.poll()。

对于@schedule解决方案，可能没有这样的示例，因为它确实足够复杂。Spring Integration中没有KafkaMessageSource的示例，因为它非常简单，并且仅限于配置适当的通道适配器。