有没有写Kafka主题的Spring时间表的例子？

Question

我们正在尝试在指定的窗口时间从Kafka读取数据（所以我们有Kafka消费者），这意味着避免在其他时间读取数据。但是，我们不确定如何在时间段到期后关闭消费者。我想知道是否有任何如何做到这一点的示例？非常感谢您对我们的帮助。

Answer 1

我有相同的用例，但是我编写了调度程序，指定一个批处理的最大轮询记录，并保存了一个计数器。如果计数器匹配最大轮询记录，那么我认为该批处理已经完成，因为它已经处理了在一次轮询中获得的记录。

然后，我取消订阅该主题并关闭消费者。下次调度程序运行时，它将再次处理指定的最大轮询记录限制。

fixedDelayString实现的目的是，一旦前一个完成，调度程序在指定的时间限制后启动。

@EnableScheduling
public class MessageScheduler{

        @Scheduled(initialDelayString = "${fixedInitialDelay.in.milliseconds}", fixedDelayString = "${fixedDelay.in.milliseconds}")
        public void run(){

            /*write your kafka consumer here with manual commit*/

            /*once your batch is finished processing unsubcribe and close the consumer*/
                kafkaConsumer.unsubscribe();
                kafkaConsumer.close();
        }

}

Answer 2

您可以禁用自动启动，然后使用 Kafka接收器endpoint注册表启动和停止方法手动启动 kafka 容器，@KafkaListener生命周期管理

public class KafkaConsumer {

 @Autowired
 private KafkaListenerEndpointRegistry registry;

  @KafkaListener(id = "myContainer", topics = "myTopic", autoStartup = "false")
  public void listen(...) { ... }

  @Schedule(cron = "")
  public void scheduledMethod() {

   registry.start();

   registry.stop()
  }

但在上述方法中，不能保证来自 kafka 的所有消息都将在该时间范围内被使用（这取决于加载和处理速度）