kafka 2.0.0 集群配置

　　本次实验将部署三个节点（broker）的kafka集群，同时zookeeper也部署三个节点，kafka和zk部署在同一台服务器，由于这两个组件都是java语言，所以只要服务器有jdk环境就可以直接启动。
　　关于kafka的介绍这里就不写了，本文将采用普通账户运行zookeeper及kafka ，同时会添加到 systemctl 服务。
基本环境介绍：
　1. 系统版本：CentOS Linux release 7.3.1611 (Core)
　2. 组件版本：zookeeper-3.4.13 ， kafka_2.11-2.0.0
官方下载地址：
　kafka https://www.apache.org/dyn/closer.cgi?path=/kafka/2.0.0/kafka_2.11-2.0.0.tgz
　zookeeper http://mirrors.hust.edu.cn/apache/zookeeper/
　　zookeeper 官网下载也会让选择国内镜像仓库下载，可以通过华中科大的镜像下载，kafka在官网直接下载即可，注意不要下载到了源码包，要下载Binary downloads 二进制包。
　3. 服务器：
　　kafka_1：192.168.8.8
　　kafka_2：192.168.8.9
　　kafka_3：192.168.8.10
准备工作：
1.关闭selinux服务，关闭iptables及firewalld；
2.三台服务器安装jdk_1.8 ，kafka 2.0版本需要jdk1.8 以上支持；
3.统一约定
　　1). kafka及zookeeper 采用同一个普通用户运行
　　2). kafka及zookeeper 程序文件位置为 /usr/local/kafka/ ，日志及数据存放到/data/kafka_data 目录

下面操作都在kafka_1 上配置，其他两台参照kafka-1 进行调整即可。
一：zookeeper 集群配置
1.创建运行用户kafka ，设置为禁止登陆

[root@kafka_1 ~]# useradd -M -s /sbin/nologin kafka

2.创建zookeeper 日志及数据存放目录

[root@kafka_1 ~]# mkdir -p /data/kafka_data/{kafka,zookeeper}
[root@kafka_1 ~]# mkdir -p /data/kafka_data/zookeeper/logs

3.修改zoo.cfg 配置

[root@kafka_1 ~]# tar zxvf /root/zookeeper-3.4.13.tar.gz -C /usr/local/kafka/
[root@kafka_1 ~]# cd /usr/local/kafka/zookeeper-3.4.13/conf/
[root@kafka_1 conf]# cp zoo_sample.cfg   zoo.cfg

[root@kafka_1 conf]# vim zoo.cfg
# The number of milliseconds of each tick
tickTime=2000
# The number of ticks that the initial 
# synchronization phase can take
initLimit=10
# The number of ticks that can pass between 
# sending a request and getting an acknowledgement
syncLimit=5

dataDir=/data/kafka_data/zookeeper
dataLogDir=/data/kafka_data/zookeeper/logs
# the port at which the clients will connect
clientPort=2181
# the maximum number of client connections.
# increase this if you need to handle more clients
#maxClientCnxns=60
#
# Be sure to read the maintenance section of the 
# administrator guide before turning on autopurge.
#
# http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
#
# The number of snapshots to retain in dataDir
#autopurge.snapRetainCount=3
# Purge task interval in hours
# Set to "0" to disable auto purge feature
#autopurge.purgeInterval=1
server.1=192.168.8.8:2888:3888
server.2=192.168.8.9:2888:3888
server.3=192.168.8.10:2888:3888

　　主要修改的地方其实只有dataDir，dataLogDir 以及对应的集群机器IP和通信端口，dataDir 设置到我们定义的目录下。
【server】的配置含义说明：
server.A=B:C:D
　其中 A 是一个数字，表示这个是第几号服务器；
　B 是这个服务器的 ip 地址；
　C 表示的是这个服务器与集群中的 Leader 服务器交换信息的端口；
　D 表示的是如果集群中的 Leader 服务器故障，需要一个端口来重新进行选举，选出一个新的 Leader，而这个端口就是用来执行选举时服务器相互通信的端口。
4. 在dataDir目录下创建ID 文件，该文件指定这是集群中的几号服务器

[root@kafka_1 conf]# echo 1 > /data/kafka_data/zookeeper/myid
[root@kafka_1 conf]# chown kafka.kafka -R /data/kafka_data
[root@kafka_1 conf]# chown kafka.kafka -R /usr/local/kafka/

5.创建systemd 服务启动文件

[root@kafka_1 conf]# vim /etc/systemd/system/zookeeper.service

[Unit]
Description=ZooKeeper Service
Documentation=http://zookeeper.apache.org
Requires=network.target
After=network.target

[Service]
Type=forking
User=kafka
Group=kafka
Environment=ZOO_LOG_DIR=/data/kafka_data/zookeeper/logs
Environment=JAVA_HOME=/usr/java/jdk1.8.0_121
ExecStart=/usr/local/kafka/zookeeper-3.4.13/bin/zkServer.sh start
ExecStop=/usr/local/kafka/zookeeper-3.4.13/bin/zkServer.sh stop
Restart=always

[Install]
WantedBy=default.target

　　Environment=ZOO_LOG_DIR 不加的话，zookeeper 的control 日志会因为权限问题普通账户会因为无法写入日志而无法启动
　　开始的时候没有加JAVA_HOME 的参数导致无法启动，查看到/data/kafka_data/zookeeper/logs 目录下zookeeper.out 文件有日志，打开文件只有一句：

nohup: failed to run command ‘java’: No such file or directory

应该是服务器普通账号无法拿到java环境，指定JAVA_HOME路径即可正常启动。

[root@kafka_1 conf]# systemctl daemon-reload
[root@kafka_1 conf]# systemctl start zookeeper

　　这样即可通过systemctl status zookeeper 查看当前节点服务启动状态，剩下的两台节点参照当前配置，只需要修改dataDir目录下创建myid 为对应服务器即可，三台节点都启动后可以在任意节点上通过zookeeper 命令查看当前机器在集群中的身份：

[root@kafka_1 ~]# /usr/local/kafka/zookeeper-3.4.13/bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/kafka/zookeeper-3.4.13/bin/../conf/zoo.cfg
Mode: leader

[root@kafka_2 ~]# /usr/local/kafka/zookeeper-3.4.13/bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /usr/local/kafka/zookeeper-3.4.13/bin/../conf/zoo.cfg
Mode: follower

由此可以看到kafka_1 被选举为了leader 节点，kafka集群就配置完成了。

二：kafka集群配置
1. kafka集群主要是配置文件的修改

[root@kafka_1 ~]# tar zxvf  kafka_2.11-2.0.0.tgz -C /usr/local/kafka/

下面是需要调整修改的参数

[root@kafka_1 ~]# cd /usr/local/kafka/kafka_2.11-2.0.0/config
[root@kafka_1 config]# vim server.properties

broker.id=0
listeners=PLAINTEXT://:9092
advertised.listeners=PLAINTEXT://192.168.8.8:9092
log.dirs=/data/kafka_data/kafka
zookeeper.connect=192.168.8.8:2181,192.168.8.9:2181,192.168.8.10:2181

　当前服务器是kafka_1 ，对应的broker.id 设置为0 ，kafka_2 则设置为1 ，kafka_2 设置为2 ；
　advertised.listeners 是对外宣告的地址，需设置为本机IP地址；
　log.dirs 是kafka文件存储位置，设置为之前约定的目录；
　zookeeper.connect 是连接zookeeper服务器的地址，这里填写上集群内的全部地址。
2. 创建systemd 服务启动文件

[root@kafka_1 config]# vim /etc/systemd/system/kafka.service

[Unit]
Description=Apache Kafka server (broker)
Documentation=http://kafka.apache.org/documentation.html
Requires=network.target remote-fs.target
After=network.target remote-fs.target kafka-zookeeper.service

[Service]
Type=simple
User=kafka
Group=kafka
Environment=JAVA_HOME=/usr/java/jdk1.8.0_121
ExecStart=/usr/local/kafka/kafka_2.11-2.0.0/bin/kafka-server-start.sh /usr/local/kafka/kafka_2.11-2.0.0/config/server.properties
ExecStop=/usr/local/kafka/kafka_2.11-2.0.0/bin/kafka-server-stop.sh

[Install]
WantedBy=multi-user.target

3. 启动kafka

[root@kafka_1 config]# systemctl daemon-reload
[root@kafka_1 config]# systemctl start kafka

查看端口监听情况

[root@kafka_1 config]# netstat -tunlp | grep java
tcp6       0      0 :::9092                 :::*                    LISTEN      4262/java           
tcp6       0      0 :::2181                 :::*                    LISTEN      3205/java

同理在其余两台节点上配置kafka参数并启动，需要注意的是broker.id 的值不可设置重复。
4. 验证kafka集群
　　在kafka_1 机器上创建一个topic ，在kafka_2 上用生产者身份连接该topic ，然后在kafka_3 上用消费者身份查看topic中的消息；
kafka_1 上创建一个名为cnyunwei的topic

[root@kafka_1 ~]# /usr/local/kafka/kafka_2.11-2.0.0/bin/kafka-topics.sh --create --zookeeper 127.0.0.1:2181 --replication-factor 1 --partitions 1 --topic cnyunwei
Created topic "cnyunwei".

kafka_2 上用生产者模式连接cnyunwei ，并模拟消息发送

[root@kafka_2 ~]# /usr/local/kafka/kafka_2.11-2.0.0/bin/kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic cnyunwei
>www.cnyunwei.cc
>www.cnyunwei.cc
>

kafka_3 上用消费者模式连接cnyunwei 进行消息查看

[root@kafka_3 config]# /usr/local/kafka/kafka_2.11-2.0.0/bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic cnyunwei --from-beginning
www.cnyunwei.cc
www.cnyunwei.cc

到此kafka 2.0 版本集群就搭建完成了。

转载请注明：菜鸟运维网 » kafka 2.0.0 集群配置