搭建kafka集群

环境准备

修改hosts文件，进行解析

1. 搭建zookeeper集群

搭建kafka集群，需要先搭建zookeeper集群，搭建zookeeper集群，依赖于jdk

1.1 下载安装jdk8

下载jdk地址：https://www.oracle.com/java/technologies/javase/javase-jdk8-downloads.html

1.2 创建test用户，上传jdk包到服务器并解压安装包

[root@ecs0001 setup]# useradd test
[root@ecs0001 setup]# su - test
上传jdk软件包
[test@ecs0001 setup]# ls
jdk-8u144-linux-x64.tar.gz
[test@ecs0001 setup]# tar -zxvf jdk-8u144-linux-x64.tar.gz
[test@ecs0001 setup]# ls
jdk1.8.0_144  jdk-8u144-linux-x64.tar.gz

1.3 修改.bash_profile配置jdk，使环境变量生效

[test@ecs0001 ~]# vim .bash_profile
export JAVA_HOME=/home/test/jdk1.8.0_144
export JRE_HOME=${JAVA_HOME}/jre/
export CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
PATH=$JAVA_HOME/bin:$PATH:$HOME/.local/bin:$HOME/bin

[test@ecs0001 ~]# source .bash_profile
[test@ecs0001 ~]# java -version
java version "1.8.0_144"
Java(TM) SE Runtime Environment (build 1.8.0_144-b01)
Java HotSpot(TM) 64-Bit Server VM (build 25.144-b01, mixed mode)

1.4 下载zookeeper

下载地址： https://zookeeper.apache.org/releases.html
这里使用的是3.6.3

1.5 上传zookeeper软件包到服务器，解压并重命名

[test@ecs0001 ~]# ls
apache-zookeeper-3.6.3-bin.tar.gz
[test@ecs0001 ~]# tar -zxvf apache-zookeeper-3.6.3-bin.tar.gz
[test@ecs0001 ~]# ls
apache-zookeeper-3.6.3
[test@ecs0001 ~]#  mv apache-zookeeper-3.6.3-bin zookeeper-3.6.3

1.6 修改配置文件

[test@ecs0001 ~]# cd zookeeper-3.6.3/conf/
[test@ecs0001 conf]# cp zoo_sample.cfg zoo.cfg
[test@ecs0001 conf]# vim zoo.cfg
dataDir=/home/test/zookeeper/data
dataLogDir=/home/test/zookeeper/log
clientPort=2181
server.1:kafka01:2182:3181
server.2:kafka02:2182:3181
server.3:kafka03:2182:3181

配置说明：
dataDir=/home/test/zookeeper/data
dataLogDir=/home/test/zookeeper/log
为zookeeper数据和日志的路径，需要在启动zookeeper之前创建好。
clientPort=2181
为服务端口，服务器开放该端口即可或者根据需求修改
server.1:kafka01:2182:3181
server.2:kafka02:2182:3181
server.3:kafka03:2182:3181
为各个zookeeper服务器间的端口配置信息，server.1表示该服务器zookeeper的id号为1（后面会讲到,zookeeper的id号从1开始，且与后面讲到的myid文件一一对应），Mongo01为服务器hostname，2182、3181为zookeeper集群内部通信端口

1.7 创建配置文件dataDir、dataLogDir路径

[test@ecs0001 ~]# mkdir -pv /home/test/zookeeper/data
[test@ecs0001 ~]# mkdir -pv /home/test/zookeeper/log

并在dataDir路径下创建名为myid的文件，并添加内容 1 （该处myid对应的就是配置文件里的server.1）
[test@ecs0001 ~]# echo 1 > /home/test/zookeeper/data/myid

1.8 修改日志配置文件

[test@ecs0001 ~]# cd zookeeper-3.6.3/conf/
[test@ecs0001 conf]# vim log4j.properties
zookeeper.root.logger=INFO, ROLLINGFILE     //修改这行，保存退出

[test@ecs0001 conf]# cd ../bin/
[test@ecs0001 bin]$ vim zkEnv.sh
if [ "x${ZOO_LOG_DIR}" = "x" ]
then
    ZOO_LOG_DIR="/home/test/zookeeper/log"
fi

if [ "x${ZOO_LOG4J_PROP}" = "x" ]
then
    ZOO_LOG4J_PROP="INFO,ROLLINGFILE"
fi
修改这两部分内容

1.9 复制到zookeeper到其他服务器

kafka01上zookeeper配置完毕，在kafka02、kafka03上创建用户、配置JDK环境，将kafka01上配置好的zookeeper安装包直接拷贝到kafka02和kafka03上，并将myid文件的内容分别修改为2和3

1.10 启动zookeeper

分别启动三台服务器，进入zookeeper-3.6.3路径下

[test@ecs0001 ~]$ cd zookeeper-3.6.3/bin/
bin/zkServer.sh start

1.11 检测zookeeper启动状态

三台启动完成后 ./zkServer.sh status 看到一下内容即为集群成功两台服务器是 follower 一台服务器是 leader

[test@ecs0001 bin]# ./zkServer.sh status
Mode: follower
[test@ecs0002 bin]# ./zkServer.sh status
Mode: leader
[test@ecs0003 bin]# ./zkServer.sh status
Mode: follower

2. kafka集群搭建

下载kafka,这里是中文官网地址 http://kafka.apachecn.org/
下载的时候使用二进制文件

2.1 上传kafka软件包到服务器，并解压

[test@ecs0001 ~]# ls kafka_2.11-0.8.2.2.tgz
kafka_2.11-0.8.2.2.tgz
[test@ecs0001 ~]# tar -zxvf kafka_2.11-0.8.2.2.tgz
[test@ecs0001 ~]# ls
kafka_2.11-0.8.2.2

2.2 修改配置文件

[test@ecs0001 ~]# cd kafka_2.11-0.8.2.2/config/
[test@ecs0001 config]# vim server.properties
broker.id=0
port=9092
host.name=kafka01
log.dirs=/home/test/kafka_2.11-0.8.2.2/log
log.cleaner.enable=true
zookeeper.connect=kafka01:2181,kafka02:2181,kafka03:2181

主要修改上面几项
解释：broker.id(标示当前server在集群中的id，从0开始)，host.name(当前的server host name)，zookeeper.connect(连接的zookeeper集群)，log.dirs(log的存储目录，记得对应的去创建这个目录)等，其他的一些配置可以看相应的注释

2.3 复制kafka包到其他服务器

把修改好的kafka安装包拷贝到Mongo02和Mongo03上，修改每台server对应的配置文件，主要是修改其中的broker.id 和 host.name 属性：broker.id从0开始递增，每台server必须唯一

2.4 启动kafka

[test@ecs0001 bin]# ./kafka-server-start.sh -daemon ../config/server.properties

检查是否启动成功使用 ps -ef | grep kafka  显示一下内容即为启动成功
[test@ecs0001 bin]# ps -ef| grep kafka

2.5 kafka的一些操作

创建topic：

[test@ecs0001 bin]# ./kafka-topics.sh --create --zookeeper kafka01:2181,kafka02:2181,kafka03:2181 --replication-factor 3 --partitions 1 --topic test

查看topic：

[test@ecs0001 bin]# ./kafka-topics.sh --zookeeper kafka01:2181,kafka02:2181,kafka03:2181 --list
test

创建producer：

[test@ecs0001 bin]# ./kafka-console-producer.sh --broker-list kafka01:9092,kafka02:9092,kafka03:9092 --topic test_topic

创建consumer：

[test@ecs0002 bin]# ./kafka-console-consumer.sh --zookeeper kafka01:2181,kafka02:2181,kafka03:2181 --topic test_topic

通过在创建好的producer控制台输入信息，在consumer的控制台检测输出来测试，如果可以同步接受到信息就说明简单的kakfa 集群搭好了，另外可以根据项目的实际需求进一步做配置。

目录CONTENT