Kafka3.8.0+Centos7.9的安装参考
- 环境准备
操作系统版本：centos7.9
用户/密码：root/1qazXSW@
| 主机名 | IP地址 | 安装软件 | 下载地址 |
| k1 | 192.168.207.131 |
| 1） Java Downloads | Oracle 2） Apache ZooKeeper 3） Apache Kafka 4） EFAK |
| k2 | 192.168.207.132 |
| |
| k3 | 192.168.207.133 |
|
以下三台机器都操作：
1.1修改/etc/hosts
127.0.0.1 localhost localhost.localdomain localhost4 localhost4.localdomain4
::1 localhost localhost.localdomain localhost6 localhost6.localdomain6
192.168.207.131 k1
192.168.207.132 k2
192.168.207.133 k3
1.2关闭防火墙：
systemctl stop firewalld
systemctl disable firewalld
1.3禁用selinux
setenforce 0# 查看临时禁用是否生效
getenforcePermissive
永久修改 /etc/selinux/config 并将 SELINUX=disabled
1.4文件描述符调整
#编辑配置文件 /etc/security/limits.conf (永久生效)
* - nofile 100000
#命令行执行（当前会话生效）
ulimit -n 100000
1.5进程可以使用的最大内存映射区域数
Kafka 之所以吞吐量高，其中的一个重要原因就是因为 Kafka 在 Consumer 读取日志文件时使用了 mmap 的方式。mmap 将磁盘文件映射到内存，支持读和写，对内存的操作会反映在磁盘文件上。当客户端读取 Kafka 日志文件时，在进行 log 文件读取的时候直接将 log 文件读入用户态进行缓存，绕过了内核态的 Page Cache，避免了内核态和用户态的切换。
我们可以通过以下方式修改进程可以使用的最大内存映射区域数。
#编辑配置文件 /etc/sysctl.conf (永久生效)
vm.max_map_count=262144
编辑完文件后命令行执行 sysctl -p 立即永久生效
#命令行执行
sysctl -w vm.max_map_count=262144 (当前会话生效)
2.安装步骤
2.1安装jdk1.8
下载地址：
Java Downloads | Oracle
[root@k1 kafka-web]# java -version
openjdk version "1.8.0_412"
OpenJDK Runtime Environment (build 1.8.0_412-b08)
OpenJDK 64-Bit Server VM (build 25.412-b08, mixed mode)
[root@k1 kafka-web]# echo $JAVA_HOME
/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.412.b08-1.el7_9.x86_64
[root@k1 kafka-web]#
/etc/profile文件末尾添加如下环境变量，并执行source /etc/profile生效：
export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.412.b08-1.el7_9.x86_64
export PATH=$PATH:$JAVA_HOME/bin
export KE_HOME=/opt/kafka-web/efak-web-3.0.1
export PATH=$PATH:$KE_HOME/bin
2.2安装zookeeper
Zookeeper特点：
1）Zookeeper：一个领导者（Leader），多个跟随者（Follower）组成的集群。
2） Zookeepe集群中只要有半数以上节点存活，Zookeeper集群就能正常服务。所以Zookeeper适合安装奇数台服务器。
3）全局数据一致：每个Server保存一份相同的数据副本，Client无论连接到哪个Server，数据都是一致的。
4）更新请求顺序执行，来自同一个Client的更新请求按其发送顺序依次执行，即先进先出。
5）数据更新原子性，一次数据更新要么成功，要么失败。
6）实时性，在一定时间范围内，Client能读到最新数据。
2.2.1下载并解压到指定路径：
下载介质：
Apache ZooKeeper
提前创建如下目录：
mkdir –p /opt/zookeeper
mkdir –p /opt/zookeeper/zkdata
mkdir –p /opt/zookeeper/zkdatalog
cd /opt/zookeeper
tar -zxvf zookeeper-xxx.tar.gz
2.2.2修改配置文件zoo.cfg：
[root@k1 conf]# pwd
/opt/zookeeper/zookeeper-3.9.2/conf
mv zoo_sample.cfg zoo.cfg并修改如下:红色的内容
[root@k1 conf]# cat zoo.cfg |grep -v "^$"|grep -v "#"
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/opt/zookeeper/zkdata
dataLogDir=/opt/zookeeper/zkdatalog
clientPort=12181
server.1=192.168.207.131:12888:13888
server.2=192.168.207.132:12888:13888
server.3=192.168.207.133:12888:13888
autopurge.purgeInterval=1
autopurge.snapRetainCount=3
#server.1 这个1是服务器的标识也可以是其他的数字， 表示这个是第几号服务器，用来标识服务器，这个标识要写到快照目录下面myid文件里
#192.168.207.131为集群里的IP地址，第一个端口是master和slave之间的通信端口，默认是2888，当前修改为12888，第二个端口是leader选举的端口，集群刚启动的时候选举或者leader挂掉之后进行新的选举的端口默认是3888，当前修改为13888
2.2.3创建myid文件
三个节点分别创建1，2，3，必须不同数值。
#k1
echo "1" > /opt/zookeeper/zkdata/myid
#k2
echo "2" > /opt/zookeeper/zkdata/myid
#k3
echo "3" > /opt/zookeeper/zkdata/myid
2.2.4启动服务
3台都操作：
#进入到Zookeeper的bin目录下
[root@k1 conf]# cd /opt/zookeeper/zookeeper-3.9.2/bin
#启动服务
[root@k1 bin]#./zkServer.sh start
#停服务
[root@k1 bin]#./zkServer.sh stop
#查看状态
[root@k1 bin]# ./zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/zookeeper-3.9.2/bin/../conf/zoo.cfg
Client port found: 12181. Client address: localhost. Client SSL: false.
Mode: follower #它是否为领导
zk集群一般只有一个leader，多个follower，主一般是相应客户端的读写请求，而从主同步数据，当主挂掉之后就会从follower里投票选举一个leader出来。
[root@k1 bin]# jps
9719 QuorumPeerMain
25384 Jps
2.3安装kafka
Kafka是一个高性能、分布式的消息队列系统，用于实时数据传输和处理。在实际应用中，为了保证高可用性和容错性，通常会部署一个由多个节点组成的Kafka集群。其中，Kafka集群至少需要3个节点，这是为了保证集群的稳定性和可靠性。
## 为什么需要三个节点：
- **容错性**：当一个节点发生故障时，集群仍然能够正常运行。如果只有两个节点，当其中一个节点宕机时，另一个节点无法构成多数派，整个集群无法正常工作。因此，至少需要三个节点，即使有一个节点宕机，仍然可以保证集群的正常运行。
- **选举机制**：Kafka使用Zookeeper来进行节点的选举，保证集群的一致性。在一个多节点的Kafka集群中，当有节点宕机时，剩余节点可以通过选举机制选择新的Leader节点来维持集群的正常运行。
- **数据复制**：Kafka采用副本机制来保证数据的可靠性和高可用性。通过在不同节点上保存数据的副本，即使有节点宕机，数据仍然可以被访问。当只有两个节点时，无法进行数据副本的分布，容易造成数据丢失。
2.3.1下载并解压到指定路径：
Apache Kafka
#创建目录
cd /opt/
mkdir kafka #创建项目目录
cd kafka
mkdir -p /opt/kafka/kafkalogs #创建kafka消息目录，主要存放kafka消息
#解压软件
tar -xaf kafka_2.13-3.8.0.tgz
2.3.2修改配置文件server.properties：
进入到config目录
cd /opt/kafka/kafka_2.13-3.8.0/config/
主要关注：server.properties 这个文件即可
[root@k1 config]# pwd
/opt/kafka/kafka_2.13-3.8.0/config
主要修改红色的部分内容：
[root@k1 config]# cat server.properties |grep -v "^$"|grep -v "#"
broker.id=1 ##另外两台是2和3，不能重复
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=/opt/kafka/kafkalogs
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.retention.hours=168
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.207.131:12181,192.168.207.132:12181,192.168.207.133:12181
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0
说明：
broker.id=0 #当前机器在集群中的唯一标识，和zookeeper的myid性质一样
port=9092 #当前kafka对外提供服务的端口默认是9092
log.dirs=/opt/kafka/kafkalogs/ #消息存放的目录，这个目录可以配置为“，”逗号分割的表达式，上面的num.io.threads要大于这个目录的个数这个目录，如果配置多个目录，新创建的topic他把消息持久化的地方是，当前以逗号分割的目录中，那个分区数最少就放那一个
参考文档：Kafka【第一篇】Kafka集群搭建-CSDN博客
2.3.3启动Kafka集群并测试
1)启动服务（3台都需要操作和启动）
cd
/opt/kafka/kafka_2.13-3.8.0/bin #进入到kafka的bin目录
##修改kafka-server-start.sh防止提示内存不足以及JMX连接的问题：
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
export KAFKA_HEAP_OPTS="-server -Xms2G -Xmx2G -XX:PermSize=128m -XX:+UseG1GC -XX:MaxGCPauseMillis=200 -XX:ParallelGCThreads=8 -XX:ConcGCThreads=5 -XX:InitiatingHeapOccupancyPercent=70"
export JMX_PORT="9999"
fi
#从后台启动Kafka集群
./kafka-server-start.sh -daemon ../config/server.properties
停服务：./kafka-server-stop.sh -daemon ../config/server.properties
2)检查服务是否启动
#执行命令jps
20348 Jps
4233 QuorumPeerMain
18991 Kafka
3)创建Topic来验证是否创建成功
#创建Topic
[root@k1 bin]# ./kafka-topics.sh --create --bootstrap-server 192.168.207.131:9092 --replication-factor 2 --partitions 2 --topic topic_jyc01
WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.
Created topic topic_jyc01.
[root@k1 bin]# ./kafka-topics.sh --list --bootstrap-server 192.168.207.132:9092
topic_jyc01
#解释
--replication-factor 2 #复制两份
--partitions 2 #创建2个分区
--topic #主题为topic_jyc01
生产者：
[root@k1 bin]# ./kafka-console-producer.sh --broker-list "192.168.207.131:9092,192.168.207.132:9092,192.168.207.133:9092" --topic topic_jyc01
>Hello Kafka!
>你好 kafka!
>test
>
消费者：
[root@k1 bin]# ./kafka-console-consumer.sh --bootstrap-server "192.168.207.131:9092,192.168.207.132:9092,192.168.207.133:9092" --topic topic_jyc01
Hello Kafka!
你好 kafka!
test
可正常收到信息:
2.4安装Kafka Eagle
Kafka Eagle是一款 Kafka 可视化和管理软件，支持对多个不同版本的 Kafka 集群进行管理。Kafka Eagle可以监控 Kafka 集群的健康状态，消费者组的消费情况，创建和删除 Topic，支持使用 KSQL 对 Kafka 消息做 Ad-hoc 查询，支持多种告警方式等等。
2.4.1下载并解压到指定路径：
EFAK
mkdir –p /opt/kafka-web
cd /opt/kafka-web
tar -zxvf kafka-eagle-bin-3.0.1.tar.gz
/etc/profile文件末尾添加如下环境变量，并执行source /etc/profile生效：
export KE_HOME=/opt/kafka-web/efak-web-3.0.1
export PATH=$PATH:$KE_HOME/bin
2.4.2修改配置文件server.properties：
mkdir -p /opt/kafka-web/db
[root@k1 config]# cd /opt/kafka-web
[root@k1 kafka-web]# ls
db efak-web-3.0.1
[root@k1 kafka-web]# cd efak-web-3.0.1/
[root@k1 efak-web-3.0.1]# ll
total 0
drwxr-xr-x. 2 root root 66 Sep 19 17:08 bin
drwxr-xr-x. 2 root root 112 Sep 20 09:25 conf
drwxr-xr-x. 2 root root 6 Sep 13 2021 db
drwxr-xr-x. 2 root root 23 Sep 19 16:46 font
drwxr-xr-x. 10 root root 103 Sep 19 17:33 kms
drwxr-xr-x. 2 root root 128 Sep 20 08:58 logs
[root@k1 efak-web-3.0.1]# cd conf
[root@k1 conf]# ll
total 16
-rw-r--r--. 1 root root 2217 Sep 13 2021 log4j.properties
-rw-r--r--. 1 root root 4295 Sep 19 17:04 system-config.properties
-rw-r--r--. 1 root root 9 Sep 13 2021 works
#修改红色的内容
[root@k1 conf]# more system-config.properties |grep -v "^$"|grep -v "#"
efak.zk.cluster.alias=cluster1
cluster1.zk.list=192.168.207.131:12181,192.168.207.132:12181,192.168.207.133:12181
cluster1.zk.acl.enable=false
cluster1.zk.acl.schema=digest
cluster1.zk.acl.username=test
cluster1.zk.acl.password=test123
cluster1.efak.broker.size=20
kafka.zk.limit.size=16
efak.webui.port=8048
efak.distributed.enable=false
efak.cluster.mode.status=master
efak.worknode.master.host=localhost
efak.worknode.port=8085
cluster1.efak.jmx.acl=false
cluster1.efak.jmx.user=keadmin
cluster1.efak.jmx.password=keadmin123
cluster1.efak.jmx.ssl=false
cluster1.efak.jmx.truststore.location=/data/ssl/certificates/kafka.truststore
cluster1.efak.jmx.truststore.password=ke123456
cluster1.efak.offset.storage=kafka
cluster2.efak.offset.storage=zk
cluster1.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=keadmin
cluster1.efak.sasl.enable=false
cluster1.efak.sasl.protocol=SASL_PLAINTEXT
cluster1.efak.sasl.mechanism=SCRAM-SHA-256
cluster1.efak.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="kafka" password="kafka-eagle";
cluster1.efak.sasl.client.id=
cluster1.efak.blacklist.topics=
cluster1.efak.sasl.cgroup.enable=false
cluster1.efak.sasl.cgroup.topics=
cluster2.efak.sasl.enable=false
cluster2.efak.sasl.protocol=SASL_PLAINTEXT
cluster2.efak.sasl.mechanism=PLAIN
cluster2.efak.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka" password="kafka-eagle";
cluster2.efak.sasl.client.id=
cluster2.efak.blacklist.topics=
cluster2.efak.sasl.cgroup.enable=false
cluster2.efak.sasl.cgroup.topics=
cluster3.efak.ssl.enable=false
cluster3.efak.ssl.protocol=SSL
cluster3.efak.ssl.truststore.location=
cluster3.efak.ssl.truststore.password=
cluster3.efak.ssl.keystore.location=
cluster3.efak.ssl.keystore.password=
cluster3.efak.ssl.key.password=
cluster3.efak.ssl.endpoint.identification.algorithm=https
cluster3.efak.blacklist.topics=
cluster3.efak.ssl.cgroup.enable=false
cluster3.efak.ssl.cgroup.topics=
efak.driver=org.sqlite.JDBC
efak.url=jdbc:sqlite:/opt/kafka-web/db/ke.db
efak.username=root
efak.password=www.kafka-eagle.org
2.4.3启动服务：
[root@k1 bin]# pwd
/opt/kafka-web/efak-web-3.0.1/bin
[root@k1 bin]# ll
total 28
-rw-r--r--. 1 root root 1848 Sep 13 2021 ke.bat
-rw-r--r--. 1 root root 6 Sep 19 17:33 ke.pid
-rwxr-xr-x. 1 root root 11289 Dec 12 2021 ke.sh
-rwxr-xr-x. 1 root root 4776 Dec 12 2021 worknode.sh
[root@k1 bin]# ./ke.sh start
输出日志最后会提示登录的地址和用户名密码。
浏览器登录管理控制台：
http://192.168.207.131:8048
admin/123456
3.日常启停顺序：
停止：
先停止kafka再停止zookeeper：
/opt/kafka/kafka_2.13-3.8.0/bin/kafka-server-stop.sh -daemon /opt/kafka/kafka_2.13-3.8.0/config/server.properties
/opt/zookeeper/zookeeper-3.9.2/bin/zkServer.sh stop
启动：
先启动zookeeper，再启动kafka：
/opt/zookeeper/zookeeper-3.9.2/bin/zkServer.sh start
/opt/kafka/kafka_2.13-3.8.0/bin/kafka-server-start.sh -daemon /opt/kafka/kafka_2.13-3.8.0/config/server.properties
查看状态：
[root@k3 conf]# /opt/zookeeper/zookeeper-3.9.2/bin/zkServer.sh status
/usr/bin/java
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper/zookeeper-3.9.2/bin/../conf/zoo.cfg
Client port found: 12181. Client address: localhost. Client SSL: false.
Mode: leader
检查：
jps
测试发布和消费：
发布：
/opt/kafka/kafka_2.13-3.8.0/bin/kafka-console-producer.sh --broker-list "192.168.207.131:9092,192.168.207.132:9092,192.168.207.133:9092" --topic topic_jyc01 --property parse.key=true
注意使用tab键分隔内容输入后回车
>jyc test
>a a
>b c
ctrl+c取消退出
消费：
/opt/kafka/kafka_2.13-3.8.0/bin/kafka-console-consumer.sh --bootstrap-server "192.168.207.131:9092,192.168.207.132:9092,192.168.207.133:9092" --topic topic_jyc01
4.参考文档：
Kafka 生产环境部署指南-阿里云开发者社区
kafka-eagle无法获取kafka cluster info-CSDN博客
Kafka【第一篇】Kafka集群搭建_kafka 集群-CSDN博客
![[Linux]Vi和Vim编辑器](https://img-blog.csdnimg.cn/img_convert/7e87d0f95f168ee07e3304fb6ec18407.png)
