4. Kafka 命令行操作：Topic 管理、生产者、消费者与消费组

说实话，很多初学者一上来就扎进 Kafka 的 API 或者源码里，反而把最基础、最常用的命令行工具给忽略了。我个人觉得，命令行是理解 Kafka 运行机制最快的方式。你敲一条命令，看它怎么报错，怎么返回，比读十页文档都管用。

这一章，我们就来把 Kafka 的命令行工具彻底捋一遍。包括 Topic 怎么创建、怎么修改，生产者怎么发消息，消费者怎么收消息，还有消费组怎么管理。这些都是你每天都会用到的操作。

4.1 Topic 管理命令

Topic 是 Kafka 里最基础的数据容器。你想想看，没有 Topic，生产者和消费者根本没法对接。所以 Topic 管理命令，是你必须熟练掌握的。

4.1.1 创建 Topic

创建 Topic 最常用的命令是 kafka-topics.sh。我记得刚带团队的时候，有个同事创建 Topic 忘了指定分区数，结果默认只有一个分区，后面业务量上来直接扛不住。所以创建的时候，分区数和副本数一定要想清楚。

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

4.1.2 查看 Topic 列表与详情

查看所有 Topic 很简单。但真正有用的是看某个 Topic 的详细信息，比如分区分布、Leader 是谁、ISR 列表。这些信息在排查问题时特别关键。

# 查看所有 Topic
bin/kafka-topics.sh --list --bootstrap-server localhost:9092

# 查看指定 Topic 的详细信息
bin/kafka-topics.sh --describe \
  --bootstrap-server localhost:9092 \
  --topic order-events

输出结果里，你会看到每个分区的 Leader、Replicas、ISR。如果某个分区的 ISR 列表比 Replicas 少，说明有副本同步滞后了。嗯，这时候就要检查网络或者磁盘了。

4.1.3 修改 Topic

业务增长后，分区数不够是常有的事。Kafka 允许你增加分区数，但注意——不能减少。为什么？因为减少分区会导致数据重新分配，Kafka 的设计哲学里不允许这种操作。

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

4.1.4 删除 Topic

删除 Topic 要谨慎。我曾经在生产环境上误删过一个 Topic，还好有备份。所以建议你在删除前，先确认一下这个 Topic 是不是真的没用了。

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

删除操作只是标记删除，实际清理由后台线程异步执行。如果你想立即生效，可以手动删除 Zookeeper 或 Kafka 存储目录下的数据，但一般不推荐这么做。

4.2 生产者命令行

生产者的命令行工具是 kafka-console-producer.sh。它主要用于测试和调试。你可以在终端里直接输入消息，看看能不能正常发送到 Kafka。

4.2.1 基本使用

# 启动生产者，发送消息到 order-events Topic
bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic order-events

启动后，你就可以在终端输入消息了。每行一条消息，按 Ctrl+C 退出。

4.2.2 带 key 的消息

如果你需要根据 key 将消息路由到特定分区，可以加上 --property parse.key=true 参数。消息格式是 key:value。

bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic order-events \
  --property parse.key=true \
  --property key.separator=:

输入示例：

user123:{"order_id": 1001, "amount": 299}
user456:{"order_id": 1002, "amount": 159}

这样，相同 user 的消息就会进入同一个分区，保证顺序性。

4.2.3 指定分区发送

有时候你想把消息直接发到某个分区，可以用 --partition 参数。不过说实话，生产环境很少这么干，一般都是让 Kafka 自己决定。

bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic order-events \
  --partition 0

4.3 消费者命令行

消费者命令行工具是 kafka-console-consumer.sh。它比生产者复杂一些，因为涉及到消费组、偏移量、分区分配等概念。

4.3.1 基本消费

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

默认情况下，消费者只消费启动后到达的新消息。如果你想看历史消息，需要加 --from-beginning 参数。

4.3.2 从头消费

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

这个命令会从 Topic 最早的消息开始消费。注意，如果 Topic 数据量很大，可能会输出很久。我一般用它来验证数据是否完整写入。

4.3.3 指定消费组

消费组是 Kafka 实现负载均衡和容错的关键。你可以用 --group 参数指定消费组名。

bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic order-events \
  --group my-consumer-group \
  --from-beginning

同一个消费组内的消费者，会分摊 Topic 的分区。比如 3 个分区，2 个消费者，一个消费者消费 2 个分区，另一个消费 1 个分区。

4.3.4 指定分区消费

如果你想只看某个分区的数据，可以用 --partition 参数。这在调试分区数据不一致时特别有用。

bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic order-events \
  --partition 0 \
  --offset 100

这个命令会从分区 0 的偏移量 100 开始消费。你可以指定任意偏移量，非常灵活。

4.4 消费组管理

消费组管理是运维中最重要的部分之一。你经常需要查看消费组的消费进度、偏移量、以及是否有消费者掉线。

4.4.1 查看消费组列表

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

这个命令会列出所有活跃的消费组。如果某个消费组不在列表里，说明它已经关闭或者从未启动过。

4.4.2 查看消费组详情

bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --group my-consumer-group \
  --describe

输出结果包含以下信息：

字段	说明
TOPIC	消费的 Topic 名称
PARTITION	分区编号
CURRENT-OFFSET	当前消费到的偏移量
LOG-END-OFFSET	分区最新的偏移量
LAG	消费滞后量（LOG-END-OFFSET - CURRENT-OFFSET）
CONSUMER-ID	消费者实例的唯一标识
HOST	消费者所在主机
CLIENT-ID	客户端 ID

4.4.3 重置消费组偏移量

有时候你需要让消费者重新消费历史数据，或者跳过某些消息。这时候就要重置偏移量。

# 将消费组偏移量重置到最早位置
bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --group my-consumer-group \
  --topic order-events \
  --reset-offsets \
  --to-earliest \
  --execute

# 将偏移量重置到指定时间戳
bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --group my-consumer-group \
  --topic order-events \
  --reset-offsets \
  --to-datetime 2024-01-01T00:00:00.000 \
  --execute

4.4.4 删除消费组

如果某个消费组不再使用，可以删除它。但注意，删除消费组不会影响已经提交的偏移量，只是从 Kafka 的元数据中移除。

bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --group my-consumer-group \
  --delete

4.5 常用命令速查表

为了方便你日常使用，我把本章涉及的所有命令整理成了一张表。你可以把它收藏起来，用的时候直接查。

操作类型	命令	说明
Topic 创建	kafka-topics.sh --create	指定分区数和副本数
Topic 列表	kafka-topics.sh --list	查看所有 Topic
Topic 详情	kafka-topics.sh --describe	查看分区、副本、ISR 信息
Topic 修改	kafka-topics.sh --alter	只能增加分区数
Topic 删除	kafka-topics.sh --delete	标记删除，异步清理
生产者	kafka-console-producer.sh	支持 key-value 和指定分区
消费者	kafka-console-consumer.sh	支持从头消费、指定分区、消费组
消费组列表	kafka-consumer-groups.sh --list	查看所有活跃消费组
消费组详情	kafka-consumer-groups.sh --describe	查看偏移量和 LAG
重置偏移量	kafka-consumer-groups.sh --reset-offsets	支持最早、最新、指定时间戳

好了，这一章的内容就到这里。命令行工具虽然看起来简单，但用好了能解决很多实际问题。你可以在自己的测试环境里多敲几遍，熟悉一下每个参数的效果。遇到问题的时候，这些命令就是你最直接的调试工具。