Kafka常用命令之kafka-console-consumer.sh解读

Kafka是一个分布式消息系统，常用于构建实时流数据管道和数据处理应用程序。kafka-console-consumer.sh是Kafka的一个命令行消费者，可以用来消费Kafka中的消息。本文将详细讲解kafka-console-consumer.sh的使用方法和常用参数。

kafka-console-consumer.sh命令的基础用法

命令格式

bin/kafka-console-consumer.sh --bootstrap-server <kafka-broker:port> [--topic <topic>] [--from-beginning] [--consumer-property <consumer-property-name>=<consumer-property-value>]

参数解释

• --bootstrap-server：Kafka broker的地址和端口号。
• --topic：要消费的Kafka主题。
• --from-beginning：从Kafka主题的起始位置开始消费消息，而不是从当前位置开始。
• --consumer-property：Kafka消费者的配置属性，可以使用任何有效的消费者属性。

示例1

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic

解释：从本地的Kafka broker中消费test-topic主题的消息。

示例2

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic --from-beginning

解释：从test-topic主题的起始位置开始消费消息。

进阶用法

消费者组

Kafka中的消费者可以通过分组的方式对同一个主题的消息进行消费，这些消费者组共享主题的订阅。kafka-console-consumer.sh也支持消费者组的功能。

命令格式

bin/kafka-console-consumer.sh --bootstrap-server <kafka-broker:port> --topic <topic> --consumer-property group.id=<consumer-group-id>

参数解释

• --bootstrap-server：Kafka broker的地址和端口号。
• --topic：要消费的Kafka主题。
• --consumer-property group.id：设置消费者所属的组的ID。

示例3

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic --consumer-property group.id=my-consumer-group

解释：使用my-consumer-group作为消费者组的ID，从test-topic主题中消费消息。

生产者非阻塞

有些时候，我们需要在生产者没有消息推送的情况下，消费者仍然保持运行状态，不阻塞等待生产者推送消息。kafka-console-consumer.sh也支持这种非阻塞的模式。

命令格式

bin/kafka-console-consumer.sh --bootstrap-server <kafka-broker:port> --topic <topic> --consumer-property consumer.timeout.ms=-1

参数解释

• --bootstrap-server：Kafka broker的地址和端口号。
• --topic：要消费的Kafka主题。
• --consumer-property consumer.timeout.ms：消费者在等待消息的超时时间，单位毫秒。当设置-1时，消费者不会在读取不到新数据时退出。

示例4

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic --consumer-property consumer.timeout.ms=-1

解释：从test-topic主题中不断地消费消息，直到手动停止消费者进程。

总结

通过对kafka-console-consumer.sh的使用和参数解释的介绍，我们可以更加灵活地使用Kafka消费者，提高对Kafka的消费效率。