Kafka3.8.0+Centos7.9的安装参考

环境准备

操作系统版本：centos7.9

用户/密码：root/1qazXSW@

主机名	IP地址	安装软件	下载地址
k1	192.168.207.131	jdk1.8 zookeeper3.9.2 kafka_2.13-3.8.0 efak-web-3.0.1	1） Java Downloads \| Oracle 2） Apache ZooKeeper 3） Apache Kafka 4） EFAK
k2	192.168.207.132	jdk1.8 zookeeper3.9.2 kafka_2.13-3.8.0
k3	192.168.207.133	jdk1.8 zookeeper3.9.2 kafka_2.13-3.8.0

以下三台机器都操作：

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"

#从后台启动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