📚 课程目录第4章 Kafka命令行操作

4、Kafka命令行操作：Topic管理命令、生产者命令行、消费者命令行、查看消费组状态

说实话，很多初学者一上来就扎进代码里，反而忽略了最基础的东西——命令行。我个人觉得，Kafka的命令行工具是你理解整个系统运作最好的入口。你想想看，你敲一行命令，就能看到Topic怎么创建、消息怎么发、消费者怎么拉数据，这比看十页文档都管用。

这一章，我们就来亲手操作一下。我会带着你把Kafka最常用的几个命令行工具过一遍。嗯，这里要注意，我假设你已经把Kafka环境搭好了，并且bin/目录在你的PATH里。如果还没搭好，先回去把上一章的实操补上。

4.1 Topic管理命令：一切的基础

Topic是Kafka里最核心的抽象。说白了，它就是一条数据流的名字。你往里面写数据，别人从里面读数据。管理Topic，我们主要用kafka-topics.sh这个脚本。

4.1.1 创建Topic

创建一个Topic，你需要指定名字、分区数和副本因子。我刚开始用的时候，经常忘记指定副本因子，结果生产环境里Topic只有一个副本， broker一挂就全完了。后来我养成了习惯，创建Topic时必带--replication-factor。

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

这里有个细节：--bootstrap-server 是连接Kafka集群的入口。早期版本用 --zookeeper，现在新版本都推荐用 --bootstrap-server。我个人习惯用这个，因为更直接，也避免了ZooKeeper的依赖。

小提示： 分区数不是越多越好。分区太多会导致文件句柄开销大、Leader选举变慢。我一般建议分区数不超过broker数量的10倍。比如你有3台broker，分区数控制在30以内比较稳妥。

4.1.2 查看Topic列表

创建完了，怎么确认它存在？很简单，用 --list 参数。

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

这个命令会返回所有Topic的名字。如果你看到刚才创建的 test-topic 在里面，那就说明创建成功了。

4.1.3 查看Topic详情

光看名字不够，你还得知道它的分区分布、副本情况。用 --describe 参数。

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

输出大概长这样：

Topic: test-topic	TopicId: abc123	PartitionCount: 3	ReplicationFactor: 2	Configs:
	Topic: test-topic	Partition: 0	Leader: 1	Replicas: 1,2	Isr: 1,2
	Topic: test-topic	Partition: 1	Leader: 2	Replicas: 2,3	Isr: 2,3
	Topic: test-topic	Partition: 2	Leader: 3	Replicas: 3,1	Isr: 3,1

我来解释一下这些字段：

字段	含义
Partition	分区编号，从0开始
Leader	当前负责读写该分区的broker ID
Replicas	该分区的所有副本所在的broker列表
Isr	In-Sync Replicas，与Leader保持同步的副本列表

你看，Partition 0的Leader是broker 1，Replicas是1和2，Isr也是1和2。这说明broker 1和2都有这个分区的数据，而且都跟Leader保持同步。如果Isr里的副本数少于Replicas，那就说明有副本落后了，需要排查。

注意： 如果Isr列表比Replicas列表短，说明有副本同步延迟或宕机。我曾经遇到过因为网络抖动导致Isr频繁收缩，结果生产环境写入延迟飙升。这时候要检查网络和磁盘IO。

4.1.4 修改Topic

有时候你需要调整分区数。注意，分区数只能增加，不能减少。为什么？因为减少分区会导致数据重新分配，逻辑上非常复杂，Kafka干脆不支持。

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

嗯，这里要提醒你：增加分区后，原有的数据不会自动重新分布。新分区是空的，只有新写入的数据才会进入新分区。如果你想让旧数据也均匀分布，需要手动做数据重分配，那又是另一个话题了。

4.1.5 删除Topic

删除Topic要谨慎。一旦删除，所有数据都没了。

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

不过，默认情况下Kafka不会立即删除Topic，而是把它标记为 marked for deletion。真正删除需要等 delete.topic.enable=true 配置生效。我建议你在测试环境里把这个开关打开，方便清理，但生产环境一定要慎重。

4.2 生产者命令行：往Topic里写数据

Topic建好了，怎么往里写数据？用 kafka-console-producer.sh。这个工具可以从标准输入读取数据，然后发送到Kafka。

bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test-topic

执行后，你会进入一个交互模式。随便输入一行文字，按回车，消息就发出去了。按 Ctrl+C 退出。

如果你想指定key，可以用 --property parse.key=true --property key.separator=: 参数。这样输入 key1:value1 时，Kafka会把 key1 当作key，value1 当作value。

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

为什么要指定key？因为key决定了消息被发送到哪个分区。如果key相同，消息会进入同一个分区，保证顺序。我在做订单系统时，就用订单ID作为key，确保同一个订单的所有状态变更消息按顺序被消费。

  核心要点： 生产者的 acks 参数控制消息的可靠性。默认是 acks=1，只要Leader确认就返回。如果要求更高，可以设 acks=all，等所有ISR副本都确认才返回。但延迟会更高。你需要根据业务场景权衡。

4.3 消费者命令行：从Topic里读数据

有写就有读。用 kafka-console-consumer.sh 来消费消息。

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

--from-beginning 表示从Topic的最早消息开始消费。如果不加这个参数，默认只消费新消息。这个参数在调试时特别有用，你可以看到Topic里所有的历史数据。

如果你只想看最新的消息，不加 --from-beginning 就行。启动后，它会一直等待新消息。你可以在另一个终端里用生产者发一条消息，这边马上就能看到。

消费者还有一个重要的概念：消费组。每个消费者都属于一个消费组，用 --group 参数指定。

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic --group my-group

同一个消费组里的消费者，会共同分担Topic里的分区。比如Topic有3个分区，组里有2个消费者，那么一个消费者消费2个分区，另一个消费1个分区。如果组里有3个消费者，那就一人一个分区。如果消费者超过分区数，多余的消费者会闲置。

小技巧： 如果你想测试消费组的分区分配情况，可以启动多个消费者终端，都指定同一个group名。然后观察每个终端收到的消息，你会发现它们各自消费不同的分区。

4.4 查看消费组状态：监控消费进度

生产环境里，你经常需要知道消费者有没有落后。用 kafka-consumer-groups.sh 可以查看消费组的详细信息。

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-group --describe

输出示例：

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
my-group        test-topic      0          15              20              5               consumer-1...   /192.168.1.2    consumer-1
my-group        test-topic      1          10              10              0               consumer-1...   /192.168.1.2    consumer-1
my-group        test-topic      2          8               12              4               consumer-2...   /192.168.1.3    consumer-2

关键字段解释：

字段	含义
CURRENT-OFFSET	消费者当前已经消费到的位置
LOG-END-OFFSET	Topic分区里最新的消息位置
LAG	滞后量，等于 LOG-END-OFFSET 减去 CURRENT-OFFSET

你看，Partition 0的LAG是5，说明还有5条消息没被消费。如果LAG持续增长，说明消费者处理速度跟不上生产速度。我曾经遇到过因为消费者代码里有个慢SQL，导致LAG涨到几十万，最后不得不重启消费者并重置偏移量才恢复。

避坑指南： 如果LAG一直不为0，而且还在增长，不要盲目重启消费者。先检查消费者的处理逻辑，看是不是有阻塞操作。我曾经因为一个外部API调用超时，导致整个消费线程卡住，LAG飙升。后来加了超时和重试机制才解决。

4.5 本章知识体系总览

为了让你更直观地理解这些命令之间的关系，我画了一张图。你可以看到，Topic管理命令负责创建和配置数据通道，生产者命令负责写入，消费者命令负责读取，而消费组状态命令则负责监控整个消费链路。

这张图把四个命令组的关系梳理得很清楚。你从上往下看：Topic管理命令先定义好数据通道，生产者命令往里写数据，消费者命令从里面读，消费组状态命令则时刻盯着消费进度。掌握了这四个工具，你就能在命令行里完成Kafka大部分日常操作了。