java - 无法按需从 Kafka 读取记录

Question

我正在使用生产者将数据推送到 kafka 队列。

然后我需要按需消费/读取数据，基本上我希望当触发rest api时我应该能够从kafka读取一条记录并返回。我不希望消费者在数据到达队列后立即被调用，而应该在我想要的时候被调用。

正常的读取方法是放置一个 while 循环，然后调用 poll 方法，然后当数据到达队列时不断调用该方法。但我想控制何时阅读记录。

Answer 1

您仍然可以按需创建使用者并让它处理并返回单个记录。像这样的事情:

...
   properties.setProperty(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1");
   properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
   KafkaConsumer consumer = new KafkaConsumer<>(properties);
   consumer.subscribe(Arrays.asList(topic));
   try {
        while (!timedOut) {
            ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
            if (!records.isEmpty()) {
                // process record and return
            }
        }
        // error, no record found
    } catch (InterruptException e) {
    } finally {
        consumer.close();
    }
...