Kafka搭建(集群版)

Kafka单机版

部署前提

VMware环境 : 两台centos系统

Jdk包:jdk-8u202-linux-x64.tar.gz
Kafka包:kafka_2.12-3.5.0.tgz
Zookeeper包:apache-zookeeper-3.7.2-bin.tar.gz

百度网盘自取: 链接: https://pan.baidu.com/s/11EWuhBoSmH3musd_3Rgodw?pwd=e32t 提取码: e32t

Kafka搭建(集群版)

创建服务器

• 修改克隆机主机名，以下以kafka-broker1举例说明

#使用root用户登录
#修改主机名
vim /etc/hostname
kafka-broker1 #末尾追加

• 配置Linux克隆机主机名称映射hosts文件，打开/etc/hosts

# 修改主机名称映射vim /etc/hosts添加如下内容:
192.168.233.106 kafka-broker1
192.168.233.107 kafka-broker2

• 重启克隆机kafka-broker1

# 重启
reboot

• 其他机器也同样配置

以下操作默认在kafka-broker1里操作

创建xsync文件

cd /root
mkdir bin
cd bin
vim xsync

• 增加如下内容

#!/bin/bash#1. 判断参数个数
if [ $# -lt 1 ]
thenecho Not Enough Arguement!exit;
fi#2. 遍历集群所有机器
for host in kafka-broker1 kafka-broker2
doecho ====================  $host  ====================#3. 遍历所有目录，挨个发送for file in $@do#4 判断文件是否存在if [ -e $file ]then#5. 获取父目录pdir=$(cd -P $(dirname $file); pwd)#6. 获取当前文件的名称fname=$(basename $file)ssh $host "mkdir -p $pdir"rsync -av $pdir/$fname $host:$pdirelseecho $file does not exists!fidone
done

• 修改xsync文件权限

# 进入/root/bin目录
cd /root/bin
# 修改权限
chmod 777 xsync

SSH无密登录配置

分发文件时，需要通过脚本切换主机进行指令操作，切换主机时，是需要输入密码的，每一次都输入就显得有点麻烦，所以这里以虚拟机kafka-broker1为例配置SSH免密登录(其他节点执行同样步骤即可)，配置完成后，脚本执行时就不需要输入密码了。

生成公钥和私钥

# 公钥和私钥
ssh-keygen -t rsa#然后敲（三个回车），就会生成两个文件id_rsa（私钥）、id_rsa.pub（公钥）

• 将公钥拷贝到要免密登录的目标机器上，拷贝过程需要输入目标机器密码

# ssh-copy-id 目标机器
ssh-copy-id kafka-broker1
ssh-copy-id kafka-broker2

安装JDK1.8

卸载现有JDK

• 不同节点都要执行操作

  rpm -qa | grep -i java | xargs -n1 sudo rpm -e --nodeps

上传Java压缩包

• 将jdk-8u202-linux-x64.tar.gz文件上传到虚拟机的/opt/software目录中

解压Java压缩包

• 进入/opt/software目录

  cd /opt/software/

• 解压缩文件到指定目录

  tar -zxvf jdk-8u212-linux-x64.tar.gz -C /opt/module/

• 进入/opt/module目录

  cd /opt/module

• 改名

  mv jdk1.8.0_212/ java

配置Java环境变量

• 新建 /etc/profile.d/my_env.sh文件

  vim /etc/profile.d/my_env.sh

• 添加内容

  #JAVA_HOME
  export JAVA_HOME=/opt/module/java
  export PATH=$PATH:$JAVA_HOME/bin
  

• 让环境变量生效

  source /etc/profile.d/my_env.sh

安装测试

java -version

分发软件

• 分发环境变量文件

  xsync /etc/profile.d/my_env.sh

• 进入/opt/module路径

  cd /opt/module

• 调用分发脚本将本机得Java安装包分发到其他两台机器

  xsync java

在每个节点让环境变量生效

安装ZooKeeper

上传ZooKeeper压缩包

将apache-zookeeper-3.7.2-bin.tar.gz文件上传到三台虚拟机的/opt/software目录中

解压ZooKeeper压缩包

• 进入到/opt/software目录中

  cd /opt/software/

• 解压缩文件到指定目录

  tar -zxvf apache-zookeeper-3.7.1-bin.tar.gz -C /opt/module/

• 进入/opt/module目录

  cd /opt/module

• 文件目录改名

  mv apache-zookeeper-3.7.1-bin/ zookeeper

配置服务器编号

• 在/opt/module/zookeeper/目录下创建zkData

• 进入/opt/module/zookeeper目录

  cd /opt/module/zookeeper

• 创建zkData文件目录

  mkdir zkData

• 创建myid文件

• 进入/opt/module/zookeeper/zkData目录

  cd /opt/module/zookeeper/zkData

• 创建myid文件

  vim myid

• 在文件中增加内容

  1

修改配置文件

• 重命名/opt/module/zookeeper/conf目录下的zoo_sample.cfg文件为zoo.cfg文件

• 进入cd /opt/module/zookeeper/conf文件目录

  cd /opt/module/zookeeper/conf

• 修改文件名称

  mv zoo_sample.cfg zoo.cfg

• 修改文件内容

  vim zoo.cfg

• 修改zoo.cfg文件

以下内容为修改内容

# **以下内容为修改内容**
dataDir=/opt/module/zookeeper/zkData#以下内容为新增内容####################### cluster ##########################
# server.A=B:C:D
#
# A是一个数字，表示这个是第几号服务器
# B是A服务器的主机名
# C是A服务器与集群中的主服务器（Leader）交换信息的端口
# D是A服务器用于主服务器（Leader）选举的端口
#########################################################server.1=kafka-broker1:2888:3888
server.2=kafka-broker2:2888:3888

设置防火墙端口

firewall-cmd --zone=public --add-port=2181/tcp --permanent
firewall-cmd --zone=public --add-port=2888/tcp --permanent
firewall-cmd --zone=public --add-port=3888/tcp --permanent
firewall-cmd --reload
firewall-cmd --zone=public --list-ports

启动ZooKeeper

在每个节点下执行如下操作

• 进入zookeeper目录

  cd /opt/module/zookeeper

• 启动ZK服务

  bin/zkServer.sh start

关闭ZooKeeper

在每个节点下执行如下操作

• 进入zookeeper目录

  cd /opt/module/zookeeper

• 关闭ZK服务

  bin/zkServer.sh stop

查看ZooKeeper状态

在每个节点下执行如下操作

• 进入zookeeper目录

  cd /opt/module/zookeeper

• 查看ZK服务状态

  bin/zkServer.sh status

分发软件

• 进入/opt/module路径

  cd /opt/module

• 调用分发脚本将本机得ZooKeeper安装包分发到其他机器

  xsync zookeeper

分别将不同虚拟机/opt/module/zookeeper/zkData目录下myid文件进行修改

vim /opt/module/zookeeper/zkData/myid
# kafka-broker1:1
# kafka-broker2:2

启停脚本

ZooKeeper软件的启动和停止比较简单，但是每一次如果都在不同服务器节点执行相应指令，也会有点麻烦，所以我们这里将指令封装成脚本文件，方便我们的调用。

• 在虚拟机kafka-broker1的/root/bin目录下创建zk.sh脚本文件

• 在/root/bin这个目录下存放的脚本，root用户可以在系统任何地方直接执行

• 进入/root/bin目录

  cd /root/bin

• 创建zk.sh脚本文件

  vim zk.sh

在脚本中增加内容:

#!/bin/bash
case $1 in
"start"){for i in kafka-broker1 kafka-broker2doecho ---------- zookeeper $i 启动 ------------ssh $i "/opt/module/zookeeper/bin/zkServer.sh start"done
};;
"stop"){for i in kafka-broker1 kafka-broker2doecho ---------- zookeeper $i 停止 ------------ssh $i "/opt/module/zookeeper/bin/zkServer.sh stop"done
};;
"status"){for i in kafka-broker1 kafka-broker2doecho ---------- zookeeper $i 状态 ------------ssh $i "/opt/module/zookeeper/bin/zkServer.sh status"done
};;
esac

增加脚本文件权限

• 给zk.sh文件授权

  chmod 777 zk.sh

脚本调用方式

• 启动ZK服务

  zk.sh start

• 查看ZK服务状态

  zk.sh status

• 停止ZK服务

  zk.sh stop

安装Kafka

上传Kafka压缩包

将kafka_2.12-3.5.0.tgz文件上传到虚拟机的/opt/software目录中

解压Kafka压缩包

• 进入/opt/software目录

  cd /opt/software

• 解压缩文件到指定目录

  tar -zxvf kafka_2.12-3.5.0.tgz -C /opt/module/

• 进入/opt/module目录

  cd /opt/module

• 修改文件目录名称

  mv kafka_2.12-3.5.0/ kafka

修改配置文件

• 进入cd /opt/module/kafka/config文件目录

  cd /opt/module/kafka/config

• 修改配置文件

  vim server.properties

• 输入以下内容：

  #broker的全局唯一编号，每个服务节点不能重复，只能是数字。
  broker.id=1
  #broker对外暴露的IP和端口 （每个节点单独配置）
  advertised.listeners=PLAINTEXT://**kafka-broker1**:9092
  #处理网络请求的线程数量
  num.network.threads=3
  #用来处理磁盘IO的线程数量
  num.io.threads=8
  #发送套接字的缓冲区大小
  socket.send.buffer.bytes=102400
  #接收套接字的缓冲区大小
  socket.receive.buffer.bytes=102400
  #请求套接字的缓冲区大小
  socket.request.max.bytes=104857600
  #kafka运行日志(数据)存放的路径，路径不需要提前创建，kafka自动帮你创建，可以配置多个磁盘路径，路径与路径之间可以用"，"分隔
  log.dirs=/opt/module/kafka/datas
  #topic在当前broker上的分区个数
  num.partitions=1
  #用来恢复和清理data下数据的线程数量
  num.recovery.threads.per.data.dir=1
  #每个topic创建时的副本数，默认时1个副本
  offsets.topic.replication.factor=1
  #segment文件保留的最长时间，超时将被删除
  log.retention.hours=168
  #每个segment文件的大小，默认最大1G
  log.segment.bytes=1073741824
  #检查过期数据的时间，默认5分钟检查一次是否数据过期
  log.retention.check.interval.ms=300000
  #配置连接Zookeeper集群地址（在zk根目录下创建/kafka，方便管理）
  zookeeper.connect=kafka-broker1:2181,kafka-broker2:2181/kafka
  

分发kafka软件

• 进入 /opt/module目录

  cd /opt/module

• 执行分发指令

  xsync kafka

按照上面的配置文件内容，在每一个Kafka节点进行配置

vim /opt/module/kafka/config/server.properties

配置环境变量

• 修改 /etc/profile.d/my_env.sh文件

  vim /etc/profile.d/my_env.sh

• 添加内容

  #KAFKA_HOME
  export KAFKA_HOME=/opt/module/kafka
  export PATH=$PATH:$KAFKA_HOME/bin
  

• 让环境变量生效

  source /etc/profile.d/my_env.sh

• 分发环境变量，并让环境变量生效

  xsync /etc/profile.d/my_env.sh

• 每个节点执行刷新操作

  source /etc/profile.d/my_env.sh

启动Kafka

启动前请先启动ZooKeeper服务

• 进入/opt/module/kafka目录

  cd /opt/module/kafka

• 执行启动指令

  bin/kafka-server-start.sh -daemon config/server.properties

关闭Kafka

• 进入/opt/module/kafka目录

  cd /opt/module/kafka

• 执行关闭指令

  bin/kafka-server-stop.sh

启停脚本

(1) 在虚拟机kafka-broker1的/root/bin目录下创建kfk.sh脚本文件，对kafka服务的启动停止等指令进行封装

• 进入/root/bin目录

  cd /root/bin

• 创建kfk.sh脚本文件

  vim kfk.sh

• 在脚本中增加内容:

  #! /bin/bash
  case $1 in
  "start"){for i in kafka-broker1 kafka-broker2doecho " --------启动 $i Kafka-------"ssh $i "/opt/module/kafka/bin/kafka-server-start.sh -daemon /opt/module/kafka/config/server.properties"done
  };;
  "stop"){for i in kafka-broker1 kafka-broker2doecho " --------停止 $i Kafka-------"ssh $i "/opt/module/kafka/bin/kafka-server-stop.sh "done
  };;
  esac
  

增加脚本文件权限

• 给文件授权

  chmod 777 kfk.sh

脚本调用方式

• 启动kafka

  kfk.sh start

• 停止Kafka

  kfk.sh stop

注意：停止Kafka集群时，一定要等Kafka所有节点进程全部停止后再停止ZooKeeper集群。因为Zookeeper集群当中记录着Kafka集群相关信息，Zookeeper集群一旦先停止，Kafka集群就没有办法再获取停止进程的信息，只能手动杀死Kafka进程了。

联合脚本

因为Kafka启动前，需要先启动ZooKeeper，关闭时，又需要将所有Kafka全部关闭后，才能关闭ZooKeeper，这样，操作起来感觉比较麻烦，所以可以将之前的2个脚本再做一次封装。

• 在虚拟机kafka-broker1的/root/bin目录下创建xcall脚本文件

• 进入/root/bin目录

  cd /root/bin

• 创建xcall文件

  vim xcall

• 在脚本中增加内容:

  #! /bin/bashfor i in kafka-broker1 kafka-broker2
  doecho --------- $i ----------ssh $i "$*"
  done
  

增加脚本文件权限

• 进入/root/bin目录

  cd /root/bin

• 增加权限

  chmod 777 xcall

创建cluster.sh脚本文件

• 进入/root/bin目录

  cd /root/bin

• 创建cluster.sh脚本文件

  vim cluster.sh

• 在脚本中增加内容:

  #!/bin/bashcase $1 in
  "start"){echo ================== 启动 Kafka集群 ==================#启动 Zookeeper集群zk.sh start#启动 Kafka采集集群kfk.sh start};;
  "stop"){echo ================== 停止 Kafka集群 ==================#停止 Kafka采集集群kfk.sh stop#循环直至 Kafka 集群进程全部停止kafka_count=$(xcall jps | grep Kafka | wc -l)while [ $kafka_count -gt 0 ]dosleep 1kafka_count=$(xcall | grep Kafka | wc -l)echo "当前未停止的 Kafka 进程数为 $kafka_count"done#停止 Zookeeper集群zk.sh stop
  };;
  esac
  

增加脚本文件权限

• 进入/root/bin目录

  cd /root/bin

• 增加权限

  chmod 777 cluster.sh

脚本调用方式

• 集群启动

  cluster.sh start

• 集群关闭

  cluster.sh stop