首先确保JDK已正确安装并设置好环境变量
安装Zookeeper
到官网下载： https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/
这里下载到/usr/local/software
下载后解压
tar -xzf zookeeper-3.4..tar.gz
进入zookeeper/conf目录，把配置复制一份
cp zoo_sample.cfg zoo.cfg
修改配置
dataDir=/opt/data/zookeeper
设置zookeeper环境变量
Linux 添加环境变量
启动
zkServer.sh start
查看状态
zkServer.sh status
使用jps查看
zookeeper集群配置
分别在两台CentOS上按上面步骤下载解压，配置环境变量，这里假设Server1(192.168.31.147)，Server2(192.168.31.146)
编辑上面的zoo.cfg，加上集群配置
server.=192.168.31.147::
server.=192.168.31.146::
2888端口表示该服务器与集群中的leader交换信息的端口,3888表示选举时服务器相互通信的端口，两台服务器防火墙打开这两个端口
firewall-cmd --zone=public --add-port=/tcp --permanent
firewall-cmd --zone=public --add-port=/tcp --permanent
firewall-cmd --reload
在主服务器（147）的/opt/data/zookeeper目录下新建一个myid文件，在里面存一个值1，zookeeper在启动时会读取myid文件中的值与zoo.cfg中的配置进行比较，以确定是哪台服务器
在从服务器（146）的/opt/data/zookeeper目录下同样建一个myid文件，存值为2
接下来分别运行再查看状态
147：
146：
这样集群配置就完成了。
安装kafka
官网下载：http://kafka.apache.org/downloads
同样下载到/usr/local/software目录
下载完后解压，修改安装目录下的/config/server.properties中zookeeper.connect配置
146配置，这里的broker.id与zookeeper的值一致
broker.id=
log.dirs=/opt/data/kafka-logs
zookeeper.connect=192.168.31.146:2181,192.168.31.147:2181
147配置
broker.id=
log.dirs=/opt/data/kafka-logs
zookeeper.connect=192.168.31.146:,192.168.31.147:
启动kafka
kafka-server-start.sh -daemon ../config/server.properties
使用jps查看
这里要注意，如果更改过broker.id的话，需要将原来生成的data删除，否则kafka会自动关闭，错误信息可以在kafka的安装目录下logs中server.log找到
两台服务器的zookeeper和kafka都启动后，在任意一台服务器上运行
zkCli.sh -server localhost:
进入控制台后执行
ls /brokers/ids
可以看到有两个节点信息，表示2台服务器kafka正常启动
查看brokers信息
可以看到brokerid为2的节点作为Leader控制器，此时我们kill掉brokerid=2的kafka进程
再查看会发现kafka集群选出另外一个节点作为leader控制器
关闭kafka节点
kafka-server-stop.sh
主题管理
创建主题
kafka-topics.sh --create --zookeeper 192.168.31.147:,192.168.31.146: --replication-factor --partitions --topic kafka-action
查看主题
为了方便，将IP添加到host中，然后使用如下脚本查看
kafka-topics.sh --list --zookeeper server-:,server-:
kafka-topics.sh --describe --zookeeper server-:,server-:
生产者
kafka-console-producer.sh --broker-list localhost: --topic kafka-action --property parse.key=true
启动一个生产者，同时指定启用消息的key配置，默认key与消息之间使用制表符分割，输入tab键，输入格式不对会报错
也可以自定义分隔符，比如空格
kafka-console-producer.sh --broker-list localhost: --topic kafka-action --property parse.key=true --perty key.separator=' '
使用以下命令验证消息是否发送成功
kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost: --topic kafka-action --time -
表示：主题名+分区编号+偏移量
如果开启了auto.create.topics.enable=true，生产者向一个不存在的topic发送消息时会自动创建一个
kafka-console-producer.sh --broker-list localhost: --topic producer-create-topic
控制台会提示主题不存在并创建一个
查看log目录
可以通过以下命令来查看消息
kafka-run-class.sh kafka.tools.DumpLogSegments --files /neworiental/logs/kafka-logs/producer-create-topic-/.log
消费者
kafka-console-consumer.sh --bootstrap-server localhost: --consumer-property group.id=new-consumer-test --consumer-property client.id=new-consumer-c1 --topic kafka-action
先启动一个消费者，然后再按上文启动一个生产者，在生产者发消息，消费者接受消息
单播和多播
要实现单播，让多个消费者同属于一个消费组即可，属于不同的消费组即可实现多播
查看消费组
kafka-consumer-groups.sh --bootstrap-server localhost: --list
查看消费偏移量
kafka-consumer-groups.sh --bootstrap-server localhost: --describe --group new-consumer-test
安装kafka-manager
下载地址：https://github.com/yahoo/kafka-manager
下载后解压，在安装目录运行
./sbt clean dist
此过程相当耗时，编译完成后会在提示目录生成一个zip文件
将此文件解压到指定位置，编辑conf目录下的application.conf，修改zookeeper地址
kafka-manager.zkhosts="localhost:2181"
cd到bin目录，执行下面命令
./kafka-manager -Dconfig.file=../conf/application.conf
后台启动
nohup ./kafka-manager -Dconfig.file=../conf/application.conf >& &
默认监听9000端口
打开浏览器
默认进来是没有集群的，需要添加一个
其他选项默认即可
添加完成后可以看到
点击名字进去
点击Topic后面的数字进去
启动一个消费者
kafka-console-consumer.sh --bootstrap-server localhost: --consumer-property group.id=new-consumer-test --consumer-property client.id=new-consumer-c1 --topic newtopic
在页面查看
