四、Kafka集群操作
1. 创建topic
创建一个名字为test的主题， 有三个分区，有两个副本：
bin/kafka-topics.sh --create --zookeeper node01:2181 --replication-factor 2 --partitions 3 --topic test
2. 查看主题命令
查看kafka当中存在的主题：
bin/kafka-topics.sh --list --zookeeper node01:2181,node02:2181,node03:2181
3. 生产者生产数据
模拟生产者来生产数据:
bin/kafka-console-producer.sh --broker-list node01:9092,node02:9092,node03:9092 --topic test
4. 消费者消费数据
执行以下命令来模拟消费者进行消费数据:
bin/kafka-console-consumer.sh --from-beginning --topic test --zookeeper node01:2181,node02:2181,node03:2181
5. 运行describe topics命令
执行以下命令运行describe查看topic的相关信息:
bin/kafka-topics.sh --describe --zookeeper node01:2181 --topic test
结果说明：
这是输出的解释。第一行给出了所有分区的摘要，每个附加行提供有关一个分区的信息。由于我们只有一个分 区用于此主题，因此只有一行。
“leader”是负责给定分区的所有读取和写入的节点。每个节点将成为随机选择的分区部分的领导者。（因为在kafka中 如果有多个副本的话，就会存在leader和follower的关系，表示当前这个副本为leader所在的broker是哪一个）
“replicas”是复制此分区日志的节点列表，无论它们是否为领导者，或者即使它们当前处于活动状态。（所有副本列表0,1,2）
“isr”是“同步”复制品的集合。这是副本列表的子集，该列表当前处于活跃状态并且已经被领导者捕获。（可用的列表数）
6. 增加topic分区数
执行以下命令可以增加topic分区数:
bin/kafka-topics.sh --zookeeper zkhost:port --alter --topic topicName --partitions 8
7. 增加配置
动态修改kakfa的配置:
bin/kafka-topics.sh --zookeeper node01:2181 --alter --topic test --config flush.messages=1
8. 删除配置
动态删除kafka集群配置:
bin/kafka-topics.sh --zookeeper node01:2181 --alter --topic test --delete-config flush.messages
9. 删除topic
目前删除topic在默认情况下知识打上一个删除的标记，在重新启动kafka后才删除。
如果需要立即删除，则需要在 server.properties中配置：
delete.topic.enable=true
然后执行以下命令进行删除topic:
kafka-topics.sh --zookeeper zkhost:port --delete --topic topicName