Kafka集群部署

1、安装准备

[root@kafka-node00 ~]# cat /etc/redhat-release 
CentOS Linux release 7.6.1810 (Core) 
[root@kafka-node00 ~]# uname -r
3.10.0-957.21.3.el7.x86_64
[root@kafka-node00 ~]# ll /opt/src/
total 283012
-rw-r--r-- 1 root root 195094741 Dec 25 16:06 jdk-8u221-linux-x64.tar.gz
-rw-r--r-- 1 root root  57028557 Mar 23  2019 kafka_2.12-2.2.0.tgz
-rw-r--r-- 1 root root  37676320 Apr  1  2019 zookeeper-3.4.14.tar.gz

2、安装JDK

[root@kafka-node00 src]# mkdir /usr/java
[root@kafka-node00 src]# tar xf jdk-8u221-linux-x64.tar.gz -C /usr/java
[root@kafka-node00 src]# ln -s /usr/java/jdk1.8.0_221 /usr/java/jdk
[root@kafka-node00 src]# vi /etc/profile

export JAVA_HOME=/usr/java/jdk
export PATH=$JAVA_HOME/bin:$JAVA_HOME/bin:$PATH
export CLASSPATH=$CLASSPATH:$JAVA_HOME/lib:$JAVA_HOME/lib/tools.jar

[root@kafka-node00 src]# source /etc/profile
[root@kafka-node00 src]# java -version
java version "1.8.0_221"
Java(TM) SE Runtime Environment (build 1.8.0_221-b11)
Java HotSpot(TM) 64-Bit Server VM (build 25.221-b11, mixed mode)

3、安装zookeeper

[root@kafka-node00 src]# tar xf zookeeper-3.4.14.tar.gz -C /opt/
[root@kafka-node00 src]# ln -s /opt/zookeeper-3.4.14 /opt/zookeeper
[root@kafka-node00 src]# mkdir -pv /data/zookeeper/data /data/zookeeper/logs
mkdir: created directory ‘/data’
mkdir: created directory ‘/data/zookeeper’
mkdir: created directory ‘/data/zookeeper/data’
mkdir: created directory ‘/data/zookeeper/logs’

修改配置文件zoo.cfg

[root@kafka-node00 ~]# vi /opt/zookeeper/conf/zoo.cfg 

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data/zookeeper/data
dataLogDir=/data/zookeeper/logs
clientPort=2181
server.0=kafka-node00.tuoren.com:2888:3888
server.1=kafka-node01.tuoren.com:2888:3888
server.2=kafka-node02.tuoren.com:2888:3888

　　①、tickTime：基本事件单元，这个时间是作为Zookeeper服务器之间或客户端与服务器之间维持心跳的时间间隔，每隔tickTime时间就会发送一个心跳；最小 的session过期时间为2倍tickTime

　　②、dataDir：存储内存中数据库快照的位置，除非另有说明，否则指向数据库更新的事务日志。注意：应该谨慎的选择日志存放的位置，使用专用的日志存储设备能够大大提高系统的性能，如果将日志存储在比较繁忙的存储设备上，那么将会很大程度上影像系统性能。

　　③、client：监听客户端连接的端口。

　　④、initLimit：允许follower连接并同步到Leader的初始化连接时间，以tickTime为单位。当初始化连接时间超过该值，则表示连接失败。

　　⑤、syncLimit：表示Leader与Follower之间发送消息时，请求和应答时间长度。如果follower在设置时间内不能与leader通信，那么此follower将会被丢弃。

　　⑥、server.A=B:C:D

　　　　A：其中 A 是一个数字，表示这个是服务器的编号；

　　　　B：是这个服务器的 ip 地址或域名（生产尽量使用域名）或主机名；

　　　　C：Leader选举的端口；

　　　　D：Zookeeper服务器之间的通信端口。

　　我们需要修改的第一个是 dataDir ,在指定的位置处创建好目录。

　　第二个需要新增的是 server.A=B:C:D 配置，其中 A 对应下面我们即将介绍的myid 文件。B是集群的各个IP地址，C:D 是端口配置。

创建myid文件

[root@kafka-node00 src]# vi /data/zookeeper/data/myid  
0

[root@kafka-node01 src]# vi /data/zookeeper/data/myid  
1

[root@kafka-node02 src]# vi /data/zookeeper/data/myid  
2

启动zookeeper服务

• 启动命令

[root@kafka-node00 ~]# zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

• 停止命令

[root@kafka-node00 ~]# zkServer.sh stop
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/bin/../conf/zoo.cfg
Stopping zookeeper ... STOPPED

• 重启命令

[root@kafka-node00 ~]# zkServer.sh restart
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/bin/../conf/zoo.cfg
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/bin/../conf/zoo.cfg
Stopping zookeeper ... STOPPED
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

• 查看集群状态命令

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

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

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

三台机器，node01 成功的通过了选举称为了leader,而剩下的两台成为了 follower。这时候，如果你将node01关掉，会发现剩下两台又会有一台变成了 leader节点。

4、安装Kafka

安装

[root@kafka-node00 src]# tar xf kafka_2.12-2.2.0.tgz -C /opt/
[root@kafka-node00 src]# ln -s /opt/kafka_2.12-2.2.0 /opt/kafka
[root@kafka-node00 src]# ll /opt/kafka/
total 56
drwxr-xr-x 3 root root  4096 Apr 28 16:05 bin
drwxr-xr-x 2 root root  4096 Apr 28 15:19 config
drwxr-xr-x 2 root root  4096 Apr 28 14:40 libs
-rw-r--r-- 1 root root 32216 Mar 10  2019 LICENSE
drwxr-xr-x 2 root root  4096 Apr 30 08:08 logs
-rw-r--r-- 1 root root   336 Mar 10  2019 NOTICE
drwxr-xr-x 2 root root  4096 Mar 10  2019 site-docs

配置

kafka-node00

[root@kafka-node00 kafka]# grep '^[^#|^$]' /opt/kafka/config/server.properties 
broker.id=0
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka/logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=kafka-node00.turoen.com:2181,kafka-node01.tuoren.com:2181,kafka-node02.tuoren.com:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
delete.topic.enable=true
host.name=kafka-node00.tuoren.com

kafka-node01

[root@kafka-node01 kafka]# grep '^[^#|^$]' /opt/kafka/config/server.properties 
broker.id=1
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka/logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=kafka-node00.turoen.com:2181,kafka-node01.tuoren.com:2181,kafka-node02.tuoren.com:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
delete.topic.enable=true
host.name=kafka-node00.tuoren.com

kafka-node02

[root@kafka-node02 kafka]# grep '^[^#|^$]' /opt/kafka/config/server.properties 
broker.id=2
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/data/kafka/logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=kafka-node00.turoen.com:2181,kafka-node01.tuoren.com:2181,kafka-node02.tuoren.com:2181
zookeeper.connection.timeout.ms=6000
group.initial.rebalance.delay.ms=0
delete.topic.enable=true
host.name=kafka-node00.tuoren.com

5、测试并验证

启动集群

[root@kafka-node00 kafka]# bin/kafka-server-start.sh -daemon config/server.properties 
[root@kafka-node01 kafka]# bin/kafka-server-start.sh -daemon config/server.properties 
[root@kafka-node02 kafka]# bin/kafka-server-start.sh -daemon config/server.properties

开启jmx

[root@kafka-node00 kafka]# JMX_PORT=9999 /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

列出所有topic

[root@kafka-node00 kafka]# bin/kafka-topics.sh --list --zookeeper kafka-node00:2181 
[root@kafka-node01 kafka]# bin/kafka-topics.sh --list --zookeeper kafka-node01:2181 
bin/kafka-topics.sh --list --zookeeper kafka-node02:2181