1、准备

1.安装包

1.Java环境jdk：https://www.oracle.com/java/technologies/javase-downloads.html

2.Kafka安装包：https://kafka.apache.org/downloads

暂定：jdk-8u261-linux-x64.tar.gz + kafka_2.12-2.1.1.tgz

2.环境配置

1.配置Java环境（见附录）

2、单机部署

1.cd /usr/local/
2.将kafka_2.12-2.1.1.tgz上传至此目录
3.解压：tar zxf kafka_2.12-2.1.1.tgz
4.解压完文件夹改名：mv kafka_2.12-2.1.1/ /usr/local/kafka
5.cd /usr/local/kafka/bin/
6.后台启动zookeeper：./zookeeper-server-start.sh …/config/zookeeper.properties &
7.后台启动kafka：./kafka-server-start.sh …/config/server.properties &
8.查看端口是否在监听：netstat -anput | grep 9092
9.kafka-tools验证

注意：
1、允许远程访问
修改/usr/local/kafka/config/server.properties配置文件

2、防火墙开放端口或关闭防火墙
查看防火墙状态：firewall-cmd --state
停止firewall：systemctl stop firewalld.service
禁止firewall开机启动：systemctl disable firewalld.service

开放80端口： firewall-cmd --zone=public --add-port=80/tcp --permanent
重启防火墙：firewall-cmd --reload
查询有哪些端口是开启的：firewall-cmd --list-port

3、集群部署

1、kafka集群的服务器至少要3台（且必须为奇数）

2、服务器信息

192.168.8.195 test01
192.168.8.196 test02
192.168.8.197 test03

3、host配置（/etc/hosts）

三台服务器配置一致，/etc/hosts文件末尾添加
192.168.8.195 test01
192.168.8.196 test02
192.168.8.197 test03

4、Zookeeper集群搭建（三台服务器均需配置）

1、将安装包kafka_2.12-2.1.1.tgz上传到/opt 目录下
2、解压：tar -zxvf kafka_2.12-2.1.1.tgz
3、进入目录：cd /opt/kafka_2.12-2.1.1/
4、创建zookeeper目录：mkdir zk_kfk_data
5、进入目录：cd /opt/kafka_2.12-2.1.1/config
6、修改zookeeper.properties文件：

三台机器上的zookeeper.properties文件配置相同，data.Dir 为zookeeper的数据目录，server.1、server.2、server.3 为集群信息。

2888端口号是zookeeper服务之间通信的端口
3888端口是zookeeper与其他应用程序通信的端口。

tickTime：CS通信心跳数
Zookeeper 服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每个 tickTime 时间就会发送一个心跳。
tickTime以毫秒为单位。
tickTime：该参数用来定义心跳的间隔时间，zookeeper的客户端和服务端之间也有和web开发里类似的session的概念，而zookeeper里最小的session过期时间就是tickTime的两倍。

initLimit：LF初始通信时限
集群中的follower服务器(F)与leader服务器(L)之间 初始连接 时能容忍的最多心跳数（tickTime的数量）

syncLimit：LF同步通信时限
集群中的follower服务器(F)与leader服务器(L)之间 请求和应答 之间能容忍的最多心跳数（tickTime的数量）

7、创建myid文件：进入/opt/kafka_2.12-2.1.1/zk_kfk_data目录，创建myid文件，将三台服务器上的myid文件分别写入1，2，3（对应zookeeper.properties文件配置）。myid是zookeeper集群用来发现彼此的标识，必须创建，且不能相同。
8、执行启动zookeeper命令：
1.进入bin目录
cd /opt/kafka_2.12-2.1.1/bin
2.后台启动zookeeper
nohup ./zookeeper-server-start.sh …/config/zookeeper.properties &>> /opt/kafka_2.12-2.1.1/zookeeper.log &
3.三台服务器执行同样操作，三台服务器zookeeper全部启动前，日志会报错，但三台全启动后，日志会正常。
2、服务器信息
服务器IP 域名
192.168.8.195 test01
192.168.8.196 test02
192.168.8.197 test03

5、Kafka集群搭建

1、进入目录：cd /opt/kafka_2.12-2.1.1/
2、创建kafka日志数据目录：mkdir kafka-logs-1
3、进入目录：cd /opt/kafka_2.12-2.1.1/config
4、修改server.properties配置文件：
注意：
自动创建topic：配置新增auto.create.topics.enable=true
彻底删除topic：配置新增delete.topic.enable=true

5、 启动kafka集群：
1.进入bin目录
cd /opt/kafka_2.12-2.1.1/bin
2.后台启动zookeeper
nohup ./kafka-server-start.sh …/config/server.properties &>> /opt/kafka_2.12-2.1.1/kafka.log &
3.三台服务器执行同样操作
三个节点均要启动；启动无报错，即搭建成功。

6、简单测试kafka集群

1、创建topic：test
./kafka-topics.sh --create --zookeeper test01:3181,test02:3181,test03:3181 --replication-factor 1 --partitions 1 --topic test001
2、列出已创建的topic列表
./kafka-topics.sh --list --zookeeper localhost:3181
3、模拟客户端去发送消息
生产者代码验证
4、模拟客户端去接受消息
消费者代码验证

7、注意（开放相应端口或关闭防火墙）

4、附录

1、Java环境

1．cd /usr/local
2．上传jdk包（jdk-8u261-linux-x64.tar.gz）至此目录
3．解压tar -zxvf /usr/local/jdk-8u261-linux-x64.tar.gz
4．解压完目录名为jdk-8u261-linux-x64，将其修改为java
mv /usr/local/jdk-8u261-linux-x64 /usr/local/java
5．修改配置文件，配置环境变量
1.cd /etc
2.修改profile文件，在最后加入
export JAVA_HOME=/usr/local/java
export PATH=$PATH:$JAVA_HOME/bin
export CLASSPATH=.:$JAV{A}_{H}OME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
export JRE_HOME=$JAVA_HOME/jre
6．让配置文件生效
source /etc/profile
7．java -version验证环境，安装成功会打印版本信息

2、部分错误信息

1.新增topic时报错replication factor: 1 larger than available brokers: 0
出现这个问题的原因是kafka没有启动的情况下想去create tpoic（可能启动时未后台启动），所以应该是先去进到kafka安装目录的bin路径下，执行./kafka-server-start.sh …/config/server.properties & 打开kafka即可！

2.zookeeper集群启动报错：Cannot open channel to * at election address /ip:3888
可能原因：
1.防火墙未关闭或防火墙开启，相关端口未开放
2.三台服务器都设置myid
3.zookeeper.properties中最后server信息与域名对应中，本机不能使用域名，应使用0.0.0.0