Kafka常用命令有哪些？

Question

Kafka常用命令有哪些？在实际运维中，如何通过命令行快速查看Topic状态、创建Topic、查看消费者组消费进度是常见问题。例如，如何使用`kafka-topics.sh`命令查看某个Topic的分区和副本信息？如何用`kafka-consumer-groups.sh`检查消费者组的滞后（Lag）情况？此外，生产消息和消费消息时，`kafka-console-producer.sh`与`kafka-console-consumer.sh`的基本用法也常被问及。掌握这些基础命令对日常开发、调试和运维至关重要。请结合实际场景说明常用命令及其关键参数。

Answer 1

Kafka常用命令详解与运维实践指南

1. Kafka命令行工具概述

Apache Kafka 提供了一套丰富的命令行工具（CLI），主要用于集群管理、Topic操作、消费者组监控以及消息的生产与消费。这些脚本通常位于 Kafka 安装目录的 bin/ 路径下，适用于 Linux/Unix 环境。对于拥有5年以上经验的IT从业者而言，熟练掌握这些命令不仅有助于日常开发调试，更是故障排查和性能优化的重要手段。

2. Topic管理：使用 kafka-topics.sh

Topic 是 Kafka 的核心概念之一，所有消息都归属于某个 Topic。以下是常见的操作命令：

• 创建 Topic：
  bin/kafka-topics.sh --create --topic user-logs --partitions 6 --replication-factor 3 --bootstrap-server localhost:9092
  关键参数说明：
  • --partitions：指定分区数，影响并行度
  • --replication-factor：副本数量，决定容错能力
  • --bootstrap-server：连接的Broker地址
• 查看所有 Topic 列表：
  bin/kafka-topics.sh --list --bootstrap-server localhost:9092
• 查看特定 Topic 的详细信息（含分区与副本）：
  bin/kafka-topics.sh --describe --topic user-logs --bootstrap-server localhost:9092
  输出示例片段：

  Topic: user-logs	PartitionCount: 6	ReplicationFactor: 3	Configs:
  	Topic: user-logs	Partition: 0	Leader: 1	Replicas: 1,2,3	Isr: 1,2,3
  	Topic: user-logs	Partition: 1	Leader: 2	Replicas: 2,3,1	Isr: 2,3,1
      

  其中，Leader 表示当前分区主节点，Replicas 为副本列表，Isr（In-Sync Replicas）表示同步中的副本。

3. 消费者组监控：kafka-consumer-groups.sh

在实际运维中，消费者滞后（Lag）是判断系统健康状态的关键指标。以下为关键命令：

操作类型	命令示例	说明
列出所有消费者组	bin/kafka-consumer-groups.sh --list --bootstrap-server localhost:9092	用于发现活跃或残留的消费者组
查看消费者组详情	bin/kafka-consumer-groups.sh --describe --group log-reader-group --bootstrap-server localhost:9092	显示每个分区的消费偏移量、LAG等

输出中关键字段解释：

• TOPIC：所属Topic名称
• PARTITION：分区编号
• CURRENT-OFFSET：当前已消费的最新偏移量
• LOG-END-OFFSET：日志末尾偏移量（即最新消息位置）
• LAG：两者之差，若持续增长则可能存在问题

4. 实时消息生产与消费测试

在开发联调或问题复现时，常通过控制台工具快速验证数据流。

生产消息：kafka-console-producer.sh

bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test-topic
执行后可直接输入文本内容，每行作为一条消息发送。
常用参数扩展：

• --producer-property key.serializer=org.apache.kafka.common.serialization.StringSerializer
• --property parse.key=true --property key.separator=: 支持键值对输入

消费消息：kafka-console-consumer.sh

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic --from-beginning
参数说明：

• --from-beginning：从最早消息开始读取
• --group test-consumer：指定消费者组名，便于后续监控
• --timeout-ms 10000：设置超时时间，避免无限等待

5. 高级运维场景分析流程图

当出现消费延迟时，可通过如下流程进行诊断：

graph TD
    A[消费者Lag持续上升] --> B{是否新启动消费者?}
    B -->|是| C[等待Rebalance完成]
    B -->|否| D[检查消费者是否存活]
    D --> E[查看JVM/GC状态]
    E --> F[检查网络延迟与Broker负载]
    F --> G[分析Topic分区分配是否均衡]
    G --> H[调整分区数或消费者实例数量]

6. 生产环境最佳实践建议

• 避免手动创建无配置的Topic，应启用 auto.create.topics.enable=false
• 定期巡检消费者组Lag，结合Prometheus + Grafana实现可视化告警
• 使用 --bootstrap-server 替代旧版 --zookeeper 参数（Kafka 2.5+ 推荐）
• 在多租户环境中，通过ACL控制Topic访问权限
• 对重要Topic设置最小ISR限制（min.insync.replicas）以保障数据一致性
• 利用 kafka-dump-log.sh 分析底层Segment文件（高级故障排查）
• 批量操作时可编写Shell脚本封装常用命令组合
• 注意控制台消费者不适用于压测，建议使用 kafka-producer-perf-test.sh 进行性能测试
• 关注Broker端日志（server.log）与控制器状态（kafka-controller.log）
• 升级Kafka版本前，务必验证CLI命令兼容性