4. Kafka 命令行工具：Topic 创建、消息生产与消费、偏移量查看

Kafka 发行版内置了一套功能完备的命令行工具，位于 $KAFKA_HOME/bin/ 目录下。这些工具是开发阶段验证集群连通性、调试消息格式、排查消费滞后最直接的手段。本章聚焦四个核心操作：创建 Topic、生产消息、消费消息 以及 查看偏移量，并给出高频参数对照与常见避坑指南。

4.1 创建 Topic

使用 kafka-topics.sh 脚本创建 Topic。以下命令创建一个名为 order-events、分区数为 3、副本因子为 2 的 Topic：

# 创建 Topic（ZooKeeper 模式）
bin/kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --replication-factor 2 \
  --partitions 3 \
  --topic order-events

# 创建 Topic（KRaft 模式，参数一致）
bin/kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --replication-factor 2 \
  --partitions 3 \
  --topic order-events

创建成功后，终端会输出 Created topic order-events.。常用参数说明如下：

4.2 生产消息

使用 kafka-console-producer.sh 从标准输入发送消息。默认以 Enter 键分隔消息，每行视为一条记录。

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

# 输入消息内容（每行一条）
> order-1001:paid
> order-1002:shipped
> order-1003:delivered

如需指定 Key 或使用自定义分区器，可附加参数：

# 带 Key 的生产示例
bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic order-events \
  --property parse.key=true \
  --property key.separator=:

# 输入：key:value
> 1001:{"orderId":"1001","status":"paid"}

4.3 消费消息

使用 kafka-console-consumer.sh 从指定 Topic 拉取消息。默认从最新偏移量开始消费，仅显示新到达的消息。

# 启动消费者，实时消费 order-events
bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic order-events \
  --from-beginning

--from-beginning 参数使消费者从 Topic 最早可用偏移量开始读取，适合验证历史数据。常用消费参数：

# 带分组、打印 Key/分区/偏移量的消费命令
bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic order-events \
  --group order-group-v1 \
  --property print.key=true \
  --property print.partition=true \
  --property print.offset=true \
  --from-beginning

4.4 查看偏移量

偏移量（Offset）是消费者在分区内的读取位置。使用 kafka-consumer-groups.sh 查看消费者组的偏移量状态与滞后（Lag）。

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

# 查看指定组的偏移量详情
bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --group order-group-v1 \
  --describe

--describe 输出包含以下关键列：

4.5 核心操作流程总览

下图展示了从 Topic 创建到偏移量查看的完整命令行操作链路，以及各工具之间的数据流向关系：

4.6 常见问题与排查

• Topic 已存在错误：重复创建同名 Topic 会提示 Topic 'xxx' already exists.，可使用 --list 确认现有 Topic 列表。
• 消费者无消息输出：检查是否遗漏 --from-beginning；确认生产者是否已成功发送消息；使用 --describe 查看消费者组偏移量是否等于 LOG-END-OFFSET。
• 偏移量重置不生效：确保消费者组处于 inactive 状态（所有消费者实例已停止），否则重置操作会被拒绝。
• 连接超时：验证 --bootstrap-server 地址与端口是否正确，防火墙是否放行 9092 端口。