本次实验的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有什么区别？
