如何让kafka消费者处于等待消费状态而不是直接退出？

Question

使用spring-boot spirng-kafka, 建了2个项目， 一个producer,一个consumer,

测试的时候使用producer向kafka发送几条消息， 然后启动consumer能够成功消费到消息。

但我的问题是，这个consumer是消费完就关闭了（因为不是web程序），如何让consumer启动后就处于等待状态，然后一旦发现有新的消息进来就马上进行消费？我是在单元测试里面跑的，如果发布成jar，有如何部署运行使得消费者一直在等待消费状态呢？

Answer 1

引用来自“battyman”的评论

你用的是spring-kafka官方的例子吧，那个例子里面就设置了消费完就退出的逻辑。O(∩_∩)O

首先声明了

private final CountDownLatch latch = new CountDownLatch(3);

然后在spring-boot的启动函数中调用await

latch.await(60, TimeUnit.SECONDS);

既然调用了await，就会阻塞当前run函数线程，也就是所谓的spring-boot主函数

消费线程调用countDown，将计数器减1，当计数器变为0的时候，就会取消之前await阻塞的线程让他继续运行，从而run那个函数执行完成，最终调用了close()

 

所以，修改的地方就很明显了：

不要加阻塞，只要让run线程不退出就可以，大不了你也可以不countDown()让它一直阻塞

Answer 2

你用的是spring-kafka官方的例子吧，那个例子里面就设置了消费完就退出的逻辑。O(∩_∩)O

首先声明了

private final CountDownLatch latch = new CountDownLatch(3);

然后在spring-boot的启动函数中调用await

latch.await(60, TimeUnit.SECONDS);

既然调用了await，就会阻塞当前run函数线程，也就是所谓的spring-boot主函数

消费线程调用countDown，将计数器减1，当计数器变为0的时候，就会取消之前await阻塞的线程让他继续运行，从而run那个函数执行完成，最终调用了close()

 

所以，修改的地方就很明显了：

不要加阻塞，只要让run线程不退出就可以，大不了你也可以不countDown()让它一直阻塞

Answer 3

引用来自“battyman”的评论

用while(true)，然后用consumer的poll方法获取records

Answer 4

用while(true)，然后用consumer的poll方法获取records