Kafka集群搭建

Posted 2021-09-18 wx6144491ba79ae

环境软件准备
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

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
出现>然后输入

启动消费者
kafka-console-consumer.sh --zookeeper 192.168.2.164:2181 --from-beginning --topic hello
看到如下图

刚才输入的信息。