少女祈祷中...
NEKO
知识库

kafka 常用命令

2026-09-09 9 0 0

主题操作类

创建主题

# 创建主题（指定分区数和副本数）
kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --topic topic-name \
  --partitions 3 \
  --replication-factor 2

列出所有主题

kafka-topics.sh --list --bootstrap-server localhost:9092

查看主题详情

# 查看单个主题详情
kafka-topics.sh --describe \
  --bootstrap-server localhost:9092 \
  --topic topic-name

# 查看所有主题详情
kafka-topics.sh --describe --bootstrap-server localhost:9092

修改主题

# 增加分区数（只能增加不能减少）
kafka-topics.sh --alter \
  --bootstrap-server localhost:9092 \
  --topic topic-name \
  --partitions 5

# 修改主题配置
kafka-configs.sh --alter \
  --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name topic-name \
  --add-config retention.ms=86400000

删除主题

kafka-topics.sh --delete \
  --bootstrap-server localhost:9092 \
  --topic topic-name

生产消息类

控制台生产者（基础）

kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic topic-name

带 key 的生产者

kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic topic-name \
  --property parse.key=true \
  --property key.separator=:

输入格式：key:value

从文件生产消息

cat messages.txt | kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic topic-name

生产时指定 acks 和压缩

kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic topic-name \
  --producer-property acks=all \
  --producer-property compression.type=gzip

消费消息类

控制台消费者（从最新开始消费）

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic topic-name

从头开始消费

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic topic-name \
  --from-beginning

指定消费者组消费

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic topic-name \
  --group group-name

显示 key 和 value

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic topic-name \
  --property print.key=true \
  --property print.value=true \
  --property key.separator=:

显示消息元数据（分区、偏移量、时间戳）

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic topic-name \
  --property print.timestamp=true \
  --property print.offset=true \
  --property print.partition=true

消费者组管理类

列出所有消费者组

kafka-consumer-groups.sh --list --bootstrap-server localhost:9092

查看消费者组详情（消费进度）

kafka-consumer-groups.sh --describe \
  --bootstrap-server localhost:9092 \
  --group group-name

重置消费者组偏移量（需先停止消费者）

# 重置到最早偏移量
kafka-consumer-groups.sh --reset-offsets \
  --bootstrap-server localhost:9092 \
  --group group-name \
  --topic topic-name \
  --to-earliest --execute

# 重置到最新偏移量
kafka-consumer-groups.sh --reset-offsets \
  --bootstrap-server localhost:9092 \
  --group group-name \
  --topic topic-name \
  --to-latest --execute

# 重置到指定时间戳（毫秒）
kafka-consumer-groups.sh --reset-offsets \
  --bootstrap-server localhost:9092 \
  --group group-name \
  --topic topic-name \
  --to-datetime 2026-01-01T00:00:00.000 \
  --execute

# 重置到指定偏移量
kafka-consumer-groups.sh --reset-offsets \
  --bootstrap-server localhost:9092 \
  --group group-name \
  --topic topic-name:0 \
  --to-offset 100 --execute

删除消费者组（需组内无活跃消费者）

kafka-consumer-groups.sh --delete \
  --bootstrap-server localhost:9092 \
  --group group-name

分区管理类

查看分区详情

# 查看 topic 所有分区及 Leader/ISR 信息
kafka-topics.sh --describe --bootstrap-server localhost:9092 --topic topic-name

查看指定分区的偏移量范围

# 查看最小偏移量
kafka-run-class.sh kafka.tools.GetOffsetShell \
  --bootstrap-server localhost:9092 \
  --topic topic-name \
  --partitions 0 \
  --time -2

# 查看最大偏移量
kafka-run-class.sh kafka.tools.GetOffsetShell \
  --bootstrap-server localhost:9092 \
  --topic topic-name \
  --partitions 0 \
  --time -1

增加分区（重复注意只增不减）

kafka-topics.sh --alter \
  --bootstrap-server localhost:9092 \
  --topic topic-name \
  --partitions 10

配置管理类

查看主题配置

kafka-configs.sh --describe \
  --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name topic-name

修改主题配置

# 修改消息保留时间（7天）
kafka-configs.sh --alter \
  --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name topic-name \
  --add-config retention.ms=604800000

# 修改最大消息大小
kafka-configs.sh --alter \
  --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name topic-name \
  --add-config max.message.bytes=10485760

删除主题配置（恢复默认）

kafka-configs.sh --alter \
  --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name topic-name \
  --delete-config retention.ms

查看 Broker 配置

kafka-configs.sh --describe \
  --bootstrap-server localhost:9092 \
  --entity-type brokers \
  --entity-name 0

Broker 管理类

查看 Broker 列表

kafka-broker-api-versions.sh --bootstrap-server localhost:9092

查看 Broker 元数据

zookeeper-shell.sh localhost:2181 <<< "ls /brokers/ids"

查看 Broker 磁盘使用情况

kafka-log-dirs.sh --describe \
  --bootstrap-server localhost:9092 \
  --broker-list 0

查看 Broker 运行时配置

kafka-configs.sh --describe \
  --bootstrap-server localhost:9092 \
  --entity-type brokers \
  --entity-name 0

ACL 权限管理类

添加 ACL（生产者权限）

kafka-acls.sh --add \
  --bootstrap-server localhost:9092 \
  --topic topic-name \
  --allow-principal User:producer_user \
  --operation Write \
  --operation Describe \
  --operation Create

添加 ACL（消费者权限）

kafka-acls.sh --add \
  --bootstrap-server localhost:9092 \
  --topic topic-name \
  --group group-name \
  --allow-principal User:consumer_user \
  --operation Read \
  --operation Describe

查看 ACL

kafka-acls.sh --list \
  --bootstrap-server localhost:9092 \
  --topic topic-name

删除 ACL

kafka-acls.sh --remove \
  --bootstrap-server localhost:9092 \
  --topic topic-name \
  --allow-principal User:producer_user \
  --operation Write

数据迁移与副本类

重新分配分区（生成迁移计划）

kafka-reassign-partitions.sh --generate \
  --bootstrap-server localhost:9092 \
  --topics-to-move-json-file topics.json \
  --broker-list "0,1,2"

执行分区重新分配

kafka-reassign-partitions.sh --execute \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file reassign.json

查看重新分配进度

kafka-reassign-partitions.sh --verify \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file reassign.json

修改副本因子

# 1. 生成 JSON 计划
kafka-reassign-partitions.sh --generate \
  --bootstrap-server localhost:9092 \
  --topics-to-move-json-file topics.json \
  --broker-list "0,1,2"

# 2. 修改 JSON 中的副本数为目标值
# 3. 执行计划
kafka-reassign-partitions.sh --execute \
  --bootstrap-server localhost:9092 \
  --reassignment-json-file increase-replication.json

运维与工具类

查看 Topic 消息总数（所有分区合计）

kafka-run-class.sh kafka.tools.GetOffsetShell \
  --bootstrap-server localhost:9092 \
  --topic topic-name \
  --time -1 \
  | awk -F ':' '{sum += $3} END {print sum}'

查看 Broker 日志大小

du -sh /kafka-logs/*

验证 Kafka 连接

kafka-broker-api-versions.sh --bootstrap-server localhost:9092

导出/导入 Topic 配置（备份用）

# 导出配置
kafka-configs.sh --describe \
  --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name topic-name > topic-config.txt

# 导入配置（重建时使用）
kafka-configs.sh --alter \
  --bootstrap-server localhost:9092 \
  --entity-type topics \
  --entity-name topic-name \
  --add-config retention.ms=604800000

查看 ZooKeeper 中存储的 Topic 元数据（旧版）

zookeeper-shell.sh localhost:2181 <<< "ls /brokers/topics"

从文件批量操作

# 批量创建主题
while read topic; do
  kafka-topics.sh --create --bootstrap-server localhost:9092 \
    --topic "$topic" --partitions 3 --replication-factor 2
done < topics.txt

查看未消费消息数（Lag）

kafka-consumer-groups.sh --describe \
  --bootstrap-server localhost:9092 \
  --group group-name \
  | awk '{if ($5 > 0) print $0}'

常用参数速查

参数

说明

--bootstrap-server

Kafka 集群地址（必填）

--topic

主题名称

--group

消费者组名称

--partitions

分区数

--replication-factor

副本因子

--from-beginning

从头消费

--execute

执行操作（需配合重置偏移量等）

--time -1

最新偏移量

--time -2

最早偏移量

--entity-type

配置类型（topics/brokers/clients）

--property

自定义属性（如 print.key=true）

# kafka

本文采用知识共享署名协议，转载请注明出处。

链接：/archives/kafka-chang-yong-ming-ling

评论