Kafka | 常用命令行操作

本次实验的kafka集群有三个节点，即有三个broker。

一、Topic 的增删查

Topic操作的执行脚本在bin目录下的 kafka-topics.sh

1.查看当前服务器中已存在的topic

# --list  --zookeeper
[root@node01 bin]# ./kafka-topics.sh --list --zookeeper node01:2181

2.新增topic

# --create 新增操作
# --topic 定义topic名称
# --partitions 定义分区数
# --replication-factor 定义副本数
[root@node01 bin]# ./kafka-topics.sh --create --zookeeper node01:2181 --topic first --partitions 2 --replication-factor 2

例子中创建名为first的topic，2个分区，2个副本。
创建后可分别在三个节点下，进入kafka的存放数据的目录，查看topic的创建情况：first-0和first-1是以topic名和分区号组成，代表了2个分区；每个分区副本都是2个，而且相同分区副本是放在不同的节点上的，例如first-0的副本在node01和node03节点上，first-1的副本在node01和node02上。

node01的topic.png
node02的topic.png
node03的topic.png

如何定义分区数和副本数？

一般而言，kafka集群的拥有越多的分区也意味着可以获取更高的吞吐。
kafka集群内部的副本机制，可以保证高可用。一个分区如果有多个副本，每个副本将会存储到不同的broker上，副本被设计为leader和follow副本。副本数不能超过broker数。

3.删除topic

# --delete 删除操作
# --topic 指定需要删除的topic
[root@node01 bin]# ./kafka-topics.sh --delete --zookeeper node01:2181 --topic first
Topic first is marked for deletion. //这个topic被标记删除
Note: This will have no impact if delete.topic.enable is not set to true.如果delete.topic.enable //这个属性不设置为true的话，将不会真正的删除topic

查看数据目录下的topic是否被删除：发现topic名称后面多了delete删除标志，不要着急，等过会来看这个topic就会被删除。

执行删除操作的topic.png

遇到topic只是被标记为删除，不是真正删除，应怎么解决？

方案一：需要在server.properties中设置delete.topic.enable=true，重启kafka。
方案二：手动删除
1）删除zookeeper上topic的数据
/brokers/ids/topics/xxx
/config/topics/xxx
2） 删除该topic所有partition和replica的数据
数据在所有broker的log.dirs目录下，文件夹结构是topic-partition的方 式，直接将该topic的整个文件夹删除即可

4.查看某个topic的详情

# --describe  查看详情操作
[root@node01 bin]# ./kafka-topics.sh --describe --zookeeper node01:2181 --topic second 
Topic: second   PartitionCount: 2   ReplicationFactor: 3    Configs: 
    Topic: second   Partition: 0    Leader: 2   Replicas: 2,3,1 Isr: 2,3,1
    Topic: second   Partition: 1    Leader: 3   Replicas: 3,1,2 Isr: 3,1,2

Isr代表的含义是什么？

二、控制台生产者和消费者

图片.png

启动控制台生产者

# --broker-list 选择发送消息的kafka服务端
# --topic 指定topic
[root@node01 bin]# ./kafka-console-producer.sh --broker-list node01:9092 --topic second
> //再这里写入消息

启动控制台消费者

# --bootstrap-server  选择接收消息的kafka服务端
# --topic 指定topic，接收哪个topic的消息
# --from-beginning：会把first主题中以往所有的数据都读取出来。根据业务场景选择是否增加该配置。
[root@node02 bin]# ./kafka-console-consumer.sh --bootstrap-server node01:9092 --topic second  --from-beginning

发送消息

#在生产者窗口输入消息，发送完成后Control+C退出
>hi
>my first kafka topic

每输入一条消息，消费者窗口会显示生产者发出的消息

--broker-list 和 --bootstrap-server有什么区别？