3. Kafka 基础运维：命令行工具与 Topic 管理

Kafka 的命令行工具，说白了就是运维人员的瑞士军刀。我刚带团队那会儿，发现不少新人喜欢装各种 GUI 工具，结果集群一出问题，最先挂掉的就是那些图形界面。所以我一直强调：命令行才是保命的本事。

这一章，咱们就聚焦三个最核心的工具：kafka-topics.sh、kafka-console-producer.sh 和 kafka-console-consumer.sh。Topic 管理的所有操作，几乎都靠它们完成。

3.1 kafka-topics.sh：Topic 的增删改查

这个工具，我一天至少要用十几次。它负责 Topic 的整个生命周期管理。说白了，Topic 就是 Kafka 里的「表」，分区就是「分片」。

3.1.1 创建 Topic

创建一个 Topic，你需要指定几个关键参数：分区数、副本因子、以及可选的配置覆盖。

# 创建一个名为 "order-events" 的 Topic，6 个分区，3 个副本
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --create \
  --topic order-events \
  --partitions 6 \
  --replication-factor 3

3.1.2 查看 Topic 列表与详情

查看所有 Topic：

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

查看某个 Topic 的详细信息：

bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic order-events

输出结果里，我最关注的是 Leader、Replicas 和 Isr 这三列。Leader 分布是否均匀？ISR 有没有缺失？这些一眼就能看出集群的健康状况。

3.1.3 修改 Topic

Kafka 允许你动态修改 Topic 的配置和分区数。但注意：分区数只能增加，不能减少。

# 增加分区数（从 6 到 12）
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --alter \
  --topic order-events \
  --partitions 12

# 修改 Topic 级别的配置
bin/kafka-configs.sh --bootstrap-server localhost:9092 \
  --alter \
  --entity-type topics \
  --entity-name order-events \
  --add-config retention.ms=86400000

3.1.4 删除 Topic

bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic order-events

3.2 kafka-console-producer.sh：快速生产数据

这个工具主要用于测试和调试。你想想看，写完一个 Topic，总得发几条消息验证一下吧？

3.2.1 基本用法

# 发送消息到指定 Topic
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test-topic

# 输入消息内容，按 Ctrl+D 结束
>hello world
>this is a test message

3.2.2 常用参数

参数	说明	我的建议
--property parse.key=true	启用 Key-Value 模式	测试分区策略时必用
--property key.separator=:	指定 Key 和 Value 的分隔符	默认是 Tab，我习惯用冒号
--producer-property acks=all	设置生产者确认级别	测试可靠性时用 all

# 带 Key 的生产消息
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 \
  --topic test-topic \
  --property parse.key=true \
  --property key.separator=:
>user001:{"name":"Alice","action":"login"}

3.3 kafka-console-consumer.sh：验证消费数据

生产了数据，总得看看能不能消费到。这个工具就是干这个的。

3.3.1 基本用法

# 从最新偏移量开始消费
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic

# 从最开始消费
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic test-topic \
  --from-beginning

3.3.2 常用参数

参数	说明	使用场景
--group	指定消费组	测试消费组偏移量
--property print.key=true	打印 Key	验证 Key 分区逻辑
--property print.partition=true	打印分区号	确认数据分布
--max-messages	消费 N 条后退出	避免一直挂在那里

# 带 Key 和分区信息的消费
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic test-topic \
  --from-beginning \
  --property print.key=true \
  --property print.partition=true \
  --max-messages 10

3.4 Topic 管理操作实战

光说不练假把式。咱们来一个完整的操作流程。

3.4.1 完整操作流程

1. 创建 Topic：指定分区和副本
2. 查看详情：确认 Leader 和 ISR 分布
3. 生产数据：用 console-producer 写入测试消息
4. 消费验证：用 console-consumer 从开始消费
5. 修改配置：调整 retention 或分区数
6. 再次验证：确认修改生效

# 步骤 1：创建
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --create --topic ops-test --partitions 3 --replication-factor 2

# 步骤 2：查看
bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic ops-test

# 步骤 3：生产
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 \
  --topic ops-test \
  --property parse.key=true \
  --property key.separator=:
>key1:value1
>key2:value2
>key3:value3

# 步骤 4：消费
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic ops-test \
  --from-beginning \
  --property print.key=true \
  --property print.partition=true \
  --max-messages 3

3.4.2 常见问题排查

现象	可能原因	排查命令
创建 Topic 失败	Broker 数量不足、副本因子过大	kafka-topics.sh --describe --topic xxx
生产消息超时	Leader 不可用、网络问题	kafka-topics.sh --describe --topic xxx 看 Leader
消费不到数据	偏移量提交错误、消费组冲突	kafka-consumer-groups.sh --describe --group xxx
ISR 列表不全	副本同步延迟、Broker 故障	kafka-topics.sh --describe --topic xxx 看 Isr

嗯，命令行工具就讲到这里。记住：工具是死的，思路是活的。理解了 Topic 的结构和数据的流向，这些命令就是你的得力助手。