当前位置： 首页 > news >正文

Rocky Linux 10 部署 Kafka 集群

news 2025/8/13 6:41:45

1.概述

Apache Kafka 是一个开源的消息系统，它提供了一个分布式、分区的、可复制的提交日志服务。

2.特点

  • 消息（Message）：Kafka 中最基本的数据单元是消息。每条消息都是一个键值对，键和值可以是任意的字节数组。
  • 主题（Topic）：消息的类别被称为 Topic。生产者将消息发布到特定的 Topic，而消费者订阅一个或多个 Topic 来接收消息。
  • 分区（Partition）：每个 Topic 可以分为多个 Partition。Partition 不仅允许消息通过多个消费者并发处理，还提供了 Kafka 的扩展性和顺序保证。每个 Partition 都是一个有序的、不可变的消息序列。
  • 偏移量（Offset）：每条消息在它所属的 Partition 中都有一个唯一的序列号称为 Offset。消费者通过 Offset 来跟踪它们已经消费了哪些消息。
  • 生产者（Producer）：负责将消息发布到 Kafka 的 Topic 中。生产者可以选择将消息发送到特定的 Partition。
  • 消费者（Consumer）：订阅一个或多个 Topic 并处理发布的消息。消费者追踪它们所消费的消息 Offset，并定期提交这些信息到 Kafka 或外部存储中。
  • 经纪人（Broker）：Kafka 集群由一个或多个服务器组成，这些服务器被称为 Brokers。每个 Broker 可以容纳多个 Topic 和 Partition。
  • 日志（Log）：从实现的角度来看，Partition 实际上就是一个分布式提交日志。新的消息不断被追加到 Partition 的末尾。
  • 复制（Replication）：为了容错，Kafka 允许为每个 Partition 设置一个或多个副本。这样即使某些 Broker 出现故障，系统也能继续运行。
  • 领导者（Leader）与追随者（Follower）：对于每个 Partition，Kafka 选择一个副本作为 Leader，其余副本则作为 Followers。Leader 负责处理读写请求，Follower 则从 Leader 同步数据。

3.系统环境准备

3.1集群信息
节点IP地址Kafka版本
node1192.168.100.52.13-4.0.0
node2192.168.100.62.13-4.0.0
node3192.168.100.72.13-4.0.0
3.2IP配置

配置node1节点IP地址

cat > /etc/NetworkManager/system-connections/ens160.nmconnection <<EOF
[connection]
id=ens160
type=ethernet
autoconnect-priority=-999
interface-name=ens160[ethernet][ipv4]
address1=192.168.100.5/24
dns=114.114.114.114;8.8.8.8;
gateway=192.168.100.254
may-fail=false
method=manual[ipv6]
addr-gen-mode=eui64
method=ignore[proxy]
EOF

配置node2节点IP地址

cat > /etc/NetworkManager/system-connections/ens160.nmconnection <<EOF
[connection]
id=ens160
type=ethernet
autoconnect-priority=-999
interface-name=ens160[ethernet][ipv4]
address1=192.168.100.6/24
dns=114.114.114.114;8.8.8.8;
gateway=192.168.100.254
may-fail=false
method=manual[ipv6]
addr-gen-mode=eui64
method=ignore[proxy]
EOF

配置node3节点IP地址

cat > /etc/NetworkManager/system-connections/ens160.nmconnection <<EOF
[connection]
id=ens160
type=ethernet
autoconnect-priority=-999
interface-name=ens160[ethernet][ipv4]
address1=192.168.100.7/24
dns=114.114.114.114;8.8.8.8;
gateway=192.168.100.254
may-fail=false
method=manual[ipv6]
addr-gen-mode=eui64
method=ignore[proxy]
EOF
3.3修改主机名

node1上执行

hostnamectl set-hostname node1

node2上执行

hostnamectl set-hostname node2

node3上执行

hostnamectl set-hostname node3
3.4关闭防火墙

3个节点均需要执行

systemctl stop firewalld
systemctl disable  firewalld
systemctl status  firewalld
3.5关闭SELINUX

3个节点均需要执行

setenforce 0
sed -i "s/^SELINUX=enforcing/SELINUX=disabled/g" /etc/selinux/config
3.6DNF源配置

3个节点均需要执行

sed -e 's|^mirrorlist=|#mirrorlist=|g'     -e 's|^#baseurl=http://dl.rockylinux.org/$contentdir|baseurl=https://mirrors.aliyun.com/rockylinux|g'     -i.bak     /etc/yum.repos.d/rocky*.repo
3.7时间同步配置

3个节点均需要执行

dnf install -y chrony
sed -i 's|pool[[:space:]]*2\.rocky\.pool\.ntp\.org[[:space:]]*iburst|server ntp.aliyun.com iburst|' /etc/chrony.conf
systemctl restart chronyd
chronyc sources
3.8主机映射配置

3个节点均需要执行

tee >> /etc/hosts <<EOF
192.168.100.5   node1
192.168.100.6   node2
192.168.100.7   node3
EOF

4.软件安装

4.1安装 Kafka 集群
4.1.1下载Kafka

3个节点均需要执行

wget https://dlcdn.apache.org/kafka/4.0.0/kafka_2.13-4.0.0.tgz
4.1.2解压Kafka

3个节点均需要执行

tar zxvf kafka_2.13-4.0.0.tgz -C /usr/local
ln -s kafka_2.13-4.0.0 kafka
4.1.3metadata.log.dir 目录创建及授权
mkdir -p /tmp/kafka-logs
chmod 755 /tmp/kafka-logs
4.1.4修改配置文件

node1上执行

cat > /usr/local/kafka/config/server.properties <<EOF
node.id=1# 节点角色：controller 和 broker
process.roles=controller,broker# 元数据日志存储目录（需提前创建并赋权）
metadata.log.dir=/tmp/kafka-logs# Controller 之间的投票节点列表（格式：id@host:port）
controller.quorum.voters=1@node1:9093,2@node2:9093,3@node3:9093# 监听器定义：名称、协议、host:port
# 注意：CONTROLLER 是用于 controller 间通信的监听器
#       PLAINTEXT 是用于 broker 对外服务的监听器
listeners=PLAINTEXT://:9092,CONTROLLER://:9093# 指定用于 controller 通信的监听器名称（必须是 listeners 中的一个）
controller.listener.names=CONTROLLER# 监听器安全协议映射
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT# Broker 间通信使用的监听器名称
inter.broker.listener.name=PLAINTEXT# 客户端连接地址（外部可访问的 host:port）
advertised.listeners=PLAINTEXT://node1:9092
EOF

node2上执行

cat > /usr/local/kafka/config/server.properties <<EOF
node.id=2# 节点角色：controller 和 broker
process.roles=controller,broker# 元数据日志存储目录（需提前创建并赋权）
metadata.log.dir=/tmp/kafka-logs# Controller 之间的投票节点列表（格式：id@host:port）
controller.quorum.voters=1@node1:9093,2@node2:9093,3@node3:9093# 监听器定义：名称、协议、host:port
# 注意：CONTROLLER 是用于 controller 间通信的监听器
#       PLAINTEXT 是用于 broker 对外服务的监听器
listeners=PLAINTEXT://:9092,CONTROLLER://:9093# 指定用于 controller 通信的监听器名称（必须是 listeners 中的一个）
controller.listener.names=CONTROLLER# 监听器安全协议映射
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT# Broker 间通信使用的监听器名称
inter.broker.listener.name=PLAINTEXT# 客户端连接地址（外部可访问的 host:port）
advertised.listeners=PLAINTEXT://node2:9092
EOF

node3上执行

cat > /usr/local/kafka/config/server.properties <<EOF
node.id=3# 节点角色：controller 和 broker
process.roles=controller,broker# 元数据日志存储目录（需提前创建并赋权）
metadata.log.dir=/tmp/kafka-logs# Controller 之间的投票节点列表（格式：id@host:port）
controller.quorum.voters=1@node1:9093,2@node2:9093,3@node3:9093# 监听器定义：名称、协议、host:port
# 注意：CONTROLLER 是用于 controller 间通信的监听器
#       PLAINTEXT 是用于 broker 对外服务的监听器
listeners=PLAINTEXT://:9092,CONTROLLER://:9093# 指定用于 controller 通信的监听器名称（必须是 listeners 中的一个）
controller.listener.names=CONTROLLER# 监听器安全协议映射
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT# Broker 间通信使用的监听器名称
inter.broker.listener.name=PLAINTEXT# 客户端连接地址（外部可访问的 host:port）
advertised.listeners=PLAINTEXT://node3:9092
EOF
4.1.5生成集群 ID

node1上执行

bin/kafka-storage.sh random-uuid
4.1.6格式化存储目录

3个节点均需要执行

cd /usr/local/kafka
bin/kafka-storage.sh format -t abcd-1234-efgh-5678 -c config/server.properties

执行结果

Formatting metadata directory /tmp/kafka-logs with metadata.version 4.0-IV3.

并且在 /tmp/kafka-logs/ 目录下会生成一个 meta.properties 文件，内容如下

cat /tmp/kafka-logs/meta.properties 
#
#Tue Aug 12 14:35:28 CST 2025
node.id=1
directory.id=WLMkRpvo2ZQFqvmHsGv2aA
version=1
cluster.id=XzHNVPgSTVKRBqsFWEC6AQ
4.1.7启动Kafka

3个节点上执行

nohup bin/kafka-server-start.sh config/server.properties &
4.1.8查看KRaft元数据集群的状态
bin/kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --status

5.验证 Kafka 集群

5.1创建一个测试 topic

在任意节点上执行

/usr/local/kafka/bin/kafka-topics.sh --create \--topic ceshi-topic \--bootstrap-server node1:9092 \--partitions 3 \--replication-factor 3
5.2查看 topic 列表

在任意节点上执行

/usr/local/kafka/bin/kafka-topics.sh --list \--bootstrap-server node1:9092
5.3启动生产者测试

node1上执行

/usr/local/kafka/bin/kafka-console-producer.sh \--bootstrap-server node1:9092 \--topic test-topic输入测试内容，如，Hello，Kafka
5.4启动消费者测试

node2上执行

/usr/local/kafka/bin/kafka-console-consumer.sh \--bootstrap-server node1:9092 \--topic test-topic \--from-beginning输出内容：Hello，Kafka
http://www.lryc.cn/news/618373.html

相关文章：

  • 工业相机镜头选型
  • 云计算核心技术
  • iPhone 17 Pro 为何被指像充电宝？
  • Stereolabs ZED相机 选型指南：双目 / 单目、短距 / 长距，如何为机器人视觉系统匹配最优方案？
  • 力扣11:盛水最多的容器
  • 深入C#异步编程基石：BeginInvoke与EndInvoke全解析
  • 使用ceph-deploy安装和配置RADOS Gateway (RGW)并使用S3访问集群
  • 串口超时参数深度解析：ReadTotalTimeoutMultiplier、ReadIntervalTimeout等
  • JVM宝典
  • 在IDEA中设置SQL解析作用域解决无法解析表的问题(详细图解)
  • Docker部署kafka实操+Java中访问
  • 【KO】大厂常见问题
  • damn the jvm again(2)
  • DDIA第五章：无主复制（去中心化复制）详解
  • 华为发布AI推理新技术，降低对HBM内存依赖
  • 阿里云国际DDoS高防：添加网站配置指南
  • LabVIEW菜单操控
  • 【USRP】基于LabVIEW的BPSK、QPSK，文本，图片
  • 区块链技术原理(7)-安全问题
  • C++少儿编程（二十二）—条件结构
  • 云原生作业(nginx)
  • QT（概述、基础函数、界面类、信号和槽）
  • Linux 服务器，安装mqtt服务
  • Linux 系统下 VS Code 降级至 1.85 版本教程：通过历史版本网站解决兼容性问题
  • 从零开始手搓一个GPT大语言模型：从理论到实践的完整指南（一）
  • Linux性能监控
  • Qt Charts 深度解析与实战指南
  • 全面解析MySQL(5)——“索引、事务、JDBC”三大核心
  • AI不再停留在概念阶段，而是在各行业核心业务场景产生实际价值。随着大模型、边缘计算等技术的突破，AI应用将向实时化、自主化、普惠化方向深度演进。
  • 高性能web服务器Tomcat