4. Kafka 命令行操作：创建 Topic、生产消息、消费消息、查看消费组状态

Kafka 命令行工具是运维与开发人员最直接的交互界面。本章聚焦四个核心操作：创建 Topic、生产消息、消费消息以及查看消费组状态。掌握这些命令，你就能在 Kafka 集群上完成最基本的消息流转闭环。

4.1 创建 Topic

Topic 是 Kafka 消息的逻辑分类。创建 Topic 时需要指定分区数（partitions）和副本因子（replication-factor）。

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

参数说明：

验证 Topic 是否创建成功：

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

输出应包含 order-events。查看 Topic 详情：

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

输出示例：

Topic: order-events  PartitionCount: 3  ReplicationFactor: 2
  Topic: order-events  Partition: 0  Leader: 1  Replicas: 1,2  Isr: 1,2
  Topic: order-events  Partition: 1  Leader: 2  Replicas: 2,3  Isr: 2,3
  Topic: order-events  Partition: 2  Leader: 3  Replicas: 3,1  Isr: 3,1

其中 Leader 表示当前负责读写的主副本，Replicas 是所有副本列表，Isr（In-Sync Replicas）是与 Leader 保持同步的副本。

4.2 生产消息

使用 kafka-console-producer.sh 向 Topic 发送消息。默认每条消息以换行符分隔。

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

输入消息内容（每行一条）：

{"orderId": "1001", "userId": "u001", "amount": 299.00}
{"orderId": "1002", "userId": "u002", "amount": 159.00}
{"orderId": "1003", "userId": "u003", "amount": 499.00}

按 Ctrl + D 结束输入。如果需要指定 key，使用 --property parse.key=true --property key.separator=:：

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

输入格式：key:value

user_1001:{"orderId": "1001", "amount": 299.00}
user_1002:{"orderId": "1002", "amount": 159.00}

4.3 消费消息

使用 kafka-console-consumer.sh 消费消息。需要指定消费组（--group）和起始偏移策略。

# 启动消费者，从最新消息开始消费
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic order-events \
  --group order-group-1 \
  --from-beginning

参数说明：

带格式输出的消费命令：

kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic order-events \
  --group order-group-1 \
  --from-beginning \
  --property print.key=true \
  --property print.partition=true \
  --property print.timestamp=true

输出示例：

CreateTime:1712567890123  Partition:0  user_1001  {"orderId": "1001", "amount": 299.00}
CreateTime:1712567890456  Partition:1  user_1002  {"orderId": "1002", "amount": 159.00}
CreateTime:1712567890789  Partition:2  user_1003  {"orderId": "1003", "amount": 499.00}

4.4 查看消费组状态

消费组状态是监控消费者健康度的核心指标。使用 kafka-consumer-groups.sh 查看。

# 列出所有消费组
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

输出：

order-group-1
order-group-2

# 查看指定消费组的详细信息
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-group-1 \
  --describe

输出示例：

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG   CONSUMER-ID     HOST            CLIENT-ID
order-group-1   order-events    0          15              20              5     consumer-1      /192.168.1.10   consumer-1
order-group-1   order-events    1          12              20              8     consumer-1      /192.168.1.10   consumer-1
order-group-1   order-events    2          18              20              2     consumer-2      /192.168.1.11   consumer-2

关键指标解读：

• CURRENT-OFFSET：当前消费组已消费到的偏移量
• LOG-END-OFFSET：分区最新消息的偏移量
• LAG：未消费的消息数量（LOG-END-OFFSET - CURRENT-OFFSET）
• CONSUMER-ID：消费者实例的唯一标识

4.5 常用操作速查表