4. Kafka命令行操作：从创建Topic到管理Consumer Group

说实话，很多初学者一上来就扎进Kafka的API开发，结果连最基本的命令行操作都磕磕绊绊。我个人觉得，命令行才是理解Kafka运行机制最快的方式。你想想看，你亲手创建了一个Topic，看着消息从生产者流向消费者，这种直观感受比看十篇理论文章都管用。

这一节，我们就来把Kafka的命令行操作彻底搞明白。我会结合我在金融项目中的实际经验，把那些容易踩坑的地方都给你指出来。

4.1 创建Topic：你的第一个Kafka主题

创建Topic是Kafka操作的第一步。命令很简单，但参数背后的门道不少。

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

这里我要特别强调一下分区数的选择。我在项目中遇到过，有人图省事，所有Topic都用默认的1个分区。结果业务量一上来，消费者根本扛不住。为什么？因为Kafka的并行消费能力，本质上就是靠分区数来支撑的。

副本因子呢？在金融场景下，我建议至少设置为2。我曾经因为磁盘故障丢过数据，那教训太深刻了。不过也别设太多，3个副本就足够了，再多就是浪费资源。

4.2 查看Topic详情：知己知彼

创建完Topic，我们得确认一下它是不是真的按预期工作了。

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

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

输出结果会告诉你分区分布、Leader副本、ISR列表等信息。这里有个小技巧：重点关注ISR（In-Sync Replicas）的数量。如果ISR数量小于副本因子，说明有副本落后了，得赶紧排查。

4.3 生产消息：把数据送进Kafka

生产消息我们用kafka-console-producer.sh。这个工具特别适合做功能验证。

# 启动生产者，发送消息
kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic financial-transactions

# 然后输入消息内容，每行一条
> {"orderId": "TX001", "amount": 1500.00, "currency": "CNY"}
> {"orderId": "TX002", "amount": 3200.50, "currency": "USD"}

嗯，这里要注意：默认情况下，生产者发送的消息是没有key的，会以轮询方式分配到各个分区。如果你需要保证相同key的消息进入同一个分区，可以加上--property parse.key=true --property key.separator=:参数。

# 带key的生产消息
kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic financial-transactions \
  --property parse.key=true \
  --property key.separator=:

> user001:{"orderId": "TX003", "amount": 500.00}

为什么强调这个？在金融系统中，同一个用户的交易记录必须有序处理。如果你不指定key，消息可能分散到不同分区，消费顺序就乱了。我见过有人因为这个原因，对账系统跑出来全是乱账。

4.4 消费消息：把数据读出来

消费消息用kafka-console-consumer.sh。这个工具默认只消费新消息，不会读取历史数据。

# 启动消费者，从最新消息开始消费
kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic financial-transactions

# 从最开始的消息开始消费
kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic financial-transactions \
  --from-beginning

如果你想看消息的key、分区、偏移量等元数据，可以加上--property print.key=true --property print.partition=true --property print.offset=true。

kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic financial-transactions \
  --property print.key=true \
  --property print.partition=true \
  --property print.offset=true \
  --from-beginning

4.5 管理Consumer Group：掌控消费进度

Consumer Group是Kafka实现高可用的核心机制。同一个Group内的消费者共同消费一个Topic，每个分区只能被一个消费者消费。

# 列出所有Consumer Group
kafka-consumer-groups.sh --list --bootstrap-server localhost:9092

# 查看指定Group的详细信息
kafka-consumer-groups.sh --describe \
  --bootstrap-server localhost:9092 \
  --group my-consumer-group

输出结果会显示每个分区的当前偏移量（CURRENT-OFFSET）、日志结束偏移量（LOG-END-OFFSET）以及消费者落后多少（LAG）。这个LAG值特别重要，它直接反映了消费者的处理能力是否跟得上生产速度。

字段	含义	监控建议
CURRENT-OFFSET	当前消费者已消费到的位置	持续增长说明消费正常
LOG-END-OFFSET	生产者写入的最新位置	与CURRENT-OFFSET的差值就是LAG
LAG	消费者落后消息数	LAG持续增大说明消费能力不足

如果你需要重置消费偏移量，可以用--reset-offsets参数。这个操作很危险，我建议只在测试环境使用。

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

4.6 本章知识体系

下面这张图展示了Kafka命令行操作的核心逻辑，从创建Topic到消费消息，再到管理Consumer Group，每一步都有明确的输入输出。

这张图把整个流程串起来了。你从创建Topic开始，然后查看详情确认配置正确，接着生产消息验证写入，再消费消息确认读取，最后通过Consumer Group管理消费进度。每一步都有对应的命令和参数。

好了，Kafka命令行操作就讲到这里。这些命令你平时开发、测试、排查问题都会频繁用到。记住，多动手，多实践，遇到问题别慌，先用命令行看看Topic和Consumer Group的状态，往往就能找到线索。

---

公众号:蓝海资料掘金营，微信deep3321