将apache kafka轮询迁移到spring kafka

Question

 while (!done.get()) {
    ConsumerRecords<byte[], <byte[]> records = kafkaConsumer.poll(<MAX_VALUE>);
    if (records.isEmpty()) {
        retryCount++;
        Thread.sleep(<some_time>);
    } else {
    // Process records;
    }
    if (retryCount > <max_retry_count>) {
    done.set(true); 
    }    
  }

尝试了以下方法：1）使用spring kafka注释(@kafkalistener)，但它不允许我们控制轮询。2）创建了“ConcurrentMessageListenerContainer”，并且setupMessageListener将记录添加到队列中进行轮询。这给了我们对使用者的控制。

我想知道，我的方向是正确的吗？使用Spring-Kafka有什么更好的解决方案来达到上述要求呢？

Answer 1

不清楚你说的“对消费者的控制”是什么意思。创建容器与使用@kafkalistener相同（容器是在封面下创建的）。

Spring使用“消息驱动”的方法。

您可以设置IdleEventInterval，如果在此期间没有收到任何记录，容器将发布ListenerContainerIdleEvent。您可以使用ApplicationListenerbean或@EventListener方法侦听这些事件。