Kafka听众，获取所有消息

Question

大家好。我有一个Kafka项目，使用SpringKafka来听一个明确的主题。我需要一天听一次所有的信息，把它们放到一个集合中，然后在那里找到特定的信息。我无法理解如何用一个@KafkaListener方法读取所有消息。我的班级是：

@Component
public class KafkaIntervalListener {

    public CountDownLatch intervalLatch = new CountDownLatch(1);
    private final SCDFRunnerService scdfRunnerService;

    public KafkaIntervalListener(SCDFRunnerService scdfRunnerService) {
        this.scdfRunnerService = scdfRunnerService;
    }

    @KafkaListener(topics = "${kafka.interval-topic}", containerFactory = "intervalEventKafkaListenerContainerFactory")
    public void intervalListener(IntervalEvent event) throws UnsupportedEncodingException, JSONException {
        System.out.println("Recieved interval message: " + event);
        IntervalType type = event.getType();
        Instant instant = event.getInterval();
        List<IntervalEvent> events = new ArrayList<>();
        events.add(event);
        events.size();


        this.intervalLatch.countDown();
    }

}

我的事件集合的大小始终为1；我尝试使用不同的循环，但后来，我的收藏被归档了530000次。

更新：我已经找到了一种方法来做它与factory.setBatchListener（真）；但我需要找到启动它与@调度（cron = "${ kafka.cron}，区=欧洲/莫斯科）.现在这个方法总是在听。现在我尝试这样的东西：

    @Scheduled(cron = "${kafka.cron}", zone = "Europe/Moscow")
public void run() throws Exception {
    kafkaIntervalListener.intervalLatch.await();
}

它不工作，在调试模式下，我的断点永远不会在这个网站上工作。

Answer 1

侦听器容器在设计上是消息驱动的。

对于按需获取消息，最好直接使用KafkaConsumerAPI，并使用poll（）方法获取消息。