1.  环境规划：

主机名	IP地址	角色
node1	192.168.56.111	ElasticSearch(master) Zookeeper Kafka
node2	192.168.56.112	ElasticSearch(slave) Kibana Zookeeper Kafka
node3	192.168.56.113	ElasticSearch(slave) Zookeeper Kafka
node4	192.168.56.114	Logstash Filebeat

2.  配置Zookeeper集群：

        1>.  安装zookeeper：三个节点操作一致。

[root@node1 ~]# mkdir /data/

[root@node1 ~]# tar xf zookeeper-3.4.10.tar.gz -C /data/

[root@node1 ~]# cd /data

[root@node1 data]# ln -sv zookeeper-3.4.10/ zookeeper
'zookeeper' -> 'zookeeper-3.4.10/'

        2>.  配置zookeeper，三个节点都要配置：

[root@node1 data]# cd zookeeper/conf

[root@node1 conf]# cp zoo_sample.cfg zoo.cfg

[root@node1 conf]# vim zoo.cfg

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data/zookeeper/data  ##这个目录用来存放快照、myid等数据
dataLogDir=/data/zookeeper/logs  ##这个目录用来存放事务日志
clientPort=2181  ##zookeeper连接端口
server.1=192.168.56.111:2888:3888
server.2=192.168.56.112:2888:3888 
server.3=192.168.56.113:2888:3888
server.4=192.168.56.114:2888:3888:observer ##选配

[root@node1 ~]# mkdir /data/zookeeper/{data,logs} 

[root@node1 ~]# echo 1 > /data/zookeeper/data/myid

[root@node2 ~]# echo 2 > /data/zookeeper/data/myid

[root@node3 ~]# echo 3 > /data/zookeeper/data/myid

注意：server.id=host:port:port，其中id为1-255，2888是集群间通讯的端口，3888为Leader

选举的端口。

        3>.  配置观察者节点observer模式：(选配)

观察者节点不会参与选举，也不参与数据事务提交的ack应答；但是数据会跟集群同步。

[root@node4 data]# cd zookeeper/conf

[root@node4 conf]# cp zoo_sample.cfg zoo.cfg

[root@node4 conf]# vim zoo.cfg

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data/zookeeper/data  ##这个目录用来存放快照、myid等数据
dataLogDir=/data/zookeeper/logs  ##这个目录用来存放事务日志
clientPort=2181  ##zookeeper连接端口
peerType=observer  ##配置该选项的节点就是observer节点
server.1=192.168.56.111:2888:3888
server.2=192.168.56.112:2888:3888 
server.3=192.168.56.113:2888:3888
server.4=192.168.56.114:2888:3888:observer

[root@node4 ~]# mkdir /data/zookeeper/{data,logs} 

[root@node4 ~]# echo 4 > /data/zookeeper/data/myid

        4>.  启动zookeeper：三个节点同时操作。

[root@node1 ~]# cd /data/zookeeper/bin/

[root@node1 bin]# ./zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /data/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

        5>.  查看各个节点zookeeper状态：

[root@node1 bin]# ./zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /data/zookeeper/bin/../conf/zoo.cfg
Mode: follower

[root@node2 bin]# ./zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /data/zookeeper/bin/../conf/zoo.cfg
Mode: leader

[root@node3 bin]# ./zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /data/zookeeper/bin/../conf/zoo.cfg
Mode: follower

3.  配置Kafka集群：

        1>. 安装kafka：三个节点操作一致。

[root@node1 ~]# tar xf kafka_2.12-2.6.0.tgz -C /data/ 

[root@node1 ~]# cd /data/ 

[root@node1 data]# ln -sv kafka_2.12-2.6.0/ kafka 
'kafka' -> 'kafka_2.12-2.6.0/'

        2>.  配置kafka：三个节点都要操作。

[root@node1 data]# cd kafka/config/

[root@node1 config]# cp server.properties{,.bak}

[root@node1 config]# vim server.properties

broker.id=1  ##zookeeper myid
delete.topic.enable=true
listeners=PLAINTEXT://192.168.56.111:9092  ##修改成本机地址
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka/kafka-logs
num.partitions=3
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.56.111:2181,192.168.56.112:2181,192.168.56.113:2181  ##配置zookeeper集群
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0

        3>.  启动Kafka：三个节点都要操作。

[root@node1 ~]# cd /data/kafka/bin/

[root@node1 bin]# ./kafka-server-start.sh -daemon /data/kafka/config/server.properties

##停止kafka
[root@node1 bin]# ./kafka-server-stop.sh

        4>.  测试kafka：

## 创建一个名为"test"的topic，它有一个分区和一个副本： 
[root@node1 kafka]# bin/kafka-topics.sh --create --zookeeper 192.168.56.111:2181,192.168.56.112:2181,192.168.56.113:2181 --replication-factor 1 --partitions 1 --topic test 
Created topic test.

## 查看topic是否创建成功：
[root@node1 kafka]# bin/kafka-topics.sh --list --zookeeper 192.168.56.111:2181,192.168.56.112:2181,192.168.56.113:2181
test

## 运行一个producer，发布消息：ctrl+d结束消息发布
[root@node1 kafka]# bin/kafka-console-producer.sh --broker-list 192.168.56.111:9092 --topic test
>hello world1
hello world2>

## 运行一个consumer，订阅主题看看是否可以收到消息：另一个节点测试
[root@node2 kafka]# bin/kafka-console-consumer.sh --bootstrap-server 192.168.56.112:9092 --topic test --from-beginning
hello world1
hello world2

##删除主题：
[root@node1 kafka]# bin/kafka-topics.sh --delete --zookeeper 192.168.56.111:2181,192.168.56.112:2181,192.168.56.113:2181 --topic test