4. Kafka 命令行操作：创建 Topic、查看描述、生产消息、消费消息、重置 Offset
Kafka 发行版内置了一套功能完备的命令行工具，位于 $KAFKA_HOME/bin/ 目录下。本章将围绕 Topic 生命周期管理与消息收发验证两个核心场景，逐一演示最常用的五个操作：创建 Topic、查看 Topic 详情、生产消息、消费消息以及重置消费者 Offset。所有命令均基于 Kafka 3.x 版本，并假设 ZooKeeper 或 KRaft 集群已正常运行。
4.1 创建 Topic
使用 kafka-topics.sh 脚本创建 Topic。创建时必须指定 Topic 名称、分区数（--partitions）和副本因子（--replication-factor）。副本因子不能超过集群中可用的 Broker 数量。
--bootstrap-server：指定 Kafka 集群地址，格式host:port。--partitions：分区数量，决定 Topic 的并行处理能力。--replication-factor：副本数量，控制数据冗余与高可用。--config：可选，设置 Topic 级别的覆盖参数（如retention.ms）。
# 创建名为 "score-events" 的 Topic，3 个分区，2 个副本
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
--create \
--topic score-events \
--partitions 3 \
--replication-factor 24.2 查看 Topic 描述
创建完成后，使用 --describe 参数查看 Topic 的详细元数据，包括分区分配、Leader 副本、ISR（In-Sync Replicas）列表等关键信息。
# 查看 "score-events" 的详细信息
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
--describe \
--topic score-events输出示例及字段含义：
| 字段 | 示例值 | 说明 |
|---|---|---|
| Topic | score-events | Topic 名称 |
| Partition | 0 | 分区编号 |
| Leader | 1 | 当前负责该分区读写请求的 Broker ID |
| Replicas | 1,2 | 该分区的所有副本所在的 Broker 列表 |
| Isr | 1,2 | 与 Leader 保持同步的副本列表（In-Sync Replicas） |
Isr 列表少于 Replicas 列表，说明有副本落后或宕机，需检查对应 Broker 的日志和网络状况。
4.3 生产消息
使用 kafka-console-producer.sh 向 Topic 发送消息。默认情况下，每条消息以换行符分隔，键与值之间用制表符（Tab）分隔。生产环境建议显式指定序列化器和分区器。
# 启动生产者，向 "score-events" 发送消息
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 \
--topic score-events \
--property "parse.key=true" \
--property "key.separator=:"输入示例（每行一条消息，键与值用 : 分隔）：
user001:{"score": 95, "subject": "math"}
user002:{"score": 88, "subject": "english"}
user003:{"score": 92, "subject": "science"}--property "parse.key=true" 可以指定消息键，Kafka 根据键的哈希值决定消息进入哪个分区，从而保证相同键的消息被路由到同一分区，维持局部顺序。
4.4 消费消息
使用 kafka-console-consumer.sh 消费消息。默认情况下，消费者从最新偏移量开始读取，只消费启动后到达的新消息。若要查看历史消息，需指定 --from-beginning。
# 启动消费者，从最早偏移量开始消费
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
--topic score-events \
--from-beginning \
--property "print.key=true" \
--property "key.separator=:"输出示例：
user001:{"score": 95, "subject": "math"}
user002:{"score": 88, "subject": "english"}
user003:{"score": 92, "subject": "science"}--group）已提交过偏移量，--from-beginning 不会生效，消费者会从上次提交的偏移量继续消费。要重新消费所有消息，需要先重置偏移量或使用新的消费者组。
4.5 重置 Offset
当需要重新处理历史数据或修复消费逻辑时，可以使用 kafka-consumer-groups.sh 重置消费者组的偏移量。重置操作支持多种策略：--to-earliest（最早）、--to-latest（最新）、--to-offset（指定偏移量）、--shift-by（相对偏移）等。
# 将消费者组 "score-group" 的所有分区偏移量重置到最早位置
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
--group score-group \
--topic score-events \
--reset-offsets \
--to-earliest \
--execute重置策略对比：
| 策略参数 | 说明 | 适用场景 |
|---|---|---|
--to-earliest |
重置到分区的最早偏移量 | 重新处理全部历史数据 |
--to-latest |
重置到分区的最新偏移量 | 跳过历史数据，只消费新消息 |
--to-offset |
重置到指定的绝对偏移量 | 从特定位置开始消费 |
--shift-by |
在当前偏移量基础上向前或向后移动 N 条 | 跳过或重复消费最近 N 条消息 |
--to-datetime |
重置到指定时间戳对应的偏移量 | 按时间回溯消费 |
- 重置前必须确保消费者组处于 非运行状态，否则操作会被拒绝。
- 使用
--dry-run参数可以预览重置结果而不实际执行，建议先预览再执行。 - 重置操作只影响指定消费者组的偏移量，不会影响 Topic 中的消息数据。
4.6 常用命令速查表
| 操作 | 命令脚本 | 核心参数 |
|---|---|---|
| 创建 Topic | kafka-topics.sh |
--create --topic --partitions --replication-factor |
| 查看 Topic 列表 | kafka-topics.sh |
--list |
| 查看 Topic 详情 | kafka-topics.sh |
--describe --topic |
| 生产消息 | kafka-console-producer.sh |
--topic --property "parse.key=true" |
| 消费消息 | kafka-console-consumer.sh |
--topic --group --from-beginning |
| 重置 Offset | kafka-consumer-groups.sh |
--group --reset-offsets --to-earliest --execute |