Linux平台Kafka高可用集群部署全攻略
本文详细介绍了如何在CentOS 7环境下搭建高可用的Kafka集群。Kafka是一个分布式流处理平台，常用于构建实时数据流管道和流应用程序。为了确保Kafka的高可用性和容错性，集群搭建是必不可少的步骤。首先，介绍了环境准备，包括服务器准备（至少需要三台服务器或虚拟机）和版本信息（Kafka 2.13-3.x.x版本，JDK 1.8+）。在搭建Kafka集群之前，需要先安装和配置JDK和ZooK
🐇明明跟你说过：个人主页
🏅个人专栏：《大数据前沿：技术与应用并进》🏅
🔖行路有良友，便是天堂🔖
目录
一、引言
1、Kafka简介
Apache Kafka 是一个开源的分布式流处理平台，它最初由 LinkedIn 开发，后来成为 Apache 软件基金会的顶级项目。Kafka 设计用于处理实时数据流，提供了一种高效、可扩展、持久化的方式来进行数据发布和订阅。它通常被描述为一种分布式发布-订阅消息队列，但它实际上超越了传统消息队列的概念。
2、Kafka核心优势
1. 高吞吐量：
- Kafka 能够处理海量数据，支持每秒数十万条消息的读写操作，即使在大规模部署中也能保持高性能。
- 通过高效的文件系统设计和内存管理机制，Kafka 能够在处理大量数据的同时保持低延迟。
2. 持久性和可靠性：
- Kafka 将数据存储在磁盘上，并支持数据复制（replication），确保即使在节点故障的情况下也能保证数据的可靠性和持久性。
- 数据以追加的方式写入日志文件，减少了磁盘的随机写操作，提高了写入速度和数据完整性。
3. 可扩展性：
- Kafka 具有良好的水平扩展能力，可以通过增加更多的节点来提升系统的处理能力和存储容量。
- 分布式架构使得 Kafka 能够轻松地在多台服务器上部署，并且能够动态扩展和收缩集群大小。
4. 灵活的发布-订阅模型：
- Kafka 支持发布-订阅模式，允许多个消费者订阅同一个主题，并且消费者可以独立消费消息。
- 消费者可以控制自己的消费进度，不会影响其他消费者的状态，实现了消息消费的解耦。
二、环境准备
1、服务器
准备3台或者5台Linux服务器，用来组建高可用集群，这里使用3台Centos 7.9来进行搭建，大家也可以使用其他的Linux发行版本
配置如下：
2、服务器环境初始化
3台机器都要执行
关闭Selinux
vi /etc/selinux/config
#修改成如下
SELINUX=disabled之后重启服务器
reboot关闭并禁用防火墙
[root@kafka1 ~]# systemctl stop firewalld && systemctl disable firewalld修改 /etc/hosts
vi /etc/hosts
# 添加以下内容
192.168.40.100 kafka1
192.168.40.101 kafka2
192.168.40.102 kafka3修改镜像源
curl -o /etc/yum.repos.d/CentOS-Base.repo https://mirrors.aliyun.com/repo/Centos-7.repo三、安装zookeeper
为什么安装Kafka时，要先安装zookeeper：
ZooKeeper 是一个分布式的协调服务，它为分布式应用程序提供了一套完整的协调服务功能，包括命名服务、配置管理、集群管理和同步等。Kafka 利用 ZooKeeper 来管理其集群中的多个组件，确保系统的稳定性和一致性。
1、上传tar包
apache-zookeeper-3.8.0-bin.tar.gz
tar包可以去官网进行下载 Apache ZooKeeper
https://zookeeper.apache.org/releases.html#download
解压tar包至 /opt 下
tar zxvf apache-zookeeper-3.8.0-bin.tar.gz -C /opt2、编辑配置文件
vim /opt/apache-zookeeper-3.8.0-bin/conf/zoo.cfg
# 输入如下内容
tickTime=2000
initLimit=10
syncLimit=5
dataDir=/opt/zookeeper/zkData
dataLogDir=/opt/zookeeper/zkLog
clientPort=2181
server.1=kafka1:2188:3888
server.2=kafka2:2188:3888
server.3=kafka3:2188:3888
4lw.commands.whitelist=*
tickTime=2000：
- tickTime 定义了 ZooKeeper 服务器之间的心跳间隔时间（毫秒）。它是 ZooKeeper 中最基本的单位时间。默认值通常是 2000 毫秒（即 2 秒）。
initLimit=10：
- initLimit 定义了初始同步阶段的最大超时时间（心跳次数）。这意味着在初始同步阶段，跟随者（follower）必须在 initLimit * tickTime 毫秒内完成与领导者（leader）的同步。例如，这里设置为 20 秒（10 * 2000 毫秒）。
syncLimit=5：
- syncLimit 定义了在领导者和跟随者之间发送消息的最大超时时间（心跳次数）。这意味着在同步阶段，跟随者必须在 syncLimit * tickTime 毫秒内响应领导者的请求。例如，这里设置为 10 秒（5 * 2000 毫秒）。
dataDir=/opt/zookeeper/zkData：
- dataDir 指定 ZooKeeper 服务器用来存储快照（snapshot）的目录。
dataLogDir=/opt/zookeeper/zkLog：
- dataLogDir 指定 ZooKeeper 服务器用来存储事务日志（transaction logs）的目录。这是从 ZooKeeper 3.4.6 开始引入的一个配置项，使得日志和数据可以分开存储。
clientPort=2181：
- clientPort 指定客户端连接到 ZooKeeper 服务器的端口，默认为 2181。
server.1=ka1:2188:3888：
- server.N 表示第 N 台服务器的信息，格式为 hostname:peerPort:leaderPort。peerPort 是服务器之间通信的端口，leaderPort 是选举领导者时使用的端口。
4lw.commands.whitelist=*：
- 4lw.commands.whitelist 指定客户端可以执行的命令白名单。* 表示允许所有命令。
3、创建数据目录
mkdir -p /opt/zookeeper/zkData
mkdir -p /opt/zookeeper/zkLog创建集群ID文件
在3台机器上分别执行
[root@kafka1 bin]# echo 1 > /opt/zookeeper/zkData/myid
[root@kafka2 bin]# echo 2 > /opt/zookeeper/zkData/myid
[root@kafka3 bin]# echo 3 > /opt/zookeeper/zkData/myid4、安装JAVA
yum install -y java-1.8.0-openjdk-devel5、启动zookeeper
cd /opt/apache-zookeeper-3.8.0-bin/bin/
./zkServer.sh start ../conf/zoo.cfg &
查看状态
[root@kafka1 bin]# ps -aux | grep zook
四、Kafka集群搭建
1、上传tar包
资源包大家可以到官网下载
解压至指定目录
[root@kafka1 ~]# tar zxvf kafka_2.13-3.1.0.tgz -C /opt/
[root@kafka2 ~]# tar zxvf kafka_2.13-3.1.0.tgz -C /opt/
[root@kafka3 ~]# tar zxvf kafka_2.13-3.1.0.tgz -C /opt/2、编辑配置文件
在kafka1上执行
[root@kafka1 config]# vim /opt/kafka_2.13-3.1.0/config/server.properties
#输入如下内容
broker.id=0
listeners=PLAINTEXT://kafka1:9092
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-logs
num.partitions=5
default.replication.factor=2
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.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=kafka1:2181,kafka2:2181,kafka3:2181/kafka
zookeeper.connection.timeout.ms=18000
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0在kafka2上执行
[root@kafka2 config]# vim /opt/kafka_2.13-3.1.0/config/server.properties
#输入如下内容
broker.id=1
listeners=PLAINTEXT://kafka2:9092
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-logs
num.partitions=5
default.replication.factor=2
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.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=kafka1:2181,kafka2:2181,kafka3:2181/kafka
zookeeper.connection.timeout.ms=18000
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0在kafka3上执行
[root@kafka3 config]# vim /opt/kafka_2.13-3.1.0/config/server.properties
#输入如下内容
broker.id=2
listeners=PLAINTEXT://kafka3:9092
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-logs
num.partitions=5
default.replication.factor=2
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.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=kafka1:2181,kafka2:2181,kafka3:2181/kafka
zookeeper.connection.timeout.ms=18000
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0
broker.id=0
- 这是 Kafka broker 的唯一标识符。每个 broker 必须有唯一的 ID。这里的值为 0，意味着这是一个集群中的单个 broker 或者是第一个 broker。
listeners=PLAINTEXT://kafka1:9092
- 定义了 broker 监听的网络接口和端口。此处使用 PLAINTEXT 协议，意味着没有加密。kafka1:9092 表示监听名为 kafka1 的主机上的 9092 端口。
num.network.threads=3
- 指定了用于网络请求处理的线程数。网络请求包括接收来自生产者的消息、发送消息给消费者等操作。这里设置为 3 个线程。
num.io.threads=8
- 指定了用于处理 I/O 请求的线程数。I/O 请求包括磁盘上的读写操作。这里设置为 8 个线程。
socket.send.buffer.bytes=102400
- 设置了发送套接字的缓冲区大小（单位：字节）。此配置影响网络数据包的发送速度。此处设置为 102400 字节。
socket.receive.buffer.bytes=102400
- 设置了接收套接字的缓冲区大小（单位：字节）。此配置影响网络数据包的接收速度。此处设置为 102400 字节。
socket.request.max.bytes=104857600
- 定义了从客户端接收的最大请求大小（单位：字节）。这有助于防止因过大请求而导致的内存溢出。此处设置为 104857600 字节，即约 100MB。
log.dirs=/opt/kafka-logs
- 指定了日志文件存储的位置。日志文件包含了 Kafka topic 的数据。这里设置的日志目录为 /opt/kafka-logs。
num.partitions=5
- 指定了默认主题分区的数量。分区越多，通常意味着更高的并发度。这里设置的主题默认分区数为 5。
default.replication.factor=2
- 指定了创建新主题时的默认复制因子。复制因子决定了每个分区的副本数量。这里设置的复制因子为 2，意味着每个分区有 2 份副本。
num.recovery.threads.per.data.dir=1
- 指定了用于恢复日志段的线程数。每个数据目录可以有不同的线程数。这里设置为 1 个线程。
offsets.topic.replication.factor=1
- 指定了 _consumer_offsets 主题的复制因子。此主题用于存储消费者的偏移量信息。这里设置的复制因子为 1，意味着只有一个副本。
transaction.state.log.replication.factor=1
- 指定了 _transactions 主题的复制因子。此主题用于记录事务状态。这里设置的复制因子为 1，意味着只有一个副本。
transaction.state.log.min.isr=1
- 指定了 _transactions 主题的最小 ISR（In-Sync Replicas）数量。ISR 是与 leader 同步的副本集合。这里设置的最小 ISR 数量为 1。
log.retention.hours=168
- 指定了日志数据保留的时间长度（单位：小时）。这里设置的日志保留时间为 168 小时，即 7 天。
log.segment.bytes=1073741824
- 指定了日志段的最大大小（单位：字节）。一旦达到这个大小，Kafka 就会创建一个新的日志段。这里设置的日志段大小为 1073741824 字节，即 1GB。
log.retention.check.interval.ms=300000
- 指定了检查日志清理的间隔时间（单位：毫秒）。这里设置的检查间隔为 300000 毫秒，即 5 分钟。
zookeeper.connect=kafka1:2181,kafka2:2181,kafka3:2181/kafka
- 指定了 ZooKeeper 服务器列表。ZooKeeper 用于协调 Kafka 集群。这里设置的 ZooKeeper 服务器为 kafka1:2181, kafka2:2181, kafka3:2181，路径为 /kafka。
zookeeper.connection.timeout.ms=18000
- 指定了 Kafka 与 ZooKeeper 之间的连接超时时间（单位：毫秒）。这里设置的超时时间为 18000 毫秒，即 18 秒。请注意，zookeeper.connection.timeout.ms 在配置中出现了两次，应该是误写，只需要保留一次即可。
group.initial.rebalance.delay.ms=0
- 指定了消费者组初始重新平衡的延迟时间（单位：毫秒）。这里设置的延迟时间为 0，即立即开始重新平衡。
3、创建数据目录
在3台机器上分别执行
mkdir /opt/kafka-logs4、启动kafka
[root@kafka1 bin]# cd /opt/kafka_2.13-3.1.0/bin/
[root@kafka1 bin]# ./kafka-server-start.sh -daemon ../config/server.properties[root@kafka2 bin]# cd /opt/kafka_2.13-3.1.0/bin/
[root@kafka2 bin]# ./kafka-server-start.sh -daemon ../config/server.properties[root@kafka3 bin]# cd /opt/kafka_2.13-3.1.0/bin/
[root@kafka3 bin]# ./kafka-server-start.sh -daemon ../config/server.properties5、查看端口状态
[root@kafka1 bin]# netstat -antupl
五、测试
1、创建Topic
前面我们已经将kafka集群搭建起来了，接下来创建一个Topic进行写入测试，如果不清楚Topic是什么，可以翻看作者之前的文章。
在kafka1上执行
[root@kafka1 bin]# cd /opt/kafka_2.13-3.1.0/bin
[root@kafka1 bin]# ./kafka-topics.sh --bootstrap-server=192.168.40.100:9092 --topic test --create --partitions=3 --replication-factor=2- --bootstrap-server：指定 Kafka broker 的地址和端口号。这里的 192.168.40.100:9092 指定了 broker 的 IP 地址为 192.168.40.100，端口号为 9092。
- --topic：指定要操作的主题名称。在这个例子中，主题名为 test。
- --create：告诉 Kafka 创建一个新主题。如果主题已经存在，这条命令将会失败，除非你配置了允许创建已存在的主题。
- --partitions：指定主题的分区数。分区数决定了主题能够并行处理消息的能力。在这个例子中，主题 test 将会有 3 个分区。
- --replication-factor：指定主题的复制因子。复制因子决定了每个分区的副本数量，这对于数据的冗余和可靠性非常重要。在这个例子中，主题 test 的每个分区将会有 2 个副本。
查看Topic
[root@kafka1 bin]# ./kafka-topics.sh --bootstrap-server=192.168.40.100:9092 --list
2、生产消息
[root@kafka1 bin]# ./kafka-console-producer.sh --bootstrap-server 192.168.40.100:9092 --topic test向我们刚刚创建的test Topic写入几条消息
3、消费消息
[root@kafka1 bin]# ./kafka-console-consumer.sh --bootstrap-server=192.168.40.100:9092 --topic test --from-beginning
如果能看到之前生产的消息，则证明集群搭建成功
💕💕💕每一次的分享都是一次成长的旅程，感谢您的陪伴和关注。希望这些关于大数据的文章能陪伴您走过技术的一段旅程，共同见证成长和进步！😺😺😺
🧨🧨🧨让我们一起在技术的海洋中探索前行，共同书写美好的未来！！！
华为开发者空间，是为全球开发者打造的专属开发空间，汇聚了华为优质开发资源及工具，致力于让每一位开发者拥有一台云主机，基于华为根生态开发、创新。
更多推荐
70
0- 0
已为社区贡献17条内容
所有评论(0)