本文介绍利用kafka命令进行topic的创建，和生产者，消费者的示例
首先要保证已经安装了zookeeper集群和kafka集群，如果没有安装，可参考：
zookeeper集群搭建
kafka集群搭建
一.环境介绍
本次教程的环境安装了三台zookeeper和三个kafka，分别放在了三个虚拟机上，之所以搭建三台zookeeper这里也是有原因的，可以参考文章：zookeeper集群为什么建议单数
而且三台虚拟机为了正常通信已经把防火墙关闭。
zookeeper版本，kafka版本：
二.启动环境
1.启动zookeeper
首先将zookeeper进行启动，因为已经把zookeeper的bin目录放到环境变量里边了，所以可以在任何地方调用zookeeper命令，关于集群搭建的链接已经放最上边了
执行启动命令：
zkServer.sh start
启动后要看是否启动成功：
zkServer.sh status
但实际我第一次启动就没有启动成功：
这个没有启动成功有很多原因，一方面是环境搭建问题，可以通过查看用户根目录下的zookeeper.out进行日志的查看锁定问题，还有一方面可能是因为其他节点还没有启动，我这里的原因是因为其他节点没有启动，所以我就在第二台虚拟机也启动zookeeper：
可以看到已经启动成功，并且从status中可以了解到这是一个主节点，然后我们再回到第一台虚拟机中执行status命令（不需要再次start）：
可以看到第一台变成了从节点
同样，在第三台虚拟机进行启动就行了
2.启动kafka
同样已经配置过kafka环境变量
启动命令：
kafka-server-start.sh /opt/kafka_2.12-2.3.1/config/server.properties &
这里记录下我启动中报的一个错误，当我启动第二台虚拟机上的kafka时第一台就报了下边错误：
这是因为我的虚拟机中没有配置/etc/hosts文件，我们在该文件中添加上对应的主机名信息就行了：
kafka启动成功：
在多台虚拟机上都进行启动就行了
三.生产者，消费者案例
1.首先创建一个topic：
kafka-topics.sh –create –zookeeper weijiabin1:2181,weijiabin2:2181,weijiabin3:2181 –replication-factor 3 –partitions 3 –topic test
–create：表示创建
–zookeeper 后面的参数是zk的集群节点
–replication-factor 3 ：表示复本数
–partitions 3：表示分区数
–topic test：表示topic的主题名称
查看topic列表：
kafka-topics.sh –list –zookeeper weijiabin1:2181
查看指定topic的详细信息：
kafka-topics.sh –topic test –describe –zookeeper weijiabin1:2181
既然能看到这篇文章，那么对上边的详细信息肯定也不是很了解，顺便做一下说明：
上边config：后边每一行都是一个partition的信息，而上边数字就是kafka的节点编号：
“leader”是负责给定分区的所有读写的节点。每个节点都将是随机选择的分区部分的leader。
“replicas”是复制此分区的日志的节点列表，而不管这些节点是主节点还是当前活动节点。
“isr”是一组“同步”副本。这是replicas的子集，该子集当前处于活动状态，并与leader保持联系。
顺便附上删除topic命令：
kafka-topics.sh –delete –zookeeper weijiabin1:2181 –topic test
2.执行生产者命令
在第一台虚拟机上执行：
kafka-console-producer.sh –broker-list weijiabin1:9092 –topic test
3.执行消费者命令
在第二台虚拟机上执行(我们也可以在第三台虚拟机上也启动，这样的结果是第二台第三台都会收到test topic的信息)：
kafka0.9版本之前：
kafka-console-consumer.sh –zookeeper weijiabin1:2181 –from-beginning –topic test
kafka0.9版本之后
kafka-console-consumer.sh –bootstrap-server localhost:9092 –topic test –from-beginning
4.结果
第一台虚拟机：
消费者所在虚拟机：
