KAFKA部署

1.单点部署zookeeper

1.准备zookeeper和kafka的安装包 部署kafka需要准备Java运行时环境
http://kafka.apache.org/downloads #kafka下载官方地址
kafka_2.12-2.5.1.tgz
zookeeper-3.4.12.tar.gz
jdk-8u191-linux-x64.tar.gz

2.创建工作目录
[root@kafka108 ~]# mkdir /zy/{data/{kafka,zookeeper},logs/{kafka,zookeeper},app} -p

3.上传安装包
[root@kafka108 ~]# cd /zy/app
[root@kafka108 app]# ll
-rw-r--r-- 1 root root  61709988 Jun 13 15:18 kafka_2.12-2.5.1.tgz
-rw-r--r-- 1 root root  36667596 Jun 13 15:18 zookeeper-3.4.12.tar.gz
-rw-r--r-- 1 root root 191753373 Jun 11 18:43 jdk-8u191-linux-x64.tar.gz

4.解压软件
[root@kafka108 app]# tar xf jdk-8u191-linux-x64.tar.gz 
[root@kafka108 app]# tar xf zookeeper-3.4.12.tar.gz 

5.配置jdk环境变量
[root@kafka108 app]# vim  /etc/profile.d/jdk.sh 
#!/bin/bash
export JAVA_HOME=/zy/app/jdk1.8.0_191
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=$PATH:$JAVA_HOME/bin
[root@kafka108 app]# source  /etc/profile.d/jdk.sh  #使环境变量生效

6.验证java
[root@kafka108 app]# java -version
java version "1.8.0_191"
Java(TM) SE Runtime Environment (build 1.8.0_191-b12)
Java HotSpot(TM) 64-Bit Server VM (build 25.191-b12, mixed mode)

7.创建配置文件
[root@kafka108 conf]# cp /zy/app/zookeeper-3.4.12/conf/{zoo_sample,zoo}.cfg 

8.修改配置文件
[root@kafka108 ~]# vim /zy/app/zookeeper-3.4.12/conf/zoo.cfg 
dataDir=/zy/data/zookeeper   #定义数据目录

9.定义zookeeper环境变量并启动
[root@kafka108 bin]# vim /etc/profile.d/zk.sh
#!/bin/bash
export ZK_HOME=/zy/app/zookeeper-3.4.12
export PATH=$PATH:$ZK_HOME/bin
[root@kafka108 bin]# source /etc/profile.d/zk.sh
[root@kafka108 ~]# zkServer.sh start  # 启动zookeeper服务
[root@kafka108 ~]# zkServer.sh status  # 查看zookeeper服务状态
[root@kafka108 ~]# zkServer.sh stop  # 停止zookeeper服务
[root@kafka108 ~]# zkServer.sh restart  # 重启zookeeper服务

10.确定zookeeper服务后，测试zookeeper服务是否可用
[root@kafka108 ~]# zkCli.sh 
...
[zk: localhost:2181(CONNECTED) 0] ls /
[zookeeper]
[zk: localhost:2181(CONNECTED) 1] 
[root@kafka108 ~]# ss -ntl|grep 2181
[root@kafka108 ~]# jps
16725 Jps
16622 QuorumPeerMain

2.集群部署zookeeper

配置集群免密登录及同步脚本:
1.修改主机列表
[root@kafka108 ~]# cat /etc/hosts
10.0.0.108 kafka108
10.0.0.109 kafka109
10.0.0.110 kafka110

2.kafka108节点上生成密钥对
[root@kafka108 ~]# ssh-keygen -t rsa -P '' -f ~/.ssh/id_rsa -q

3.kafka108配置所有集群节点的免密登录
[root@kafka108 ~]# for ((host_id=108;host_id<=110;host_id++));do ssh-copy-id kafka${host_id} ;done

4.链接测试
[root@kafka108 ~]# ssh kafka110 
Last login: Wed Jun 26 20:55:36 2024 from 10.0.0.1

5.所有节点安装rsync数据同步工具
[root@kafka108 ~]# yum -y install rsync
[root@kafka109 ~]# yum -y install rsync
[root@kafka110 ~]# yum -y install rsync

6.编写同步脚本
[root@kafka108 ~]# vim /usr/local/sbin/data_rsync.sh
#!/bin/bash
# Auther: Zhu Yang

if  [ $# -ne 1 ];then
   echo "Usage: $0 /path/to/file(绝对路径)"
   exit
fi 

# 判断文件是否存在
if [ ! -e $1 ];then
    echo "[ $1 ] dir or file not find!"
    exit
fi

# 获取父路径
fullpath=`dirname $1`

# 获取子路径
basename=`basename $1`

# 进入到父路径
cd $fullpath

for ((host_id=109;host_id<=110;host_id++))
  do
    # 使得终端输出变为绿色
    tput setaf 2
    echo ===== rsyncing kafka${host_id}: $basename =====
    # 使得终端恢复原来的颜色
    tput setaf 7
    # 将数据同步到其他两个节点
    rsync -apz $basename  `whoami`@kafka${host_id}:$fullpath
    if [ $? -eq 0 ];then
      echo "命令执行成功!"
    fi
done

[root@kafka108 ~]# chmod +x /usr/local/sbin/data_rsync.sh  #给脚本授权

7.修改配置文件
[root@kafka108 ~]# vim /zy/app/zookeeper-3.4.12/conf/zoo.cfg
# 定义最小单元的时间范围tick。
tickTime=2000
# 启动时最长等待tick数量。
initLimit=10
# 数据同步时最长等待的tick时间进行响应ACK
syncLimit=5
# 指定数据目录
dataDir=/zy/data/zookeeper
#指定日志目录
dataLogDir=/zy/logs/zookeeper
# 监听端口
clientPort=2181
# 开启四字命令允许所有的节点访问。
4lw.commands.whitelist=*
# server.ID=A:B:C[:D]
# ID:
#    zk的唯一编号。
# A:
#    zk的主机地址。
# B:
#    leader的选举端口，是谁leader角色，就会监听该端口。
# C: 
#    数据通信端口。
# D:
#    可选配置，指定角色。
server.0=10.0.0.108:2888:3888
server.1=10.0.0.109:2888:3888
server.2=10.0.0.110:2888:3888

8.同步zookeeper软件包及目录结构
[root@kafka108 app]# data_rsync.sh /zy/app/jdk1.8.0_191/
[root@kafka108 app]# data_rsync.sh /etc/profile.d/jdk.sh 
[root@kafka108 app]# data_rsync.sh /etc/profile.d/zk.sh 

9.去其他节点使环境变量生效
[root@kafka109 ~]# source /etc/profile.d/jdk.sh 
[root@kafka110 ~]# source /etc/profile.d/jdk.sh 
[root@kafka109 ~]# source /etc/profile.d/zk.sh 
[root@kafka110 ~]# source /etc/profile.d/zk.sh 

10.每个zookeeper节点创建唯一标识ID
[root@kafka108 ~]# vim /zy/data/zookeeper/myid
0
[root@kafka109 ~]# vim /zy/data/zookeeper/myid
1
[root@kafka110 ~]# vim /zy/data/zookeeper/myid
2

11.编写zookeeper集群管理脚本
[root@kafka108 ~]# vim /usr/local/sbin/manager_zk.sh
#!/bin/bash

#判断用户是否传参
if [ $# -ne 1 ];then
    echo "无效参数，用法为: $0  {start|stop|restart|status}"
    exit
fi

#获取用户输入的命令
cmd=$1

#定义函数功能
function zookeeperManger(){
    case $cmd in
    start)
        echo "启动服务"        
        remoteExecution start
        ;;
    stop)
        echo "停止服务"
        remoteExecution stop
        ;;
    restart)
        echo "重启服务"
        remoteExecution restart
        ;;
    status)
        echo "查看状态"
        remoteExecution status
        ;;
    *)
        echo "无效参数，用法为: $0  {start|stop|restart|status}"
        ;;
    esac
}


#定义执行的命令
function remoteExecution(){
    for (( i=108 ; i<=110 ; i++ )) ; do
            tput setaf 2
            echo ========== kafka${i} zkServer.sh  $1 ================
            tput setaf 9
            ssh kafka${i}  "source /etc/profile.d/zk.sh; zkServer.sh $1 2>/dev/null"
    done
}

#调用函数
zookeeperManger
[root@kafka108 ~]# chmod +x /usr/local/sbin/manager_zk.sh

12.启动zookeeper集群
[root@kafka108 ~]# manager_zk.sh start
启动服务
========== kafka108 zkServer.sh start ================
Starting zookeeper ... STARTED
========== kafka109 zkServer.sh start ================
Starting zookeeper ... STARTED
========== kafka110 zkServer.sh start ================
Starting zookeeper ... STARTED

13.查看集群状态
[root@kafka108 zookeeper-3.4.12]# manager_zk.sh status
查看状态
========== kafka108 zkServer.sh status ================
Mode: follower
========== kafka109 zkServer.sh status ================
Mode: leader
========== kafka110 zkServer.sh status ================
Mode: follower

14.zk的leader选举流程
    myid:
        表示当前zookeeper server的server id，用于唯一表示当前的服务器。

    zxid:
        表示zookeeper transaction id，即zookeeper集群事务ID。
leader的选举流程如下:
    (1)启动kafka108和kafka109这两个节点，先检查有没有leder节点     
        此时2个节点进入looking状态，它们认为此时集群没有leader。

    (2)若没有leader角色kafka108和kafka109这两个节点会投自己一票:      
        kafka108节点启动zk服务，开始发起投票 ---> [myid,zxid] --->[108, 0]      
        kafka109节点启动zk服务，开始发起投票 ---> [myid,zxid] --->[109, 0]

    (3)两个节点开始接收彼此的投票信息，并处理投票进行如下的比较:      
        1)优先检查zxid，即事务ID，谁的zxid较大优先成为leader;      
        2)如果zxid相同，那么就比较myid，较大的myid作为leader服务器;      
        一旦确定leader角色，此时leader服务器的状态变更为leading。      
        没有成为leader角色的节点，会主动变更为following状态。

    (4)kafka110节点启动，先检查有没有leader，发现集群leader在上一步已经存在      
        server.0=10.0.0.108:2888:3888      
        server.1=10.0.0.109:2888:3888      
        server.2=10.0.0.110:2888:3888      
        由于leader角色已经选出来了，此时会主动变更为"following状态"

    (5)kafka110主从将自己的角色变更为follwer。      
        集群半数以上节点投票完毕就已经确定了leader角色，无需全票通过。
        
        
        
zookeeper基于zab协议写入数据：
    zookeeper原子广播协议，即Zookeeper Atomic Broadcast(简称"ZAB")。

    zookeeper是通过zab协议来保证分布式事务的最终一致性。

    zab工作流程如下:
        (1)客户端如果对follower节点发起写操作;
        (2)follower节点会将请求转发给leader;
        (3)leader生成一个新的事务并为这个事务生成一个分布式唯一的ZXID，leader将这个事物提议(propose)发送给所有的followers节点;
        (4)所有的follower角色开始写入数据，并将写入结果返回给leader，即响应ACK;
        (5)leader节点收到ACK后，需要当前集群半数以上的ACK写入成功，则提交本次事务;
        (6)数据写入成功则响应客户端;
        
        
    
    
    
zookeeper的客户端常用命令:
    1.连接客户端
zkCli.sh # 连接本地
zkCli.sh -server kafka108:2181  # 连接指定节点
zkCli.sh -server kafka108:2181,kafka109:2181,kafka110:2181  # 连接集群

    2.zookeeper的基本使用
- 查:
    - ls /
        查看/路径下有多少个znode。(可以理解为linux目录)
        
    - get /linux86
        查看"/linux86"是否有数据。
    
    - stat /school
        查看/school的znode状态信息。
    
- 增:
    - create /linux86
        在"/"路径下创建一个linux86的znode。
        
    - create /school www.oldboyedu.com
        在创建"/school"同时写入数据。

    -  create /school/class linux86
        在"/school"路径下，创建一个名为"class"的znode，其数据为linux86。
    
    - create -s /school/class 123
        创建"/school/class"为前缀，
                
- 改:
    - set /linux86 123
        修改"/linux86"的znode的数据。
            
- 删:
    - delete /linux86
        删除"/linux86"的znode。
    
    - deleteall /school
        递归删除"/school"下的所有znode数据。

3.kafka单点部署

1.解压软件包
[root@kafka108 app]# tar xf kafka_2.12-2.5.1

2.配置环境变量
[root@kafka108 app]# vim /etc/profile.d/kafka.sh
#!/bin/bash

export KAFKA_HOME=/zy/app/kafka_2.12-2.5.1
export PATH=$PATH:${KAFKA_HOME}/bin

[root@kafka108 app]# source /etc/profile.d/kafka.sh

3.修改配置文件
[root@kafka108 app]# vim $KAFKA_HOME/config/server.properties 
...
# 指定broker的唯一ID
broker.id=0
...
# 指定数据目录
log.dirs=/zy/logs/kafka
...
# 指定ZK的集群并指定chroot目录
zookeeper.connect=kafka108:2181,kafka109:2181,kafka110:2181/kafka

4.启动kafka服务
[root@kafka108 app]# kafka-server-start.sh $KAFKA_HOME/config/server.properties 
[root@kafka108 app]# kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties #后台启动

4.kafka集群部署

1.同步kafka程序包到其他节点
[root@kafka108 ~]# data_rsync.sh /zy/app/kafka_2.12-2.5.1
[root@kafka108 ~]# data_rsync.sh /etc/profile.d/kafka.sh

2.修改其他节点的配置文件
[root@kafka109 ~]# vim $KAFKA_HOME/config/server.properties
...
broker.id=1
[root@kafka110 ~]# vim $KAFKA_HOME/config/server.properties
...
broker.id=2

3.各节点启动服务
[root@kafka108 ~]# for ((host_id=108;host_id<=110;host_id++)) do ssh kafka${host_id} "kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties ";done

5.使用kafka连接工具测试

我采用的连接工具是kafka-tools

6.kafka topic管理

1.增
(1) 创建一个名为"oldboyedu-test02"的topic
[root@kafka108 bin]# kafka-topics.sh --bootstrap-server 10.0.0.108:9092,10.0.0.109:9092 --create --topic oldboyedu-test02
Created topic oldboyedu-test01.
(2) 创建一个名为"oldboyedu-test03"的topic，分区数为3
[root@kafka108 bin]# kafka-topics.sh --bootstrap-server 10.0.0.108:9092,10.0.0.109:9092 --create --topic oldboyedu-test03 --partitions 3
Created topic oldboyedu-test03.
(3) 创建一个名为"oldboyedu-test04"的topic，分区数为3，副本数为2
[root@kafka108 bin]#  kafka-topics.sh --bootstrap-server 10.0.0.108:9092,10.0.0.109:9092 --create --topic oldboyedu-test04 --partitions 3 --replication-factor 2
Created topic oldboyedu-test04.

2.查
(1) 查看topic列表
[root@kafka108 bin]# kafka-topics.sh --bootstrap-server 10.0.0.108:9092,10.0.0.109:9092 --list
oldboyedu-test02
oldboyedu-test03
oldboyedu-test04
(2) 查看所有topic的详细信息 
[root@kafka108 bin]# kafka-topics.sh --bootstrap-server 10.0.0.108:9092,10.0.0.109:9092 --describe
(3) 查看指定topic的详细信息
[root@kafka108 bin]# kafka-topics.sh --bootstrap-server 10.0.0.108:9092,10.0.0.109:9092 --describe --topic oldboyedu-test03

4.改
(1) 将"oldboyedu-test02"的topic分区数增加为5个
[root@kafka108 bin]# kafka-topics.sh --bootstrap-server 10.0.0.108:9092,10.0.0.109:9092 --alter --topic oldboyedu-test02 --partitions 5

5.删
(1) 删除一个名为"oldboyedu-test03"的topic
[root@kafka108 bin]# kafka-topics.sh --bootstrap-server 10.0.0.108:9092,10.0.0.109:9092 --delete --topic oldboyedu-test03

7.生产者和消费者

1.启动生产者
[root@kafka108 bin]# kafka-console-producer.sh --bootstrap-server 10.0.0.108:9092 --topic oldboyedu-test02
>n
>sncj     
>lixiaolong

2. 启动消费者 (表示从头开始拉取数据)
(1) 从头开始拉取数据
[root@kafka108 bin]# kafka-console-consumer.sh --bootstrap-server 10.0.0.108:9092 --topic oldboyedu-test02 --from-beginning
n
sncj
lixiaolong

(2) 表示从最新的offset拉取数据
[root@kafka108 bin]# kafka-console-consumer.sh --bootstrap-server 10.0.0.108:9092 --topic oldboyedu-test02 

(3) kafka消费者组管理
[root@kafka108 bin]# kafka-consumer-groups.sh --bootstrap-server  10.0.0.108:9092 --list       #列出所有的消费者组信息
[root@kafka108 bin]# kafka-consumer-groups.sh --bootstrap-server  10.0.0.108:9092 --describe --all-groups  
 #列出所有消费者组的详细信息，包括偏移量，消费者ID，LEO等信息
 
 kafka-console-consumer.sh --bootstrap-server 10.0.0.108:9092,10.0.0.109:9092 --topic __consumer_offsets --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter"   --from-beginning
 #查看内置topic的offset数据
 
 (4) 基于配置文件指定消费者组
 [root@kafka108 ~]# vim  $KAFKA_HOME/config/consumer.properties 
...
# 指定消费者组的名称
group.id=oldboyedu-test02

[root@kafka108 ~]# kafka-console-consumer.sh --bootstrap-server 10.0.0.108:9092 --topic oldboyedu-test02 --consumer.config $KAFKA_HOME/config/consumer.properties

(5) 基于命令行参数指定消费者组
[root@kafka108 ~]#  kafka-console-consumer.sh --bootstrap-server 10.0.0.108:9092 --topic oldboyedu-test02 --consumer-property group.id=oldboyedu-test02

8.配置身份验证

1.修改配置文件
(1) 分别在三台服务修改kafka配置文件server.properties
[root@kafka108 config]# vim server.properties
#增加以下内容：
advertised.listeners=SASL_PLAINTEXT://10.0.0.108:9092   #这里ip地址是设置本服务器的ip，修改其他两节点需要修改成本机的ip
listeners=SASL_PLAINTEXT://10.0.0.108:9092              #这里ip地址是设置本服务器的ip，修改其他两节点需要修改成本机的ip
#使用的认证协议
security.inter.broker.protocol=SASL_PLAINTEXT
#SASL机制
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
#完成身份验证的类
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
#如果没有找到ACL（访问控制列表）配置，则允许任何操作。
allow.everyone.if.no.acl.found=true
#需要开启设置超级管理员, 开启以下命令
super.users=User:admin

(2) 修改配置文件producer.properties
[root@kafka108 config]# vim producer.properties
#增加以下内容：
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

(3) 修改配置文件consumer.properties
[root@kafka108 config]# vim consumer.properties
#增加以下内容：
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

2.新增配置文件
(1) 新增 kafka_server_jaas.conf配置文件
[root@kafka108 config]# vim kafka_server_jaas.conf
KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
        username="admin"
        password="Gfgopdb@2022"
        user_admin="Gfgopdb@2022"
        user_fms="Gfgopdb@2022";
};

(2) 新增 kafka_client_jaas.conf配置文件
[root@kafka108 config]# vim kafka_client_jaas.conf
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="Gfgopdb@2022"
    user_admin="Gfgopdb@2022"
    user_fms="Gfgopdb@2022";
};

3.修改启动脚本
    (1) 修改配置文件 kafka-server-start.sh
[root@kafka108 bin]# vim kafka-server-start.sh
#在文件最上方增加以下内容：
if [ "x$KAFKA_OPTS"  ]; then
    export KAFKA_OPTS="-Djava.security.auth.login.config=/zy/app/kafka_2.12-2.5.1/config/kafka_server_jaas.conf"
fi

(2) 修改配置文件 kafka-console-producer.sh
[root@kafka108 bin]# vim kafka-console-producer.sh
#在文件最上方增加以下内容：
if [ "x$KAFKA_OPTS"  ]; then
    export KAFKA_OPTS="-Djava.security.auth.login.config=/zy/app/kafka_2.12-2.5.1/config/kafka_client_jaas.conf"
fi

4.启动kafka
(1) 要确保zookeeper集群已经启动

(2) 分别在三台服务器停掉之前启动的kafka
[root@kafka108 config]# ps -ef|grep -v grep|grep kafka|awk '{print $2}'|xargs kill -9

(3) 各节点启动kafka
[root@kafka108 config]# for ((host_id=108;host_id<=110;host_id++)) do ssh kafka${host_id} "kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties ";done

9.重新在kafka连接工具测试

username="fms" password="Gfgopdb@2022"      #用户密码

在连接工具里面有个JAAS Config文件里面增加以下内容：
org.apache.kafka.common.security.plain.PlainLoginModule required username="fms" password="Gfgopdb@2022";