读取完所有消息后终止Kafka Console Consumer

Question

我知道必须有办法做到这一点,但我无法弄清楚这一点.一旦我从队列中读取了所有消息,我就需要停止kafka消费者.

有人可以提供这方面的任何信息吗？

Answer 1

目前，Kafka 2.11-2.1.1 版本有一个名为的脚本kafka-console-consumer.sh。

它有一个新标志：--timeout-ms。

基本上，这个标志是当没有新日志等待时退出之前等待的最长时间。这是以毫秒为单位的。

您可以使用此属性在阅读所有消息后结束控制台消费者。

Answer 2

您可以在启动使用者时使用参数传递参数:-consumer-timeout-ms,如果在此期间未读取任何消息,则会抛出异常.例如,如果在最后2秒内没有新消息到达,则停止使用者:kafka.consumer.ConsoleConsumer -consumer-timeout-ms 2000

您可以在此处查看此选项和所有其他输入选项

完整命令看起来像这样:`./kafka-console-consumer.sh --bootstrap-server my-host:9092 --new-consumer --topic test-topic --from-beginning --timeout-ms 2000`我尝试了它并且它工作正常,虽然它确实在它退出时打印错误消息和堆栈跟踪. (3认同)

Answer 3

您可以将 SimpleConsumerShell 与 no-wait-at-logend 选项一起使用。请参阅系统工具-SimpleConsumerShell

例如：

./kafka-run-class.bat kafka.tools.SimpleConsumerShell --broker-list localhost:9092 --topic kafkademo --partition 0 --no-wait-at-logend