阅读量:0
1. 环境准备
环境准备 10.0.0.101，10.0.0.102，10.0.0.103， 101上做到102，103的免密登录 #为了方便执行命令和传输文件，这里先在101创建2个脚本 ]# cat > /usr/bin/data-rsync.sh <<'EOF' #!/bin/bash [ $# -ne 1 ] && { echo "usage: $0 PATH" exit 1 } for ((id=102;id<=103;id++)) do echo "---root@10.0.0.${id}----开始---" tput setaf 2 rsync -avz --delete $1 root@10.0.0.${id}:`dirname $1` tput setaf 9 echo -e "---root@10.0.0.${id}----结束---------------\n" done EOF ]# cat > /usr/bin/cmd-cluster.sh <<'EOF' #!/bin/bash [ $# -eq 0 ] && { echo "usage: $0 CMD" exit 1 } for ((id=101;id<=103;id++)) do echo "--10.0.0.${id}--开始-" tput setaf 2 ssh root@10.0.0.${id} "$@" tput setaf 9 echo -e "--10.0.0.${id}--结束--------------------\n" done EOF chmod +x /usr/bin/data-rsync.sh /usr/bin/cmd-cluster.sh 2. zookeeper相关
2.1. zookeeper单节点部署
zookeeper单节点部署 (1)下载zookeeper软件 mkdir -p /data/tools/zookeeper/ && cd /data/tools/zookeeper/ [root@elk1 ~]# wget https://dlcdn.apache.org/zookeeper/zookeeper-3.8.4/apache-zookeeper-3.8.4-bin.tar.gz (2)解压软件包 [root@elk1 ~]# tar xf apache-zookeeper-3.8.4-bin.tar.gz (3)创建符号连接 [root@elk1~]# ln -sv apache-zookeeper-3.8.4-bin zk (4)声明zk的环境变量 [root@elk1]# cat > /etc/profile.d/kafka.sh <<'EOF' export ZK_HOME=/data/tools/zookeeper/zk export PATH=$PATH:$ZK_HOME/bin EOF echo -e "\nexport PATH=$PATH" >> /root/.bashrc source /etc/profile.d/kafka.sh /root/.bashrcv (5)创建zk的配置文件 [root@elk1 ~]# cp /data/tools/zookeeper/zk/conf/{zoo_sample.cfg,zoo.cfg} (6)启动zk节点 [root@elk1 ~]# zkServer.sh start [root@elk1 ~]# zkServer.sh status # 查看zk服务的状态信息 [root@elk1 ~]# zkServer.sh stop [root@elk1 ~]# zkServer.sh restart (7)连接ZK节点 [root@elk101 ~]# zkCli.sh 2.2. zookeeper常用命令
功能 | 命令 |
创建节点(znode) | create /path data -s: 创建顺序节点 -e: 创建临时节点 |
获取znode的数据 | get /path |
设置znode的数据 | set /path data |
创建子节点 | create /paren/path/subNode data |
列出子节点 | ls /path |
查看znode的状态 | stat /path |
删除znode | delete /path |
递归删除其子节点使用 | deleteall /path |
2.3. zookeeper集群部署与管理
2.3.1. zookeeper集群部署
zookeeper集群部署 (1)101下载zookeeper软件 [root@elk1 ~]# mkdir -p /data/tools/zookeeper/ && cd /data/tools/zookeeper/ [root@elk1 ~]# wget https://dlcdn.apache.org/zookeeper/zookeeper-3.8.4/apache-zookeeper-3.8.4-bin.tar.gz （1）配置101到101,102，103的免密 [root@elk101 ~]# ssh-keygen [root@elk101 ~]# ssh-copy-id 10.0.0.101 [root@elk101 ~]# ssh-copy-id 10.0.0.102 [root@elk101 ~]# ssh-copy-id 10.0.0.103 (2)101节点解压软件包 [root@elk101 ~]# tar xf apache-zookeeper-3.8.4-bin.tar.gz (3)101节点创建符号连接 [root@elk101 ~]# ln -sv apache-zookeeper-3.8.4-bin zk (4)101节点声明zk的环境变量 [root@elk101 ~]# cat > /etc/profile.d/kafka.sh <<'EOF' export ZK_HOME=/data/tools/zookeeper/zk export PATH=$PATH:$ZK_HOME/bin EOF (5)101节点创建zk的数据目录 [root@elk101 ~]# mkdir -p /data/data/zk (6)101节点创建zk的配置文件 [root@elk101 ~]# cat > /data/tools/zookeeper/zk/conf/zoo.cfg << EOF # 定义最小单元的时间范围tick。 tickTime=2000 # 启动时最长等待tick数量。 initLimit=5 # 数据同步时最长等待的tick时间进行响应ACK syncLimit=2 # 指定数据目录 dataDir=/data/data/zk # 监听端口 clientPort=2181 # 开启四字命令允许所有的节点访问。 4lw.commands.whitelist=* # server.ID=A:B:C[:D] # ID: # zk的唯一编号。 # A: # zk的主机地址。 # B: # leader的选举端口，是谁leader角色，就会监听该端口。 # C: # 数据通信端口。 # D: # 可选配置，指定角色。 server.101=10.0.0.101:2888:3888 server.102=10.0.0.102:2888:3888 server.103=10.0.0.103:2888:3888 EOF (7)101同步相关目录到102，103节点 [root@elk101 ~]# data-rsync.sh /data/data/zk [root@elk101 ~]# data-rsync.sh /data/tools/zookeeper [root@elk101 ~]# data-rsync.sh /etc/profile.d/kafka.sh (8)所有节点创建myid文件 [root@elk101 ~]# cmd-cluster.sh do ssh 10.0.0.${host_id} "echo ${host_id} > /data/data/zk/myid"; done (9)所有节点加载zk环境变量 [root@elk101 ~]# cmd-cluster.sh "source /etc/profile.d/kafka.sh;echo export PATH=$PATH >> /root/.bashrc;source /root/.bashrc" (10)所有节点启动zk服务 [root@elk101 ~]# cmd-cluster.sh "zkServer.sh start" (11)所有节点查看zk的角色状态 [root@elk101 ~]# cmd-cluster.sh "zkServer.sh status"2.3.2. zookeeper集群管理
编写zookeeper的集群管理脚本 [root@elk101]# cat > /usr/local/sbin/zkManager.sh <<'EOF' #!/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=101 ; i<=103 ; i++ )) ; do tput setaf 2 echo ========== 10.0.0.${i} zkServer.sh $1 ================ tput setaf 9 ssh 10.0.0.${i} "zkServer.sh $1 2>/dev/null" done } #调用函数 zookeeperManger EOF #给脚本添加执行权限 [root@elk1~]# chmod +x /usr/local/sbin/zkManager.sh #使用 zkManager.sh start|stop|restart|status2.4. 修改zookeeper的堆内存大小
修改zookeeper的堆内存大小: 温馨提示: 修改zookeeper的堆内存大小，一般情况下，生产环境给到2G足以，如果规模较大可以适当调大到4G。 (1)查看zk默认的堆内存大小为1GB。 [root@elk101 ~]# jmap -heap `jps | awk '/QuorumPeerMain/{print $1}'`|grep -i heapsize (2)配置ZK的堆内存 [root@elk101 ~]# cat > /data/tools/zookeeper/zk/conf/java.env << 'EOF' export JVMFLAGS="-Xms256m -Xmx256m $JVMFLAGS" EOF (3)将配置文件同步到集群的其他zk节点上 [root@elk101 ~]# data-rsync.sh /data/tools/zookeeper/zk/conf/java.env (4)重启ZK集群 [root@elk101 ~]# zkManager.sh restart (5)验证堆内存 [root@elk101 ~]# jmap -heap `jps | awk '/QuorumPeerMain/{print $1}'`|grep -i heapsize3. kafka相关
3.1. kafka单点部署
快速搭建kafka单点环境 (1)下载kafka软件包 [root@elk101 ~]# mkdir -p /data/tools/kafka/ && cd /data/tools/kafka/ [root@elk101 ~]# wget https://downloads.apache.org/kafka/3.8.0/kafka_2.13-3.8.0.tgz (2)解压软件包 [root@elk101 ~]# tar xf kafka_2.13-3.8.0.tgz (3)创建符号连接 [root@elk101 ~]# ln -svf /data/tools/kafka/kafka_2.13-3.8.0 /data/tools/kafka/kafka (4)配置环境变量 [root@elk101 ]# cat > /etc/profile.d/kafka.sh <<'EOF' export ZK_HOME=/data/tools/zookeeper/zk export KAFKA_HOME=/data/tools/kafka/kafka export PATH=$PATH:$KAFKA_HOME/bin:$ZK_HOME/bin EOF [root@elk101 ]# source /etc/profile.d/kafka.sh [root@elk101 ]# echo "export PATH=$PATH" >> /root/.bashrc [root@elk101 ]# source /root/.bashrc (5)修改配置文件 [root@elk101 ~]# egrep '^broker|^zookeeper.connect=' /data/tools/kafka/kafka/config/server.properties broker.id=101 zookeeper.connect=10.0.0.101:2181,10.0.0.102:2181,10.0.0.103:2181 (6)启动kafka单点 [root@elk101 ]# kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties (7)验证zookeeper的源数据信息 zkCli.sh -server 10.0.0.101:2181,10.0.0.102:2181,10.0.0.103:2181 (CONNECTED) 0] ls /brokers/ids [101]3.2. kafka集群部署
kafka集群部署 (1)101节点下载kafka软件包 [root@elk101 ~]# mkdir -p /data/tools/kafka/ && cd /data/tools/kafka/ [root@elk101 ~]# wget https://downloads.apache.org/kafka/3.8.0/kafka_2.13-3.8.0.tgz (2)101节点解压软件包 [root@elk101 ~]# tar xf kafka_2.13-3.8.0.tgz (3)101节点创建符号连接 [root@elk101 ~]# ln -svf /data/tools/kafka/kafka_2.13-3.8.0 /data/tools/kafka/kafka (4)101节点配置环境变量 [root@elk101 ]# cat > /etc/profile.d/kafka.sh <<'EOF' export ZK_HOME=/data/tools/zookeeper/zk export KAFKA_HOME=/data/tools/kafka/kafka export PATH=$PATH:$KAFKA_HOME/bin:$ZK_HOME/bin EOF [root@elk101 ]# source /etc/profile.d/kafka.sh [root@elk101 ]# echo "export PATH=$PATH" >> /root/.bashrc [root@elk101 ]# source /root/.bashrc (5)101节点修改配置文件 [root@elk101 ~]# ]# egrep '^broker|^zookeeper.connect=|log.dirs' /data/tools/kafka/kafka/config/server.properties broker.id=101 log.dirs=/data/data/kafka-logs zookeeper.connect=10.0.0.101:2181,10.0.0.102:2181,10.0.0.103:2181 (6)101同步相关目录到102，103节点 [root@elk101 ~]# data-rsync.sh /data/data/kafka-logs [root@elk101 ~]# data-rsync.sh /data/tools/kafka [root@elk101 ~]# data-rsync.sh /etc/profile.d/kafka.sh (7)102,103配置文件修改 [root@elk102 ~]# sed -i 's#^broker.id.*#broker.id=102#' /data/tools/kafka/kafka/config/server.properties [root@elk103 ~]# sed -i 's#^broker.id.*#broker.id=103#' /data/tools/kafka/kafka/config/server.properties (8)所有节点加载kafka环境变量 [root@elk101 ~]# cmd-cluster.sh "source /etc/profile.d/kafka.sh; echo export PATH=$PATH >> /root/.bashrc ; source /root/.bashrc" (9)所有节点启动kafka服务 [root@elk101 ~]# cmd-cluster.sh "kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties " (10)验证zookeeper的源数据信息 zkCli.sh -server 10.0.0.101:2181,10.0.0.102:2181,10.0.0.103:2181 (CONNECTED) 0] ls /brokers/ids [101, 102, 103]3.3. kafka的常用术语
kafka的常用术语: kafka cluster（broker list）: kafka集群。 kafka Server (broker): 指的是kafka集群的某个节点。 Producer: 生产者，即往kafka集群写入数据的角色。 Consumer: 消费者，即从kafka集群中读取数据的角色。一个消费者隶属于一个消费者组。 Concumer Group: 消费者组，里面有一个或多个消费者。 Topic: 主题，是一个逻辑概念，用于区分业务，一个主题最少要有1个分区和一个副本。 Partition: 分区，分区可以暂时理解为分区编号。 replica: 副本，副本是实际存储数据的地方，分为两种角色，即leader和follower。 leader: 负责读写。 follower: 负责从leader节点同步数据，无法对集群外部提供任何服务。当leader无法访问时，follower会接管leader的角色。 AR: 所有的副本，包含leader和follower副本。 ISR: 表示和leader同步的所有副本集合。 OSR: 表示和leader不同步的所有副本即可。 zookeeper集群: kafka 0.9之前的版本维护消费者组的offset，之后kafka内部的topic进行维护。 协调kafka的leader选举，控制器协调者选举等.... client: consumer API: 即消费者，指的是从boker拉取数据的角色。 每个消费者均隶属于一个消费者组(consumer Group)，一个消费者组内可以有多个消费者。 producer API: 即生产者，指的是往broker写入数据的角色。 admin API: 集群管理的相关API，包括topic，parititon，replica等管理。 stream API: 数据流处理等API，提供给Spark，Flink，Storm分布式计算框架提供数据流管道。 connect API: 连接数据库相关的API，例如将MySQL导入到kafka。 常见问题: Q1: 分区和副本有啥区别? 分区可以暂时理解为分区编号，它包含该分区编号的所有副本，和磁盘的分区没关系。 副本是实际存储数据的地方， Q2: offset存储在kafka集群，客户端在kafka集群任意一个节点如何获取偏移量。 通过内部的消费者组的偏移量读取即可。（"__consumer_groups"）3.4. kafka topic管理命令
topic管理: -1 查看topic #查看topic列表 [root@elk101 ~]# kafka-topics.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --list #查看指定topic的详细信息 [root@elk101 ~]# kafka-topics.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --describe --topic tom #查看所有的topic详细信息。 [root@elk101 ~]# kafka-topics.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --describe -2 创建topic 一个topic是生产者(producer)和消费者(consumer)进行通信的逻辑单元。 底层存储数据的是对应一个或多个分区(partition)副本(replica)。 #创建一个名为"tom"，分区数为3，副本数量为2的topic。 [root@elk101 ~]# kafka-topics.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --create --partitions 3 --replication-factor 2 --topic tom #创建一个名为"jack"，分区数为3，副本数量为2的topic。 [root@elk101 ~]# kafka-topics.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --create --partitions 3 --replication-factor 2 --topic jack -3 修改topic（分区数量可以调大，但不可以调小!） [root@elk101 ~]# kafka-topics.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --alter --topic tom --partitions 5 -4 刪除topic [root@elk101 ~]# kafka-topics.sh --bootstrap-server 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092 --delete --topic tom -5 创建生产者 [root@elk102 ~]# kafka-console-producer.sh --bootstrap-server 10.0.0.103:9092 --topic jack -6 创建消费者 [root@elk103 ~]# kafka-console-consumer.sh --bootstrap-server 10.0.0.102:9092 --topic jack --from-beginning -7 消费者组案例 (1)创建topic [root@elk103 ~]# kafka-topics.sh --bootstrap-server 10.0.0.101:9092 --create --partitions 3 --replication-factor 2 --topic group-test-topic (2)启动生产者 [root@elk102 ~]# kafka-console-producer.sh --bootstrap-server 10.0.0.103:9092 --topic group-test-topic (3)启动消费者加入同一个消费者组 #加入消费者组方式1： [root@elk101 ~]# kafka-console-consumer.sh --bootstrap-server 10.0.0.102:9092 --topic group-test-topic --consumer-property group.id=test --from-beginning #加入消费者组方式2： [root@elk103 ~]# kafka-console-consumer.sh --bootstrap-server 10.0.0.102:9092 --topic group-test-topic --consumer.config /data/tools/kafka/kafka/config/consumer.properties --from-beginning (4)测试 在生产者端写入数据，观察消费者的输出即可。 (5)观察消费者组的详细信息 [root@elk101 ~]# kafka-consumer-groups.sh --bootstrap-server 10.0.0.101:9092 --describe --group group-test-topic3.5. kafka内存调优
kafka堆内存调优: (1)修改启动脚本 [root@elk101 ~]# vim `which kafka-server-start.sh ` +28 ... if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then ... # 将原有的注释掉 # export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G" export KAFKA_HEAP_OPTS="-server -Xmx256m -Xms256m -XX:PermSize=128m -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:ParallelGCThreads=8 -XX:ConcGCThreads=5 -XX:InitiatingHeapOccupancyPercent=70" export JMX_PORT="8888" fi (2)同步集群启动脚本 [root@elk101 ~]# data-rsync.sh `which kafka-server-start.sh` (3)重启kafka集群 [root@elk101 ~]# cmd-cluster.sh "kafka-server-stop.sh" [root@elk101 ~]# cmd-cluster.sh "kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties" (4)验证kafka集群的内存大小 [root@elk101 ~]# jmap -heap `jps | awk '/Kafka/{print $1}'`|grep MaxHeapSize [root@elk102 ~]# jmap -heap `jps | awk '/Kafka/{print $1}'`|grep MaxHeapSize [root@elk103 ~]# jmap -heap `jps | awk '/Kafka/{print $1}'`|grep MaxHeapSize3.6. kafka-eagle管理kafka
3.6.1. 启动kafka的jmx端口
-1 启动kafka的JMX端口 (1)所有节点停止kafka [root@elk101 ~]# cmd-cluster.sh "kafka-server-stop.sh" (2)所有节点修改kafka的配置文件 vim `which kafka-server-start.sh` ... # export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G" # 注视掉该行，并将下面2行复制即可 export KAFKA_HEAP_OPTS="-server -Xmx256M -Xms256M -XX:PermSize=128m -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:ParallelGCThreads=8 -XX:ConcGCThreads=5 -XX:InitiatingHeapOccupancyPercent=70" export JMX_PORT="8888" (3)所有节点启动kafka服务 [root@elk101 ~]# cmd-cluster.sh "kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties" cd /data/tools/kafka/ windows 下载https://codeload.github.com/smartloli/kafka-eagle-bin/tar.gz/refs/tags/v3.0.1 上传至101节点 3.6.2. 启动zookeeper的jmx端口
- 启动zookeeper的JMX端口 (1)修改配置文件 [root@elk101 ~]# vim /data/tools/zookeeper/zk/conf/zoo.cfg # 添加下面的一行，启动zk的4字监控命令 4lw.commands.whitelist=* (2)修改zk的启动脚本 [root@elk101 ~]# vim /data/tools/zookeeper/zk/bin/zkServer.sh +77 ... # 如果修改上面的方式不生效，则需修改zkServer.sh脚本中77行之后ZOOMAIN的值即可。 ZOOMAIN="-Dzookeeper.4lw.commands.whitelist=* ${ZOOMAIN}" (3)修改环境变量开启JMX [root@elk101 ~]# vim /data/tools/zookeeper/zk/bin/zkEnv.sh ... JMXLOCALONLY=false JMXPORT=21812 JMXSSL=false JMXLOG4J=false (4)同步脚本 [root@elk101 ~]# data-rsync.sh /data/tools/zookeeper/zk/conf/zoo.cfg [root@elk101 ~]# data-rsync.sh /data/tools/zookeeper/zk/bin/zkServer.sh [root@elk101 ~]# data-rsync.sh /data/tools/zookeeper/zk/bin/zkEnv.sh (5)重启zookeeper [root@elk101 ~]# zkManager.sh restart (6)windows jconsole验证是否能连接JMX端口3.6.3. 安装mysql,并配置
yum -y install mariadb-server #配置mariadb的配置文件 # 关闭MySQL的反向解析功能 echo "skip-name-resolve=1" >> /etc/my.cnf skip-name-resolve=1 systemctl enable mariadb.service --now #创建数据库，授权用户 CREATE DATABASE efka DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; GRANT ALL ON efka.* TO 'efka_admin'@'%' IDENTIFIED BY '123456'; SHOW GRANTS FOR efka_admin; #测试用户 mysql -u efka_admin -p123456 -h 10.0.0.1023.6.4. 安装kafka-eagle监控
#1 下载压缩包，上传，解压 windows下载https://codeload.github.com/smartloli/kafka-eagle-bin/tar.gz/refs/tags/v3.0.1 [root@elk101 ~]# cd /data/tools/kafka/ [root@elk101 ~]# tar -xvf kafka-eagle-bin-3.0.1.tar.gz [root@elk101 ~]# cd kafka-eagle-bin-3.0.1 [root@elk101 ~]# tar -xvf efak-web-3.0.1-bin.tar.gz [root@elk101 ~]# mv efak-web-3.0.1 /data/tools/kafka [root@elk101 ~]# cp /data/tools/kafka/efak-web-3.0.1/conf/{system-config.properties,.back} #2 修改配置 [root@elk101 ~]# cat > /data/tools/kafka/efak-web-3.0.1/conf/system-config.properties <<EOF efak.zk.cluster.alias=prod prod.zk.list=10.0.0.101:2181,10.0.0.102:2181,10.0.0.103:2181 prod.efak.broker.size=20 kafka.zk.limit.size=32 efak.webui.port=8048 prod.efak.offset.storage=kafka prod.efak.jmx.uri=service:jmx:rmi:///jndi/rmi://%s/jmxrmi efak.metrics.charts=true efak.metrics.retain=15 efak.sql.topic.records.max=5000 efak.sql.topic.preview.records.max=10 efak.topic.token=admin efak.driver=com.mysql.cj.jdbc.Driver efak.url=jdbc:mysql://10.0.0.101:3306/efka?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull efak.username=efka_admin efak.password=123456 EOF #3 配置环境变量 cat >> /etc/profile.d/kafka.sh <<'EOF' export KE_HOME=/data/tools/kafka/efak-web-3.0.1/ export PATH=$PATH:$KE_HOME/bin EOF source /etc/profile.d/kafka.sh #4 修改堆内存大小 [root@elk101 ~]# vim `which ke.sh ` ... export KE_JAVA_OPTS="-server -Xmx256m -Xms256m -XX:MaxGCPauseMillis=20 -XX:+UseG1GC -XX:MetaspaceSize=128m -XX:InitiatingHeapOccupancyPercent=35 -XX:G1HeapRegionSize=16M -XX:MinMetaspaceFreeRatio=50 -XX:MaxMetaspaceFreeRatio=80" #5 启动服务 [root@elk101 ~]# ke.sh start #6 登录界面 http://10.0.0.101:8048 Account:admin ,Password:123456 