Kafka集群搭建 _kafka集群

宁可枝头抱香死，何曾吹落北风中。这篇文章主要讲述Kafka集群搭建相关的知识，希望能为你提供帮助。
环境软件准备
Centos7 64位系统 3台（2*n+1）
kafka_2.12-0.11.0.1
zookeeper-3.4.6.tar
jdk-8u161-linux-x64.tar
配置网络ip
关闭防火墙临时关闭systemctl stop firewalld.service
关闭开机自启动systemctl disable firewalld.service
【Kafka集群搭建】2、软件安装
JDK安装
先安装jdk-8u161-linux-x64.tar
系统安装好之后，用xshell远程链接服务器，用xftp将 kafka ,zookeeper,jdk上传到服务器上

文章图片

（1）先安装jdk配置环境变量。
（2）切换到/data目录下运行 tar -zxvf jdk-8u161-linux-x64.tar.gz
（3）配置jdk环境变量运行 vim /etc/profile 再按i键编辑profile 在最后加上
export java_HOME=/data/jdk1.8.0_161
export PATH=JAVA_HOME/bin:JAVAH?OME/bin:PATH
export CLASSPATH=.:JAVA_HOME/lib/dt.jar:JAVAH?OME/lib/dt.jar:JAVA_HOME/lib/tools.jar

文章图片

按ESC 然后:wq保存再执行source /etc/profile 使刚才改的profile立即生效。
（4）验证jdk是否正确安装运行Java -version

文章图片

验证环境变量是否配置正确运行Javac

文章图片

Zookeeper安装
安装zookeeper-3.4.6.tar
同样再/data目录下解压zookeeper-3.4.6.tar 运行 tar -zxvf zookeeper-3.4.6.tar
（2）配置zookeeper的环境变量和jdk是一样的只需在原有的基础上添加。添加完如下格式
export JAVA_HOME=/data/jdk1.8.0_161
export CLASSPATH=.:JAVA_HOME/lib/dt.jar:JAVAH?OME/lib/dt.jar:JAVA_HOME/lib/tools.jar
export ZOOKEEPER_HOME=/data/zookeeper-3.4.6
export PATH=JAVA_HOME/bin:JAVAH?OME/bin:PATH:$ZOOKEEPER_HOME/bin
然后再任何目录下运行zkS三个字母然后按tab键就可以提示出zkServer.sh 等完整命令。
找到/data/zookeeper-3.4.6/conf下

文章图片

看到zoo_sample.cfg 配置文件这个是文件是例子，然后复制出一个改名为zoo.cfg
参数按照下图配置

文章图片

server.1=192.168.237.6:4888:5888
这里的1是一个数字，与myid文件中的id是一致的。右边可以配置两个端口，第一个端口用于F和L之间的数据同步和其它通信，第二个端口用于Leader选举过程中投票通信
然后在zookeeper保存数据的目录新建zookeeper-data文件夹，在这个文件夹新建myid文件内容写1命令如下

文章图片

这个1就是server.1.代表集群中的第一台机器的标识
Kafka安装
（1）运行tar -zxvf kafka_2.12-0.11.0.1.tgz
（2）配置kafka环境变量

文章图片

export JAVA_HOME=/data/jdk1.8.0_161
export CLASSPATH=.:JAVA_HOME/lib/dt.jar:JAVAH?OME/lib/dt.jar:JAVA_HOME/lib/tools.jar
export ZOOKEEPER_HOME=/data/zookeeper-3.4.6
export KAFKA_HOME=/data/kafka_2.12-0.11.0.1
export PATH=JAVA_HOME/bin:JAVAH?OME/bin:PATH:ZOOKEEPER_HOME/bin:ZOOKEEPERH?OME/bin:KAFKA_HOME/bin
（3）修改/data/kafka_2.12-0.11.0.1/config路径下的server.properties文件如下附件

文章图片

#每一个broker在集群中的唯一表示，和zookeeper的myid性质一样要求是正数。当该服务器的IP地址发生改变时，broker.id没有变化，则不会影响consumers的消息情况，
broker.id=1
#监听地址一般场景下只配置listeners即可，在Kafka集群内部通讯走内网，而对外提供服务的是公网或者隔离网络，则需要配置
listeners=PLAINTEXT://10.10.11.91:9092
advertised.listeners=PLAINTEXT://10.10.11.91:9092
#zookeeper连接服务器地址
zookeeper.connect=10.10.11.91:2181
#kafka日志的存放地址，多个地址的话用逗号分割 /data/kafka-logs-1，/data/kafka-logs-2
log.dirs=/data/kafka-logs

文章图片

文章图片

先启动zookeeper 服务器执行zkServer.sh start

文章图片

启动完后说明启动成功

文章图片

然后进入kafka_2.12-0.11.0.1文件夹，执行 nohup bin/kafka-server-start.sh config/server.properties & 后台运行

文章图片

执行jps,可以查看到后台运行程序，说明kafka启动成功。

文章图片

kafka启动成功
一台服务器搭建完毕，然后再克隆出两台服务器。复制出来服务器务必修改server.properties中的broker.id、listeners 和advertised.listeners还有zookeeper日志目录下myid 修改成相对应的值
同时启动三台服务器，先启动zookeeper 三台服务器顺序执行zkServer.sh start

文章图片

启动完之后三台服务器再顺序执行如下三张图

文章图片

文章图片

说明zookeeper集群搭建和启动成功
然后三台服务器，分别进入kafka_2.12-0.11.0.1文件夹，执行 nohup bin/kafka-server-start.sh config/server.properties &
如下图

文章图片

执行jps,可以查看到后台运行程序，说明kafka启动成功。

文章图片

三台kafka都启动成功
3、 kafka测试
随便一个节点创建主题执行 kafka-topics.sh --create --topic hello --zookeeper 192.168.2.164:2181 --partitions 3 --replication-factor 3 如下图

文章图片

查看主题执行kafka-topics.sh --describe --zookeeper 192.168.2.164:2181 --topic hello

文章图片

Replication-Factor :备份数量
PartitionCount:分区数量
Topic:主题名称
Leader:现在使用的副节点节点序号。
Replicas:副节点优先级
Isr:可用的副节点列表
使用kafka-启动producer发送消息
执行下面的命令，启动producer:
kafka-console-producer.sh --broker-list 192.168.2.164:9092 --topic hello
出现> 然后输入