中间件Kafka 高可用企业级集群交付文档

Kafka集群配置说明

Kafka集群开启JMX端口后，在Broker执行kafka-topic命令会存在JMX端口占用情况，所有命令需要到Kafka Client节点执行

配置文件介绍

以下配置文件均在/opt/app/conf/kafka/server.properties中填写

一、Kafka服务配置

二、高可用配置

三、Kafka Log相关配置

四、数据删除策略

五、性能调优

六、Kafka-Zookeeper配置

七、SASL安全配置

SASL详细配置参数可以参考下面文档

Kafka 安全认证SASL

新闻联播老司机

• 23年9月4日
• 喜欢：0
• 浏览：359

Zookeeper集群部署

配置文件介绍

Zookeeper配置文件默认路径为/opt/app/conf/zookeeeper/zoo.cfg

部署交付

修改主机名和host

# vim /etc/hosts
192.168.31.70  zook01
192.168.31.71  zook02
192.168.31.72  zook03

数据存储在/data/zookeeper

mkdir -p /data/zookeeper

zookeeper为3台节点，需在每台节点上执行相同步骤

wget -P /opt/  https://archive.apache.org/dist/zookeeper/zookeeper-3.6.3/apache-zookeeper-3.6.3-bin.tar.gz 
cd /opt/ && tar xf apache-zookeeper-3.6.3-bin.tar.gz
ln -s /opt/apache-zookeeper-3.6.3-bin /opt/zookeeper

编辑配置文件 /opt/app/conf/zookeeper/zoo.cfg

mkdir -p /opt/app/conf/zookeeper

cat >/opt/app/conf/zookeeper/zoo.cfg<<EOF
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/data/zookeeper
clientPort=2181
maxClientCnxns=1000
autopurge.snapRetainCount=3
autopurge.purgeInterval=1
4lw.commands.whitelist=mntr,conf,cons,crst,dump,envi,gtmk,isro,ruok,srst,srvr,stat,stmk,wchc,wchp,wchs
tcpKeepAlive=false

server.1=zook01:2888:3888
server.2=zook02:2888:3888
server.3=zook03:2888:3888
EOF

还需要额外在/data/zookeeper/下创建myid文件,这里配置每台节点不一样，根据实际情况修改

echo "1" >>/data/zookeeper/myid

#Zookeeper 01节点ID设置为1
#Zookeeper 02节点ID设置为2
#Zookeeper 03节点ID设置为3

启动Zookeeper

[root@zook01 bin]# cd /opt/zookeeper/bin/ && ./zkServer.sh --config /opt/app/conf/zookeeper start

启动文件目录/opt/app/conf/zookeeper

查看每台节点集群状态，此时集群已经部署完毕

root@zook01:/opt/zookeeper/bin# ./zkServer.sh --config /opt/app/conf/zookeeper status
ZooKeeper JMX enabled by default
Using config: /opt/app/conf/zookeeper/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: follower      --->从节点

root@zook02:/opt/zookeeper/bin# ./zkServer.sh --config /opt/app/conf/zookeeper status
ZooKeeper JMX enabled by default
Using config: /opt/app/conf/zookeeper/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: leader        --->主节点

root@zook03:/opt/zookeeper/bin#  ./zkServer.sh --config /opt/app/conf/zookeeper status
ZooKeeper JMX enabled by default
Using config: /opt/app/conf/zookeeper/zoo.cfg
Client port found: 2181. Client address: localhost. Client SSL: false.
Mode: follower      --->从节点

Kafka部署集群

• Kafka为3节点，Zookeeper同为3节点

所有节点都需要安装JDK环境 （kafka-manager节点需要JDK11版本）

wget https://d.frps.cn/file/tools/jdk/jdk-8u231-linux-x64.tar.gz  --no-check-certificate
mkdir /usr/java
tar zxvf jdk-8u231-linux-x64.tar.gz -C /usr/java
ln -s /usr/java/jdk1.8.0_231/ /usr/java/jdk
cat > /etc/profile.d/java_home.sh << EOF
# set java environment  
export JAVA_HOME=/usr/java/jdk
export CLASSPATH=.:\$JAVA_HOME/lib/dt.jar:\$JAVA_HOME/lib/tools.jar:\$JAVA_HOME/jre/lib/rt.jar
export PATH=\$PATH:\$JAVA_HOME/bin
EOF
source /etc/profile

Kafka集群部署

wget -P /opt https://downloads.apache.org/kafka/3.9.0/kafka_2.13-3.9.0.tgz
cd  /opt/ && tar xf kafka_2.13-3.9.0.tgz 
ln -s /opt/kafka_2.13-3.9.0 /opt/kafka

所有节点创建kafka数据存储目录以及配置文件目录

mkdir -p /data/kafka/kafka-logs
mkdir -p /opt/app/conf/kafka

kafka01节点主配置文件

cat >/opt/app/conf/kafka/server.properties<<EOF

broker.id=1
broker.rack=abcdocker
host.name=zook01

#log
log.dirs=/data/kafka/kafka-logs
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.check.interval.ms=300000

port=9092
replica.fetch.max.bytes=1000000

#zk
zookeeper.connect=zook01:2181,zook02:2181,zook03:2181
zookeeper.connection.timeout.ms=6000
zookeeper.session.timeout=6000

#sasl
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
listeners=SASL_PLAINTEXT://192.168.31.70:9092

#kafka
auto.create.topics.enablee=true
compression.type=producer
default.replication.factor=3
delete.topic.enable=true
group.max.session.timeout.ms=1800000
group.min.session.timeout.ms=6000
kafka.scala.version=2.13

log.cleaner.enablee=true
log.cleanup.policy=delete
log.retention.bytes=-1
log.retention.hours=168
log.segment.bytes=1073741824
log.segment.delete.delay.ms=60000
message.max.bytes=1000000
num.io.threads=8
num.network.threads=3
num.partitions=3
num.recovery.threads.per.data.dir=1
num.replica.fetchers=2
offsets.topic.replication.factor=3
queued.max.requests=500
replica.log.time.max.ms=10000
socket.receive.buffer.bytes=102400
socket.send.buffer.bytes=102400
unclean.leader.election.enable=true
EOF

kafka02节点主配置文件

cat >/opt/app/conf/kafka/server.properties<<EOF

broker.id=2
broker.rack=abcdocker
host.name=zook02

#log
log.dirs=/data/kafka/kafka-logs
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.check.interval.ms=300000

port=9092
replica.fetch.max.bytes=1000000

#zk
zookeeper.connect=zook01:2181,zook02:2181,zook03:2181
zookeeper.connection.timeout.ms=6000
zookeeper.session.timeout=6000

#sasl
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
listeners=SASL_PLAINTEXT://192.168.31.71:9092

#kafka
auto.create.topics.enablee=true
compression.type=producer
default.replication.factor=3
delete.topic.enable=true
group.max.session.timeout.ms=1800000
group.min.session.timeout.ms=6000
kafka.scala.version=2.13

log.cleaner.enablee=true
log.cleanup.policy=delete
log.retention.bytes=-1
log.retention.hours=168
log.segment.bytes=1073741824
log.segment.delete.delay.ms=60000
message.max.bytes=1000000
num.io.threads=8
num.network.threads=3
num.partitions=3
num.recovery.threads.per.data.dir=1
num.replica.fetchers=2
offsets.topic.replication.factor=3
queued.max.requests=500
replica.log.time.max.ms=10000
socket.receive.buffer.bytes=102400
socket.send.buffer.bytes=102400
unclean.leader.election.enable=true
EOF

kafka03节点主配置文件

cat >/opt/app/conf/kafka/server.properties<<EOF

broker.id=3
broker.rack=abcdocker
host.name=zook03

#log
log.dirs=/data/kafka/kafka-logs
log.flush.interval.messages=10000
log.flush.interval.ms=1000
log.retention.check.interval.ms=300000

port=9092
replica.fetch.max.bytes=1000000

#zk
zookeeper.connect=zook01:2181,zook02:2181,zook03:2181
zookeeper.connection.timeout.ms=6000
zookeeper.session.timeout=6000

#sasl
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
listeners=SASL_PLAINTEXT://192.168.31.72:9092

#kafka
auto.create.topics.enablee=true
compression.type=producer
default.replication.factor=3
delete.topic.enable=true
group.max.session.timeout.ms=1800000
group.min.session.timeout.ms=6000
kafka.scala.version=2.13

log.cleaner.enablee=true
log.cleanup.policy=delete
log.retention.bytes=-1
log.retention.hours=168
log.segment.bytes=1073741824
log.segment.delete.delay.ms=60000
message.max.bytes=1000000
num.io.threads=8
num.network.threads=3
num.partitions=3
num.recovery.threads.per.data.dir=1
num.replica.fetchers=2
offsets.topic.replication.factor=3
queued.max.requests=500
replica.log.time.max.ms=10000
socket.receive.buffer.bytes=102400
socket.send.buffer.bytes=102400
unclean.leader.election.enable=true
EOF

我们还需要在本地添加client加密配置文件，因为配置了密码，本地kafka-topic.sh执行时也需要指定密码配置文件

这个文件是配置Broker服务端的JASS，在Kafka程序目录的config目录中创建kafka_servre_jass.conf

需要在所有机器上同步执行

cat >/opt/app/conf/kafka/kafka_server_jaas.conf<<EOF
KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="kafka"
    password="kafka-secret"
    user_kafka="kafka-secret";
};
EOF

• Username 用户名
• Password 密码
• user_admin 密码

开启jmx端口号

二进制Kafka集群接入JMX_Exporter以及Kafka_exporter监控

新闻联播老司机

• 23年12月21日
• 喜欢：0
• 浏览：168

Kafka开通JMX端口配置

新闻联播老司机

• 23年6月15日
• 喜欢：0
• 浏览：1.2k

#添加jmx端口号，默认9988
vim /opt/kafka/bin/kafka-server-start.sh

#在JMX参数上面添加9999端口号
 export JMX_PORT=9999
#jmx端口不可以设置9999，会影响执行topic脚本

修改Kafka启动脚本，引用kafka_servre_jass.conf配置

vim /opt/kafka/bin/kafka-run-class.sh

#在JMX参数旁边添加引用kafka_servre_jass.conf配置
KAFKA_JMX_OPTS="$KAFKA_JMX_OPTS  -Djava.security.auth.login.config=/opt/app/conf/kafka/kafka_server_jaas.conf"

现在已经开启了客户端-->broker的SASL加密，我们需要进行验证

验证同样需要配置密码配置文件

mkdir /ssl

cat >>/ssl/kafka.config<<EOF
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="kafka" password="kafka-secret";
EOF

#username=SASL认证用户名
#paasword=SASL认证密码
#所有节点创建

启动Kafka (所有节点同步创建即可)

[root@zook01 ~]# cd /opt/kafka
[root@zook01 kafka]# ./bin/kafka-server-start.sh -daemon /opt/app/conf/kafka/server.properties

#指定配置文件路径/opt/app/conf/kafka/server.properties

执行以下命令，进入命令行工具所在目录。

cd /opt/kafka/bin

#Kafka生产者，消费者，topic命令都需要进入到/opt/kafka/bin中执行

执行以下命令，向 Topic 发送消息。

./kafka-console-producer.sh --bootstrap-server <Kafka 连接地址> --topic <topic_name> --producer.config /ssl/kafka.config
./kafka-console-producer.sh --bootstrap-server 192.168.31.70:9092 --topic kafka_test --producer.config /ssl/kafka.config

手动创建 kafka_test topic

root@zook03:/opt/kafka/bin# ./kafka-topics.sh --create --topic kafka_test_01 --bootstrap-server zook01:9092,zook02:9092,zook03:9092 --partitions 3 --replication-factor 1 --command-config /ssl/kafka.config
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 kafka_test_01.

#--topc kafka_test_01为Kafka topic名称
#zook01:9092,zook02:9092,zook03:9092  Kafka Broker节点(可以写ip，或者host解析)

生产数据

./kafka-console-producer.sh --bootstrap-server <Kafka 连接地址> --topic <topic_name> --producer.config /ssl/kafka.config
root@zook03:/opt/kafka/bin# ./kafka-console-producer.sh --bootstrap-server zook01:9092,zook02:9092,zook03:9092 --topic kafka_test_01 --producer.config /ssl/kafka.config
>ddddd
>cccc
>vvvv
>ffff
>

消费数据

./kafka-console-consumer.sh --bootstrap-server <Kafka 连接地址> --topic <topic_name> --group <group_name> --from-beginning --consumer.config /ssl/kafka.config
./kafka-console-consumer.sh --bootstrap-server  zook01:9092,zook02:9092,zook03:9092 --topic kafka_test --from-beginning --consumer.config /ssl/kafka.config

查看所有topic

[root@kafka-manager bin]# ./kafka-topics.sh --bootstrap-server zook01:9092,zook02:9092,zook03:9092 --command-config /ssl/kafka.config --list
__consumer_offsets
kafka_test
kafka_test_01

Kafka Manager部署

github下载kafka-manager包

wget https://github.com/yahoo/CMAK/releases/download/3.0.0.6/cmak-3.0.0.6.zip
unzip cmak-3.0.0.6.zip
mv cmak-3.0.0.6 /opt/
ln -s /opt/cmak-3.0.0.6 /opt/cmak

Kafka-manager需要JDK 11.0版本

wget wget https://repo.huaweicloud.com/openjdk/11.0.1/openjdk-11.0.1_linux-x64_bin.tar.gz  --no-check-certificate
mkdir /usr/java
tar xf openjdk-11.0.1_linux-x64_bin.tar.gz -C /usr/java
ln -s /usr/java/jdk-11.0.1 /usr/java/jdk
cat > /etc/profile.d/java_home.sh << EOF
# set java environment  
export JAVA_HOME=/usr/java/jdk
export CLASSPATH=.:\$JAVA_HOME/lib/dt.jar:\$JAVA_HOME/lib/tools.jar:\$JAVA_HOME/jre/lib/rt.jar
export PATH=\$PATH:\$JAVA_HOME/bin
EOF
source /etc/profile

#检查版本
[root@kafka-manager java]# java -version
openjdk version "11.0.1" 2018-10-16
OpenJDK Runtime Environment 18.9 (build 11.0.1+13)
OpenJDK 64-Bit Server VM 18.9 (build 11.0.1+13, mixed mode)

修改CMAK中Zookeeper连接信息

[root@kafka-manager ~]# vim /opt/cmak/conf/application.conf

# Settings prefixed with 'kafka-manager.' will be deprecated, use 'cmak.' instead.
# https://github.com/yahoo/CMAK/issues/713
#kafka-manager.zkhosts="zook01:2181,zook02:2181,zook03:2181"
kafka-manager.zkhosts="zook01:2181,zook02:2181,zook03:2181"
#kafka-manager.zkhosts=${?ZK_HOSTS}
cmak.zkhosts="zook01:2181,zook02:2181,zook03:2181"
#cmak.zkhosts=${?ZK_HOSTS}

#注释多余配置文件，保留kafka-manager.zkhosts和cmak.zkhosts
#填写ZK连接信息

配置用户名密码
cmak需要配置用户访问密码，同样在application.conf中调整

[root@kafka-manager ~]# vim /opt/cmak/conf/application.conf

#开启密码访问，默认为flase
basicAuthentication.enabled=true

#管理员用户
basicAuthentication.username="admin"
#管理员密码
basicAuthentication.password="password"

启动cmak

/opt/cmak/bin
nohup ./cmak  -Dconfig.file=/opt/cmak/conf/application.conf -Dhttp.port=9000 &

检查9000端口是否正常

[root@kafka-manager ~]# lsof -i:9000
COMMAND   PID USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
java    39864 root  140u  IPv6  97965      0t0  TCP *:cslistener (LISTEN)
[root@kafka-manager ~]#
[root@kafka-manager ~]# netstat -lntup|grep 9000
tcp6       0      0 :::9000                 :::*                    LISTEN      39864/java
[root@kafka-manager ~]#
[root@kafka-manager ~]#
[root@kafka-manager ~]# curl localhost:9000 -I
HTTP/1.1 401 Unauthorized
WWW-Authenticate: Basic realm="Kafka-Manager"
Date: Fri, 29 Nov 2024 03:09:56 GMT
Content-Length: 0
访问kafka-manager (CMAK)地址，默认端口号9000
http://192.168.31.73:9000

#有认证默认用户admin
#默认密码password

验证集群