Kafka集群安装部署教程2(不依赖ZooKeeper)
作者:hangge | 2024-05-15 08:33
Kafka 是一个分布式消息系统,我在前文介绍如何结合 ZooKeeper 进行 Kafka 集群的部署(点击查看)。本文接着演示如何在无需依赖 ZooKeeper 的情况下进行 Kafka 集群的部署。
1,基本介绍
(1)我们知道在 Kafka 2.X 版本需要依赖 ZooKeeper,具体来说 ZooKeeper 提供如下作用:
- 协调与管理: ZooKeeper 主要用于协调和管理分布式系统中的各个节点。在旧版 Kafka 中,ZooKeeper 负责管理 Kafka 集群的元数据,包括 Kafka Broker 的信息、Topic 的配置以及分区的分配等。
- 领导者选举: ZooKeeper 在 Kafka 中被用于实现领导者选举。每个分区都有一个领导者(leader),负责处理消息的读写。ZooKeeper 确保当领导者失效时,能够选举新的领导者,保证系统的可用性。
- 分布式锁: Kafka 使用 ZooKeeper 来实现分布式锁,以确保在集群中只有一个 Broker 可以充当 Topic 的分区的领导者。这有助于避免数据一致性问题。
(2)而 Kafka 3.X 版本引入了 KRaft(Kafka Raft)协议,以取代依赖 ZooKeeper 进行元数据和领导者选举的旧版 Kafka 集群,从而使得新版 Kafka 集群不再需要 ZooKeeper,因此可以独立运行。具体来熟,KRaft 起到了如下作用:
- 元数据存储: KRaft 将 Kafka 的元数据(例如 Topic 和分区的配置信息)存储在 KRaft 集群中,不再依赖于 ZooKeeper。这简化了 Kafka 集群的架构,使得 Kafka 可以在不需要额外的 ZooKeeper 的情况下运行。
- 领导者选举: KRaft 提供了一种新的领导者选举机制,替代了 ZooKeeper 的角色。KRaft 集群中的每个分区都有多个副本,其中一个是领导者,其余是追随者。KRaft 使用分布式 Raft 协议确保领导者的选举和复制日志的一致性。
- 强一致性: KRaft 在设计上追求强一致性,确保在各种故障情况下仍然能够保持数据的一致性。这使得 Kafka 更适合一些对一致性要求较高的应用场景。
2,准备工作
(1)首先我们准备三台 linux 服务器进行集群搭建,具体信息如下:
主机名 | 内网IP | 外网IP |
node1 | 172.31.4.99 | 13.211.44.248 |
node2 | 172.31.3.72 | 3.106.170.197 |
node3 | 172.31.6.103 | 3.25.88.196 |
(2)为方便后续 Kafka 集群的配置,我们在三台服务上均执行如下命令编辑的 host 文件,使得可以通过主机名相互访问:
echo ' 172.31.4.99 node1 172.31.3.72 node2 172.31.6.103 node3' >> /etc/hosts
(3)由于 Kafka3.0 不再支持 JDK8,建议安装 JDK11 或 JDK17,具体安装方法可以参考我之前写的文章:
3,下载 Kafka 安装包
(1)三台服务器均执行如下命令下载 Kafka 安装包,这里我选择是 3.3.1 版本:wget https://archive.apache.org/dist/kafka/3.3.1/kafka_2.12-3.3.1.tgz
tar -zxvf kafka_2.12-3.3.1.tgz
(3)最后将解压出来的文件夹移动到合适的位置,这个可以根据个人习惯修改:
mv kafka_2.12-3.3.1 /usr/local/kafka
4,修改 kafka 配置
(1)首先进入 kafka 目录:
cd /usr/local/kafka
(2)编辑 Kraft 协议配置文件:
vi config/kraft/server.properties
(3)node1 服务器的配置文件修改如下:
注意:advertised.listeners 指定 kafka 通过代理暴漏的地址,如果都是局域网使用,就配置 PLAINTEXT://:9092 即可。如果需要暴露给外部使用,则需要填写外部代理地址。比如我使用的是云服务器搭建 Kafka 集群,node1 对应的 172.31.4.99 位内部局域网地址,而其外网地址是 13.211.44.248。如果需要从外网访问 Kafka,那么 advertised.listeners 就要配置成 PLAINTEXT://13.211.44.248:9092。其他节点配置同理。
# The node id associated with this instance's roles node.id=1 # The connect string for the controller quorum controller.quorum.voters=1@node1:9093,2@node2:9093,3@node3:9093 listeners=PLAINTEXT://node1:9092,CONTROLLER://node1:9093 advertised.listeners=PLAINTEXT://:9092
(4)node2 服务器的配置文件修改如下:
# The node id associated with this instance's roles node.id=2 # The connect string for the controller quorum controller.quorum.voters=1@node1:9093,2@node2:9093,3@node3:9093 listeners=PLAINTEXT://node2:9092,CONTROLLER://node2:9093 advertised.listeners=PLAINTEXT://:9092
(5)node3 服务器的配置文件修改如下:
# The node id associated with this instance's roles node.id=3 # The connect string for the controller quorum controller.quorum.voters=1@node1:9093,2@node2:9093,3@node3:9093 listeners=PLAINTEXT://node3:9092,CONTROLLER://node3:9093 advertised.listeners=PLAINTEXT://:9092
5,格式化存储目录
(1)我们在任意一台 Kafka 服务器上执行一次如下命令即可,用于生成一个唯一的集群 ID
(2)比如我这里生成的是 I9keF5LZTTOZyMOaD5vh6w![](http://hangge.com/blog_uploads/202312/2023122718214721502.png)
(4)格式化操作完成之后,你会发现在我们定义的 log.dirs 目录下多出一个 meta.properties 文件:
![](http://hangge.com/blog_uploads/202312/2023122718283494743.png)
![](http://hangge.com/blog_uploads/202312/2023122718294077885.png)
![](http://hangge.com/blog_uploads/202312/2023122718300566141.png)
(2)启动后可以执行 jps 命令查看是否启动:
提示:这一个步骤是在安装 kafka 2.0 版本的时候不存在的。
/usr/local/kafka/bin/kafka-storage.sh random-uuid
(2)比如我这里生成的是 I9keF5LZTTOZyMOaD5vh6w
![](http://hangge.com/blog_uploads/202312/2023122718214721502.png)
(3)接着三台服务器都要执行如下命令,注意其中的 ID 替换成我们前面生成的:
./bin/kafka-storage.sh format \ -t I9keF5LZTTOZyMOaD5vh6w \ -c ./config/kraft/server.properties
![](http://hangge.com/blog_uploads/202312/2023122718282879275.png)
(5)meta.properties 文件中存储了当前的 kafka 节点的 id(node.id),当前节点属于哪个集群:
![](http://hangge.com/blog_uploads/202312/2023122718283494743.png)
(6)另外两个节点的 meta.properties 文件内容如下:
![](http://hangge.com/blog_uploads/202312/2023122718294077885.png)
![](http://hangge.com/blog_uploads/202312/2023122718300566141.png)
6,启动集群
(1)三台服务器均执行如下命令启动 Kafka 集群:
cd /usr/local/kafka ./bin/kafka-server-start.sh -daemon ./config/kraft/server.properties
(2)启动后可以执行 jps 命令查看是否启动:
![](http://hangge.com/blog_uploads/202312/2023122719202018583.png)
(3)如果没有启动成功可以查看日志文件查找原因。比如:Kafka 程序默认 jvm 内存为 1G,如果服务器内存太小,就是无法正常启动。我们可以编辑启动脚本:
vi ./bin/kafka-server-start.sh
(4)将 JVM 参数修改为最大 256M 最小 128M,然后保存退出。
export KAFKA_HEAP_OPTS="-Xmx256M -Xms128M"
![](http://hangge.com/blog_uploads/202312/2023122718465032046.png)
./bin/kafka-server-start.sh -daemon ./config/kraft/server.properties
7,测试 Kafka 集群
(1)比如我们在任意一个节点上创建一个生产者连接 node1 节点的 hangge-test 这个 topic:./bin/kafka-console-producer.sh --broker-list node1:9092 --topic hangge-test
(2)接着再换一个节点创建一个消费者监听 node2 节点的 hangge-test 这个 topic:
./bin/kafka-console-consumer.sh --bootstrap-server node2:9092 --topic hangge-test
(3)生产者这边输入一些数据:
![](http://hangge.com/blog_uploads/202312/2023122719310045639.png)
![](http://hangge.com/blog_uploads/202312/2023122719310871410.png)
全部评论(0)