Kafka基本命令

Kafka是一个分布式流处理平台，提供了高吞吐量的数据传输能力。通过Kafka命令行工具，你可以轻松地与Kafka集群进行交互，包括生产、消费数据以及查看和修改偏移量。以下是一份使用指南，帮助你掌握这些基本操作。

基本命令包含创建Topic，查看所有Topic，查看Topic详情，删除Topic，模拟生产者生产数据，模拟消费者消费数据。

这也是上个博客的一种广度上的一种延伸，后续还会接着在这里补充。

【Spark笔耕不辍(三)】Kafka生产者消费者API及核心原理 | SilverSucks (weiswift.github.io)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# 创建主题：指定分区数和副本数
kafka-topics.sh --create --topic bigdata01 --partitions 3 --replication-factor 2 --bootstrap-server node1:9092,node2:9092,node3:9092

# 创建主题：不指定分区数和副本数，默认是1个分区，1个副本
kafka-topics.sh --create --topic test1 --bootstrap-server node1:9092,node2:9092,node3:9092

# 查看所有主题
kafka-topics.sh --list -bootstrap-server node1:9092,node2:9092,node3:9092

# 查看某一个主题的详情
kafka-topics.sh --describe --topic bigdata01  --bootstrap-server node1:9092,node2:9092,node3:9092
kafka-topics.sh --describe --topic test1  --bootstrap-server node1:9092,node2:9092,node3:9092

# 删除topic
kafka-topics.sh --delete --topic test1 --bootstrap-server node1:9092,node2:9092,node3:9092

#5、模拟生成者和消费者
#生成者
kafka-console-producer.sh --topic bigdata01 --broker-list node1:9092,node2:9092,node3:9092

#消费者
kafka-console-consumer.sh --topic bigdata01 --bootstrap-server node1:9092,node2:9092,node3:9092  --from-beginning

高级命令

生产者(指定参数)，消费者(指定参数)，以及查看主题中有多少条数据。

1. 生产数据
使用kafka-console-producer命令行工具可以向Kafka主题发送消息。以下是使用该工具的示例：

1
kafka-console-producer --broker-list <broker地址>:<端口> --topic <主题名称>

在打开的终端中，输入你要发送的消息，然后按下回车键。消息将被发送到指定的Kafka主题中。
2. 消费数据
使用kafka-console-consumer命令行工具可以从Kafka主题中消费消息。以下是使用该工具的示例：

1
kafka-console-consumer --bootstrap-server <broker地址>:<端口> --topic <主题名称> --from-beginning

这将从指定的Kafka主题中消费所有消息，包括从该主题创建以来的所有消息。如果你只想消费最新的消息，可以将--from-beginning选项替换为--from-latest。
3. 查看偏移量
要查看消费者当前的偏移量，可以使用以下命令：

1
kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list <broker地址>:<端口> --topic <主题名称> --time -1 --topic <副主题名称>

这将显示指定主题中每个分区当前偏移量的列表。--time -1选项表示获取最新的偏移量。你可以根据需要修改该选项来获取不同时间点的偏移量。
4. 修改偏移量
在某些情况下，你可能需要手动修改消费者的偏移量。可以使用以下命令来修改指定分区的偏移量：

1
kafka-run-class.sh kafka.tools.ResetOffsetShell --broker-list <broker地址>:<端口> --topic <主题名称> --partition <分区编号> --offset <新偏移量> --delete --topic <副主题名称>