Kafka运维宝典 （四）- Kafka 常用命令介绍

Kafka 提供了一套全面的命令行工具，支持主题管理、消费者组管理、生产与消费数据等操作。熟练掌握这些命令可以帮助你更高效地管理 Kafka 集群，同时快速解决问题，提高系统运行效率。

1. Kafka Broker 管理相关命令

1.1 查看 Kafka Broker 信息

用途：查看集群中所有 Kafka Broker 的 broker.id。

命令：

bin/zookeeper-shell.sh <zookeeper_host>:2181 <<< "ls /brokers/ids"

输出：

[0, 1, 2]

表示当前 Kafka 集群中有 3 个 Broker，broker.id 分别为 0, 1, 2。

1.2 查看 Kafka Broker API 版本

用途：检查 Kafka Broker 支持的 API 版本和兼容性。

命令：

bin/kafka-broker-api-versions.sh --bootstrap-server <broker_host>:9092

输出：

Supported APIs:API Name        Max Version   Min VersionFetch           15            0Produce         15            0Metadata        12            0

2. Kafka 主题管理相关命令

2.1 查看所有主题

用途：查看 Kafka 集群中的所有主题。

命令：

bin/kafka-topics.sh --bootstrap-server <broker_host>:9092 --list

输出：

topic1
topic2
topic3

2.2 创建主题

用途：在 Kafka 集群中创建新的主题。

命令：

bin/kafka-topics.sh --bootstrap-server <broker_host>:9092 --create --topic my_topic --partitions 3 --replication-factor 2

说明：

• --partitions 3：创建 3 个分区。
• --replication-factor 2：每个分区有 2 个副本。

输出：

Created topic my_topic.

2.3 查看主题详情

用途：查看某个主题的详细信息，包括分区、副本和 ISR（同步副本）。

命令：

bin/kafka-topics.sh --bootstrap-server <broker_host>:9092 --describe --topic my_topic

输出示例：

    Topic: my_topic    PartitionCount: 3    ReplicationFactor: 2    Configs:Topic: my_topic    Partition: 0    Leader: 1    Replicas: 1,2    Isr: 1,2Topic: my_topic    Partition: 1    Leader: 2    Replicas: 2,3    Isr: 2,3Topic: my_topic    Partition: 2    Leader: 3    Replicas: 3,1    Isr: 3,1

2.4 删除主题

用途：删除指定的主题（需启用 delete.topic.enable=true 配置）。

命令：

bin/kafka-topics.sh --bootstrap-server <broker_host>:9092 --delete --topic my_topic

输出：

Topic my_topic is marked for deletion.

2.5 修改主题分区数

用途：为现有主题增加分区数（只能增加，不能减少）。

命令：

bin/kafka-topics.sh --bootstrap-server <broker_host>:9092 --alter --topic my_topic --partitions 5

输出：

Updated topic my_topic.

3. 生产者命令

3.1 向主题发送消息

用途：通过生产者工具向 Kafka 主题发送消息。

命令：

bin/kafka-console-producer.sh --broker-list <broker_host>:9092 --topic my_topic

操作：
输入消息后按回车即可发送。例如：

> hello world
> this is a test message

4. 消费者命令

4.1 从主题消费消息

用途：使用消费者工具从 Kafka 主题消费消息。

命令：

bin/kafka-console-consumer.sh --bootstrap-server <broker_host>:9092 --topic my_topic --from-beginning

说明：

• --from-beginning：从主题的起始位置消费消息。

输出：

hello world
this is a test message

5. 消费者组命令

5.1 查看所有消费者组

用途：查看 Kafka 集群中的所有消费者组。

命令：

bin/kafka-consumer-groups.sh --bootstrap-server <broker_host>:9092 --list

输出示例：

consumer_group_1
consumer_group_2
consumer_group_3

5.2 查看消费者组详情

用途：查看某个消费者组的分区偏移量和滞后量。

命令：

bin/kafka-consumer-groups.sh --bootstrap-server <broker_host>:9092 --describe --group consumer_group_1

输出示例：

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID
consumer_group_1 my_topic        0          100             200             100             consumer-1
consumer_group_1 my_topic        1          150             150             0               consumer-2

5.3 重置消费者组的偏移量

用途：将消费者组的偏移量重置到指定位置。

命令：

bin/kafka-consumer-groups.sh --bootstrap-server <broker_host>:9092 --group consumer_group_1 --topic my_topic --reset-offsets --to-earliest --execute

说明：

• --to-earliest：将偏移量重置到最早位置。
• --to-latest：将偏移量重置到最新位置。
• --shift-by N：向前或向后移动偏移量。

6. 数据转储命令

6.1 导出主题数据到文件

用途：将 Kafka 主题中的消息导出到文件。

命令：

bin/kafka-console-consumer.sh --bootstrap-server <broker_host>:9092 --topic my_topic --from-beginning > /path/to/output.txt

6.2 从文件导入数据到主题

用途：将文件中的数据导入到 Kafka 主题。

命令：

bin/kafka-console-producer.sh --broker-list <broker_host>:9092 --topic my_topic < /path/to/input.txt

7. ZooKeeper 相关命令

7.1 查看 Kafka 在 ZooKeeper 中的信息

用途：通过 ZooKeeper 查看 Kafka 的元数据。

命令：

bin/zookeeper-shell.sh <zookeeper_host>:2181

操作：
进入 ZooKeeper CLI 后，可执行以下命令：

• 查看所有 Broker：

  ls /brokers/ids
  

• 查看所有消费者组：

  ls /consumers
  

8. 分区重新分配命令

8.1 生成分区重新分配计划

用途：生成新的分区分配方案。

命令：

bin/kafka-reassign-partitions.sh --bootstrap-server <broker_host>:9092 --generate --topics-to-move-json-file topics.json --broker-list "0,1,2"

示例 topics.json 文件：

{"topics": [{ "topic": "my_topic" }],"version": 1
}

8.2 执行分区重新分配

用途：执行分区重新分配计划。

命令：

bin/kafka-reassign-partitions.sh --bootstrap-server <broker_host>:9092 --execute --reassignment-json-file reassignment.json