【Kafka基础】消费者命令行完全指南：从基础到高级消费

Kafka消费者是消息系统的关键组成部分，掌握/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh工具的使用对于调试、测试和监控都至关重要。本文将全面介绍该工具的各种用法，帮助您高效地从Kafka消费消息。

1 基础消费模式

1.1 从最新位置消费

/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh \--bootstrap-server 192.168.10.33:9092 \--topic testtopic

参数解析：

--bootstrap-server: 指定Kafka集群地址
--topic: 指定消费的主题名称

特点：

从该消费者组最后提交的offset开始消费
如果没有提交记录，则从最新消息开始

1.2 从最早位置消费

/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh \--bootstrap-server 192.168.10.33:9092 \--topic testtopic \--from-beginning

关键参数：

--from-beginning: 从主题最早的消息开始消费

应用场景：

数据回溯
新消费者组初始化

2 消息元数据展示

2.1 显示消息Key

/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh \--bootstrap-server 192.168.10.33:9092 \--topic testtopic \--property print.key=true \--property key.separator=":"

参数说明：

print.key=true: 显示消息key
key.separator: 指定key/value分隔符

2.2 显示完整元数据

/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh \--bootstrap-server 192.168.10.33:9092 \--topic testtopic \--property print.key=true \--property print.value=true \--property print.partition=true \--property print.offset=true \--property print.timestamp=true \--property key.separator=":" \--property line.separator="\n"

元数据参数：

print.partition: 显示分区号
print.offset: 显示消息offset
print.timestamp: 显示时间戳

3 精准消费控制

3.1 指定分区消费

/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh \--bootstrap-server 192.168.10.33:9092 \--topic testtopic \--partition 1

参数说明：

--partition: 指定消费的分区编号

3.2 指定Offset消费

/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh \--bootstrap-server 192.168.10.33:9092 \--topic testtopic \--partition 0 \--offset 1000

参数说明：

--offset: 指定开始消费的offset位置

3.3 消费超时设置

/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh \--bootstrap-server 192.168.10.33:9092 \--topic testtopic \--timeout-ms 10000

参数说明：

--timeout-ms: 设置无消息时的超时时间（毫秒）

4 消费者组管理

4.1 使用消费者组

/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh \--bootstrap-server 192.168.10.33:9092 \--topic testtopic \--group mygroup

参数说明：

--group: 指定消费者组名称

特点：

支持offset自动提交
支持消费者组rebalance

4.2 手动控制offset提交

/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh \--bootstrap-server 192.168.10.33:9092 \--topic testtopic \--group mygroup \--property enable.auto.commit=false

参数说明：

enable.auto.commit=false: 关闭自动提交

5 高级消费配置

5.1 限制消费消息数

/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh \--bootstrap-server 192.168.10.33:9092 \--topic testtopic \--max-messages 100

参数说明：

--max-messages: 最大消费消息数

5.2 过滤消费

/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh \--bootstrap-server 192.168.10.33:9092 \--topic testtopic \--property filter.key.regex="test.*"

参数说明：

filter.key.regex: 按key正则过滤

5.3 消费速率控制

/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh \--bootstrap-server 192.168.10.33:9092 \--topic testtopic \--property fetch.min.bytes=1024 \--property fetch.max.wait.ms=500

参数说明：

fetch.min.bytes: 最小拉取字节数
fetch.max.wait.ms: 最大等待时间

6 常用命令扩展

6.1 消费多个主题

/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh \--bootstrap-server 192.168.10.33:9092 \--whitelist "topic1|topic2"

6.2 显示消息头信息

/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh \--bootstrap-server 192.168.10.33:9092 \--topic testtopic \--property print.headers=true

6.3 消费特定时间后的消息

/export/home/kafka_zk/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh \--bootstrap-server 192.168.10.33:9092 \--topic testtopic \--property offsets.storage=kafka \--property timestamp=1625097600000

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/diannao/77272.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！