前言

软件环境:

主机	IP	系统	hostname
server1	192.168.0.57	centos7.4	kafka-1
server2	192.168.0.58	centos7.4	kafka-2
server3	192.168.0.59	centos7.4	kafka-3

软件依赖:

Kafka集群是把状态保存在Zookeeper中的，首先要搭建Zookeeper集群。
Zookeeper是用java写的所以需要搭建JAVA环境(JDK版本 >= 1.7)。

安装策略:

先在主机server1上搭建好环境，然后直接将安装配置拷贝到server2，server3上。

1.安装JDK

目前官网不能直接下载，在网上找到1.8.0版本安装包下载到本地。

1.1.下载jdk

#远程登陆主机server1 (使用Xshell)
$ssh root@192.168.0.57

#将jdk包传输到root用户根目录下(使用FileZilla)

#查看root用户根目录是否存在jdk-8u171-linux-x64.tar.gz
kafka-1:~ # ls
anaconda-ks.cfg  jdk-8u171-linux-x64.tar.gz

1.2.安装jdk

#创建java安装目录
kafka-1:~ # mkdir /opt/java

#解压jdk包到安装目录
kafka-1:~ # tar zxvf jdk-8u171-linux-x64.tar.gz -C /opt/java

1.3.配置java环境变量

#在文件中追加配置:
kafka-1:~ # vim /etc/profile
export JAVA_HOME=/opt/java/jdk1.8.0_171
export PATH=$JAVA_HOME/bin:$PATH

#使配置生效
kafka-1:~ # source /etc/profile

1.4.验证jdk是否安装成功

kafka-1:~ # java -version
java version "1.8.0_171"
Java(TM) SE Runtime Environment (build 1.8.0_171-b11)
Java HotSpot(TM) 64-Bit Server VM (build 25.171-b11, mixed mode)

2.安装zookeeper集群

下载地址：https://mirrors.cnnic.cn/apache/zookeeper (这里我们选择最新的3.4.12版本)

2.1.下载zookeeper

#windows浏览器下载
1. 访问 https://mirrors.cnnic.cn/apache/zookeeper/zookeeper-3.4.12/zookeeper-3.4.12.tar.gz
2. 将下载的zookeeper-3.4.12.tar.gz传输到root用户根目录(/root)下

#linux命令直接下载
kafka-1:~ # wget https://mirrors.cnnic.cn/apache/zookeeper/zookeeper-3.4.12/zookeeper-3.4.12.tar.gz

#查看linux是否已存在zookeeper-3.4.12.tar.gz
kafka-1:~ # ls
anaconda-ks.cfg  jdk-8u171-linux-x64.tar.gz  zookeeper-3.4.12.tar.gz

2.2.安装zookeeper

#创建安装目录/opt/zookeeper
kafka-1:~ # mkdir /opt/zookeeper

#解压zookeeper包到安装目录
kafka-1:~ # tar zxvf zookeeper-3.4.12.tar.gz -C /opt/zookeeper/

2.3.配置zookeeper环境变量

#在文件中追加配置:
kafka-1:~ # vim /etc/profile
export ZOOKEEPER_HOME=/opt/zookeeper/zookeeper-3.4.12
export PATH=$ZOOKEEPER_HOME/bin:$PATH

#使配置生效
kafka-1:~ # source /etc/profile

2.4.验证zookeeper是否安装成功

#验证环境变量是否配置成功
kafka-1:~ # echo $ZOOKEEPER_HOME
/opt/zookeeper/zookeeper-3.4.12

#输入zk按TAB键补全,看是否会有下面命令输出
kafka-1:~ # zk
zkCleanup.sh zkCli.cmd zkCli.sh zkEnv.cmd zkEnv.sh zkServer.cmd zkServer.sh

5.创建zookeeper数据存储路径

#创建data目录
kafka-1:~ # mkdir /opt/zookeeper/data

#在data目录下创建zkdata目录存放快照日志 
kafka-1:~ # mkdir /opt/zookeeper/data/zkdata 

#在data目录下创建zkdatalog目录存放事物日志 
kafka-1:~ # mkdir /opt/zookeeper/data/zkdatalog

2.6.拷贝zookeeper配置文件

#进入zookeeper安装路径的conf目录
kafka-1:~ # cd /opt/zookeeper/zookeeper-3.4.12/conf

#复制zoo_sample.cfg生成新的zoo.cfg文件
kafka-1:/opt/zookeeper/zookeeper-3.4.12/conf # cp zoo_sample.cfg zoo.cfg
kafka-1:/opt/zookeeper/zookeeper-3.4.12/conf # ls
configuration.xsl  log4j.properties  zoo.cfg  zoo_sample.cfg

zoo_sample.cfg 这个文件是官方给我们的zookeeper的样板文件，给他复制一份命名为zoo.cfg，zoo.cfg是官方指定的文件命名规则.

2.7.修改zookeeper配置文件

kafka-1:/opt/zookeeper/zookeeper-3.4.12/conf # vim zoo.cfg
#将dataDir路径改为我们上面创建的路径
dataDir=/opt/zookeeper/data/zkdata

#在dataDir下面添加dataLogDir路径配置
dataLogDir=/opt/zookeeper/data/zkdatalog

#在文件尾部追加服务器配置
server.1=192.168.0.57:2888:3888
server.2=192.168.0.58:2888:3888
server.3=192.168.0.59:2888:3888

zoo.cfg 配置文件参数详解:

#tickTime：
这个时间是作为 Zookeeper 服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个tickTime 时间就会发送一个心跳。
#initLimit：
这个配置项是用来配置 Zookeeper接受客户端（这里所说的客户端不是用户连接 Zookeeper 服务器的客户端，而是 Zookeeper 服务器
集群中连接到Leader的Follower服务器）初始化连接时最长能忍受多少个心跳时间间隔数。当已经超过5个心跳的时间（也就是tickTime）
长度后Zookeeper服务器还没有收到客户端的返回信息，那么表明这个客户端连接失败。总的时间长度就是 5*2000=10 秒
#syncLimit：
这个配置项标识 Leader与Follower之间发送消息，请求和应答时间长度，最长不能超过多少个 tickTime 的时间长度，总的时间长度就是5*2000=10秒
#dataDir：
快照日志的存储路径
#dataLogDir：
事物日志的存储路径，如果不配置这个那么事物日志会默认存储到dataDir制定的目录，这样会严重影响zk的性能，当zk吞吐量较大的时候，
产生的事物日志、快照日志太多。
#clientPort：
这个端口就是客户端连接Zookeeper服务器的端口，Zookeeper会监听这个端口，接受客户端的访问请求。修改他的端口改大点。

#集群服务器配置:
#server.1
这个1是服务器的标识也可以是其他的数字，表示这个是第几号服务器，用来标识服务器，这个标识要写到快照目录下面myid文件里。
#192.168.0.57:2888:3888
192.168.0.57为集群里的IP地址，第一个端口是master和slave之间的通信端口，默认是2888，第二个端口是leader选举的端口，集群刚启动的时候
选举或者leader挂掉之后进行新的选举的端口默认是3888。

2.8.创建myid文件

在快照目录/opt/zookeeper/data/zkdata/下创建服务器标识文件myid
这里的”1”是根据zoo.cfg配置文件的服务器配置来设置的

kafka-1:~ # echo "1" > /opt/zookeeper/data/zkdata/myid

2.9.将zookeeper安装及配置拷贝到其他两台主机

#1.在server1上拷贝java安装文件到server2,server3
kafka-1:~ # scp -r /opt/java root@192.168.0.58:/opt/
kafka-1:~ # scp -r /opt/java root@192.168.0.59:/opt/

#2.在server1上拷贝zookeeper安装文件到server2,server3
kafka-1:~ # scp -r /opt/zookeeper root@192.168.0.58:/opt/
kafka-1:~ # scp -r /opt/zookeeper root@192.168.0.59:/opt/

#3.在server1上拷贝zookeeper环境变量配置文件到server2,server3
kafka-1:~ # scp /etc/profile root@192.168.0.58:/etc/profile
kafka-1:~ # scp /etc/profile root@192.168.0.59:/etc/profile

#4.在server2上修改服务器标识为2,并使环境变量配置生效
kafka-2:~ # echo "2" > /opt/zookeeper/data/zkdata/myid
kafka-2:~ # source /etc/profile

#5.在server3上修改服务器标识为3,并使环境变量配置生效
kafka-3:~ # echo "3" > /opt/zookeeper/data/zkdata/myid
kafka-3:~ # source /etc/profile

2.10.验证zookeeper集群是否搭建成功

#分别在server1,server2,server3上启动zookeeper服务
kafka-1:~ #zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/zookeeper-3.4.12/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

kafka-2:~ # zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/zookeeper-3.4.12/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

kafka-3:~ # zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/zookeeper-3.4.12/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

#查看集群中各服务器的zookeeper状态
kafka-1:~ #zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/zookeeper-3.4.12/bin/../conf/zoo.cfg
Mode: follower

kafka-2:~ #zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/zookeeper-3.4.12/bin/../conf/zoo.cfg
Mode: leader

kafka-3:~ #zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/zookeeper-3.4.12/bin/../conf/zoo.cfg
Mode: follower

#可以用"jps"查看zookeeper的进程,这个是zk的整个工程的main.
kafka-1:~ # jps
5952 Jps
5439 QuorumPeerMain

#zk集群一般只有一个leader,多个follower,主一般是响应客户端的读写请求,
#而备从主同步数据,当主挂掉之后就会从follower里投票选举一个leader出来.

#可以看到zookeeper集群已经搭建成功,server2被选举为leader.

3.安装kafka集群

下载地址：https://mirrors.cnnic.cn/apache/kafka (这里我们选择最新的2.12-1.1.0版本)

3.1.下载kafka

#windows浏览器下载
1. 访问 https://mirrors.cnnic.cn/apache/kafka/1.1.0/kafka_2.12-1.1.0.tgz
2. 将下载的kafka_2.12-1.1.0.tgz传输到root用户根目录下

#linux命令直接下载
kafka-1:~ # wget https://mirrors.cnnic.cn/apache/kafka/1.1.0/kafka_2.12-1.1.0.tgz

#查看linux是否已存在kafka_2.12-1.1.0.tgz
kafka-1:~ # ls
anaconda-ks.cfg  jdk-8u171-linux-x64.tar.gz  kafka_2.12-1.1.0.tgz  zookeeper-3.4.12.tar.gz

3.2.安装kafka

#创建安装目录/opt/kafka
kafka-1:~ # mkdir /opt/kafka

#解压kafka包到安装目录
kafka-1:~ # tar zxvf kafka_2.12-1.1.0.tgz -C /opt/kafka

3.3.配置kafka环境变量

#在文件中追加配置:
kafka-1:~ # vim /etc/profile
export KAFKA_HOME=/opt/kafka/kafka_2.12-1.1.0
export PATH=KAFKA_HOME/bin:$PATH

#使配置生效
kafka-1:~ # source /etc/profile

3.4.验证kafka是否安装成功

#验证环境变量是否配置成功
kafka-1:~ # echo $KAFKA_HOME
/opt/kafka/kafka_2.12-1.1.0

#输入kafka-按TAB键补全,看是否会有下面命令输出
kafka-1:~ # kafka-
kafka-acls.sh kafka-console-producer.sh kafka-delete-records.sh ...

3.5.创建kafka消息存放目录

#创建data目录
kafka-1:~ # mkdir /opt/kafka/data

#在data目录下创建kafkalogs目录存放kafka消息
kafka-1:~ # mkdir /opt/kafka/data/kafkalogs

3.6.修改kafka配置文件

#进入config目录
kafka-1:~ # cd /opt/kafka/kafka_2.12-1.1.0/config
kafka-1:/opt/kafka_2.12-1.1.0/config #ll
total 72
-rw-r--r-- 1 root root  906 Mar 24 06:51 connect-console-sink.properties
-rw-r--r-- 1 root root  909 Mar 24 06:51 connect-console-source.properties
-rw-r--r-- 1 root root 5807 Mar 24 06:51 connect-distributed.properties
-rw-r--r-- 1 root root  883 Mar 24 06:51 connect-file-sink.properties
-rw-r--r-- 1 root root  881 Mar 24 06:51 connect-file-source.properties
-rw-r--r-- 1 root root 1111 Mar 24 06:51 connect-log4j.properties
-rw-r--r-- 1 root root 2730 Mar 24 06:51 connect-standalone.properties
-rw-r--r-- 1 root root 1221 Mar 24 06:51 consumer.properties
-rw-r--r-- 1 root root 4727 Mar 24 06:51 log4j.properties
-rw-r--r-- 1 root root 1919 Mar 24 06:51 producer.properties
-rw-r--r-- 1 root root 7033 Jun 25 19:29 server.properties
-rw-r--r-- 1 root root 6851 Jun 19 17:22 server.properties.bak
-rw-r--r-- 1 root root 1032 Mar 24 06:51 tools-log4j.properties
-rw-r--r-- 1 root root 1023 Mar 24 06:51 zookeeper.properties

主要关注：server.properties 这个文件即可，我们可以发现在目录下：
有很多文件，这里可以发现有Zookeeper文件，我们可以根据Kafka内带的zk集群来启动，但是建议使用独立的zk集群。不过在单节点部署中是启用的本身的ZK服务。

server.properties 配置文件详解:

broker.id=1  #当前机器在集群中的唯一标识，和zookeeper的myid性质一样
host.name=192.168.0.57 #这个参数默认是关闭的，在0.8.1有个bug，DNS解析问题，失败率的问题。
port=9092 #当前kafka对外提供服务的端口，默认是9092
delete.topic.enable=true  #允许删除topic，默认是false
num.network.threads=3 #这个是borker进行网络处理的线程数
num.io.threads=8 #这个是borker进行I/O处理的线程数
log.dirs=/opt/kafka/data/kafkalogs
#消息存放的目录，这个目录可以配置为“，”逗号分割的表达式，上面的num.io.threads要大于这个目录的个数这个目录，
如果配置多个目录，新创建的topic他把消息持久化的地方是，当前以逗号分割的目录中，那个分区数最少就放那一个
socket.send.buffer.bytes=102400 #发送缓冲区buffer大小，数据不是一下子就发送的，先回存储到缓冲区了到达一定的大小后在发送，能提高性能
socket.receive.buffer.bytes=102400 #kafka接收缓冲区大小，当数据到达一定大小后在序列化到磁盘
socket.request.max.bytes=104857600 #这个参数是向kafka请求消息或者向kafka发送消息的请请求的最大数，这个值不能超过java的堆栈大小
num.partitions=1 #默认的分区数，一个topic默认1个分区数
log.retention.hours=168 #默认消息的最大持久化时间，168小时，7天
message.max.byte=5242880  #消息保存的最大值5M
default.replication.factor=2  #kafka保存消息的副本数，如果一个副本失效了，另一个还可以继续提供服务
replica.fetch.max.bytes=5242880  #取消息的最大直接数
log.segment.bytes=1073741824 #这个参数是：因为kafka的消息是以追加的形式落地到文件，当超过这个值的时候，kafka会新起一个文件
log.retention.check.interval.ms=300000 #每隔300000毫秒去检查上面配置的log失效时间(log.retention.hours=168),到目录查看是否有过期的消息如果有则删除
log.cleaner.enable=false #是否启用log压缩，一般不用启用，启用的话可以提高性能
zookeeper.connect=192.168.0.57:2181,192.168.0.58:2181,192.168.0.59:2181 #设置zookeeper的连接端口

上面是参数的解释，实际的修改项为：

#每台服务器的broker.id都不能相同
broker.id=1  

#在broker.id下面新增 
host.name=192.168.0.57
delete.topic.enable=true

#修改消息存放路径为上面创建的目录
log.dirs=/opt/kafka/data/kafkalogs

#在log.retention.hours=168下面新增下面三项
message.max.byte=5242880
default.replication.factor=2 
replica.fetch.max.bytes=5242880 

#设置zookeeper的连接端口
zookeeper.connect=192.168.0.57:2181,192.168.0.58:2181,192.168.0.59:2181

3.7.将kafka安装及配置拷贝到其他两台主机

#1.在server1上拷贝kafka安装文件到server2,server3
kafka-1:~ # scp -r /opt/kafka root@192.168.0.58:/opt/
kafka-1:~ # scp -r /opt/kafka root@192.168.0.59:/opt/

#2.在server1上拷贝kafka环境变量配置文件到server2,server3
kafka-1:~ # scp /etc/profile root@192.168.0.58:/etc/profile
kafka-1:~ # scp /etc/profile root@192.168.0.59:/etc/profile

#3.在server2,,server3上分别使环境变量生效
kafka-2:~ # source /etc/profile
kafka-3:~ # source /etc/profile

#4.在server2上修改kafka配置文件
kafka-2:~ # vim /opt/kafka/kafka_2.12-1.1.0/config/server.properties
broker.id=2
host.name=192.168.0.58

#5.在server3上修改kafka配置文件
kafka-3:~ # vim /opt/kafka/kafka_2.12-1.1.0/config/server.properties
broker.id=3
host.name=192.168.0.59

3.8.验证kafka集群是否搭建成功

#1.分别在server1,server2,server3上启动kafka服务
kafka-1:~ # kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties
kafka-2:~ # kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties
kafka-3:~ # kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties

#2.检查server1,server2,server3上kafka服务是否启动
kafka-1:~ # jps
6504 Jps
6428 Kafka
5439 QuorumPeerMain
kafka-2:~ # jps
24562 Jps
24501 Kafka
8681 QuorumPeerMain
kafka-3:~ # jps
19037 Jps
25737 Kafka
3153 QuorumPeerMain

#3.创建topic，在server1,server2,,server3上查看
kafka-1:~ # kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 2 --partitions 1 --topic test
Created topic "test".
kafka-1:~ # kafka-topics.sh --list --zookeeper localhost:2181
test
kafka-2:~ # kafka-topics.sh --list --zookeeper localhost:2181
test
kafka-3:~ # kafka-topics.sh --list --zookeeper localhost:2181
test

3.9.kafka日志说明

kafka-1:~ # ll /opt/kafka/kafka_2.12-1.1.0/logs/
total 104
-rw-r--r-- 1 root root  7121 Jun 29 17:34 controller.log
-rw-r--r-- 1 root root     0 Jun 29 17:30 kafka-authorizer.log
-rw-r--r-- 1 root root     0 Jun 29 17:30 kafka-request.log
-rw-r--r-- 1 root root  6127 Jun 29 17:32 kafkaServer-gc.log.0.current
-rw-r--r-- 1 root root 27026 Jun 29 17:34 kafkaServer.out
-rw-r--r-- 1 root root   172 Jun 29 17:32 log-cleaner.log
-rw-r--r-- 1 root root 46419 Jun 29 17:34 server.log
-rw-r--r-- 1 root root  5525 Jun 29 17:34 state-change.log

#server.log 
kafka的运行日志
#state-change.log  
kafka他是用zookeeper来保存状态，所以他可能会进行切换，切换的日志就保存在这里
#controller.log 
kafka选择一个节点作为“controller”,当发现有节点down掉的时候它负责在所有的follower节点中选择新的leader,这使得Kafka可以批量的
高效的管理所有分区节点的主从关系。如果controller down掉了，活着的节点中的一个会被切换为新的controller.

3.10.zookeeper客户端操作

#使用客户端进入zookeeper
kafka-1:~ # zkCli.sh 
#如果修改了zookeeper端口可以通过加 '-server' 参数来运行
#例如: zkCli.sh -server localhost:port

#查看目录情况 执行 "ls /"
[zk: localhost:2181(CONNECTED) 0] ls /
[cluster, controller_epoch, controller, brokers, zookeeper, admin, isr_change_notification, consumers, 
log_dir_event_notification, latest_producer_id_block, config]
#上面的显示结果中：只有zookeeper是原生的，其他都是Kafka创建的

#显示kafka配置中broker.id为1的节点信息
[zk: localhost:2181(CONNECTED) 1] get /brokers/ids/1
{"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},"endpoints":["PLAINTEXT://192.168.0.57:9092"],
"jmx_port":-1,"host":"192.168.0.57","timestamp":"1530264745420","port":9092,"version":4}
cZxid = 0x200000010
ctime = Fri Jun 29 17:32:32 CST 2018
mZxid = 0x200000010
mtime = Fri Jun 29 17:32:32 CST 2018
pZxid = 0x200000010
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x1007b29e9360000
dataLength = 194
numChildren = 0

#查看topic为test的partition信息
[zk: localhost:2181(CONNECTED) 2] get /brokers/topics/test/partitions/0
null
cZxid = 0x20000003b
ctime = Fri Jun 29 17:34:15 CST 2018
mZxid = 0x20000003b
mtime = Fri Jun 29 17:34:15 CST 2018
pZxid = 0x20000003c
cversion = 1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 1

4.zookeeper&kafka操作命令

#启动zookeeper
zkServer.sh start

#关闭zookeeper
zkServer.sh stop

#查看zookeeper状态
zkServer.sh status

#启动kafka
kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties

#关闭kafka
kafka-server-stop.sh

#创建topic
kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 2 --partitions 1 --topic topicname
#--replication-factor 2   #复制两份
#--partitions 1 #创建1个分区
#--topic #主题为topicname

#展示topic
kafka-topics.sh --list --zookeeper localhost:2181

#查看topic状态
kafka-topics.sh --describe --zookeeper localhost:2181 --topic topicname

#删除topic
kafka-topics.sh --delete --zookeeper 192.168.0.57:2181 --topic topicname
#需要在配置文件中设置delete.topic.enable=true,默认是false
#然后删除log.dirs指定的文件目录，重新启动kafka

#创建生产者
kafka-console-producer.sh --broker-list 192.168.0.57:9092 --topic topicname

#创建消费者
kafka-console-consumer.sh --zookeeper 192.168.0.57:2181 --topic topicname --from-beginnin