4. kafka

官网：https://kafka.apache.org/

版本背景与生命周期定位

kafka4比3版本有更强的元数据优化，吞吐量和延迟优化也大幅提升，这里推荐使用4.0以上版本

4.1 简介

1. 核心定位

Apache Kafka 是一个分布式事件流平台，具备以下三大核心能力：

• 发布与订阅事件流（消息队列）

• 持久化存储事件流（日志系统）

• 实时处理事件流（流处理引擎）

2. 关键特性

3. 核心概念

• Broker：Kafka 服务节点

• Topic：消息逻辑分类

• Partition：Topic 的物理分片，有序不可变日志

• Replica：分区副本，含 Leader 与 Follower

• Producer: 消息生产者，可指定分区策略与 ACK 级别

• Consumer / Consumer Group: 消息消费者，组内消费者均衡消费不同 Partition。

• ISR（In-Sync Replica）：与 Leader 保持同步的副本集合

• Controller：集群元数据管理者（KRaft 模式下由 Controller Quorum 承担）

4.2 3节点高可用集群架构设计

1. 架构图（KRaft模式）

2. 高可用设计原则

• ✅ 3 节点满足多数派（Quorum）：容忍 1 节点故障（f = 1, 2f+1 = 3）

• ✅ Controller 与 Broker 共存：简化部署，减少资源开销

• ✅ 副本因子 ≥ 3：确保单点故障不影响数据可用性

• ✅ min.insync.replicas = 2：配合 acks=all 保证强一致性

• ❌ 禁用 unclean leader election：避免数据丢失（默认 false）

4.3 环境准备

1. 服务器规划

2. 基础依赖

• 操作系统：CentOS 7+/Ubuntu 20.04+

• Java：OpenJDK 11 或 17（禁止使用 JDK 8）

• 网络：节点间 9092（客户端）、19091（Controller）端口互通

• 用户：创建专用 kafka 用户，禁止 root 运行

3. 系统与内核优化

# 1. 系统参数优化（/etc/sysctl.conf）
cat >> /etc/sysctl.conf <<'EOF'
net.core.somaxconn = 32768
net.ipv4.tcp_max_syn_backlog = 65536
net.ipv4.tcp_tw_reuse = 1
net.ipv4.ip_local_port_range = 1024 65535
vm.swappiness = 1
vm.dirty_ratio = 80
vm.dirty_background_ratio = 5
vm.overcommit_memory = 1
EOF
sysctl -p

# 2. 文件描述符与进程数
cat >> /etc/security/limits.conf <<'EOF'
kafka soft nofile 1000000
kafka hard nofile 1000000
kafka soft nproc 65536
kafka hard nproc 65536
EOF

# 3. 禁用 Swap
swapoff -a && sed -i '/ swap / s/^\(.*\)$/#\1/g' /etc/fstab

4. 创建用户与目录

useradd -r -m -s /bin/bash kafka
mkdir -p /opt/kafka /data/kafka /data/kafka_meta
chown -R kafka:kafka /opt/kafka/* /data/kafka* 

5. 安装JDK 17+

此处省略.

[root@elk-91 ~]# java -version 
openjdk version "25.0.1" 2025-10-21
OpenJDK Runtime Environment (build 25.0.1+8-27)
OpenJDK 64-Bit Server VM (build 25.0.1+8-27, mixed mode, sharing)

4.4 KRaft模式集群搭建步骤

1. 下载与解压

[root@elk-91 ~]# wget https://mirrors.aliyun.com/apache/kafka/4.1.1/kafka_2.13-4.1.1.tgz
[root@elk-91 ~]# tar -xzf kafka_2.13-4.1.1.tgz -C /opt/
[root@elk-91 ~]# ln -sf /opt/kafka_2.13-4.1.1 /opt/kafka
[root@elk-91 ~]# echo 'export KAFKA_HOME=/opt/kafka' >> /etc/profile
[root@elk-91 ~]# echo 'export PATH=$KAFKA_HOME/bin:$PATH' >> /etc/profile
[root@elk-91 ~]# source /etc/profile

2. KRaft配置文件（节点差异化）

kafka1（10.0.0.91） /opt/kafka/config/kraft/server.properties

[root@elk-91 ~]# mkdir /opt/kafka/config/kraft
[root@elk-91 ~]# vi /opt/kafka/config/kraft/server.properties
[root@elk-91 ~]# cat /opt/kafka/config/kraft/server.properties
# ===== KRaft Identity =====
node.id=1
process.roles=broker,controller
controller.quorum.voters=1@10.0.0.91:9093,2@10.0.0.92:9093,3@10.0.0.93:9093

# ===== Listeners =====
listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
advertised.listeners=PLAINTEXT://10.0.0.91:9092
controller.listener.names=CONTROLLER
inter.broker.listener.name=PLAINTEXT

# ===== Storage =====
log.dirs=/data/kafka
# Kafka 4.x 建议独立元数据日志目录
metadata.log.dir=/data/kafka_meta

# ===== Replication & Partitions =====
num.partitions=6
default.replication.factor=3
min.insync.replicas=2
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2

# ===== Performance Tuning =====
num.network.threads=8
num.io.threads=16
socket.send.buffer.bytes=1024000
socket.receive.buffer.bytes=1024000
socket.request.max.bytes=104857600

# ===== Log Management =====
log.segment.bytes=1073741824
log.retention.hours=168
log.retention.bytes=107374182400
log.cleanup.policy=delete
log.roll.ms=3600000

# ===== High Availability =====
unclean.leader.election.enable=false
auto.create.topics.enable=false

kafka2（10.0.0.92）差异项

node.id=2
advertised.listeners=PLAINTEXT://10.0.0.92:9092

kafka3（10.0.0.93）差异项

node.id=3
advertised.listeners=PLAINTEXT://10.0.0.93:9092

3. 集群初始化与启动

a. 生成 Cluster ID（只需一次，任选一节点）

 [root@elk-91 ~]# sudo su - kafka
[kafka@elk-91 ~]$ $KAFKA_HOME/bin/kafka-storage.sh random-uuid
tH_XIOuNSta8NXbPZH9aEg

b. 格式化存储（所有节点执行，使用同一 Cluster ID）

[kafka@elk-91 ~]$ $KAFKA_HOME/bin/kafka-storage.sh format \
-t tH_XIOuNSta8NXbPZH9aEg \
-c $KAFKA_HOME/config/kraft/server.properties \
--ignore-formatted

c. 启动服务

# 使用 systemd 管理服务（推荐生产）
[root@elk-91 ~]# cat > /etc/systemd/system/kafka.service <<'EOF'
[Unit]
Description=Apache Kafka 4.x KRaft Server
After=network.target

[Service]
User=kafka
Group=kafka
Type=simple
Environment="JAVA_HOME=/usr/share/elasticsearch/jdk/"
Environment="PATH=$JAVA_HOME/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/opt/kafka/bin"
##注意这里生产环境配置的8G,可根据服务器配置来配
Environment="KAFKA_HEAP_OPTS=-Xmx8G -Xms8G"
Environment="KAFKA_JMX_OPTS=-Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.port=9999"
ExecStart=/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/kraft/server.properties
ExecStop=/opt/kafka/bin/kafka-server-stop.sh
Restart=on-failure
RestartSec=10
LimitNOFILE=1000000

[Install]
WantedBy=multi-user.target
EOF

[root@elk-91 ~]# systemctl daemon-reload
[root@elk-91 ~]# systemctl enable kafka
[root@elk-91 ~]# systemctl start kafka
[root@elk-91 ~]# systemctl status kafka

## 启动有问题，则查看日志
tail -200f  /opt/kafka/logs/server.log
journalctl -xe
#常见问题
1.权限问题
chown -R kafka:kafka /opt/kafka/* /data/kafka*
2.内存问题
修改启动脚本中的内存即可
3.数据目录问题
删除数据，重新格式化存储

4.5 集群验证与功能测试

1. 检查Quorum状态

[root@elk-91 ~]# kafka-metadata-quorum.sh --bootstrap-server 10.0.0.91:9092 describe --status
ClusterId:              tH_XIOuNSta8NXbPZH9aEg
LeaderId:               1
LeaderEpoch:            1
HighWatermark:          2317
MaxFollowerLag:         0
MaxFollowerLagTimeMs:   0
CurrentVoters:          [{"id": 1, "endpoints": ["CONTROLLER://10.0.0.91:9093"]}, {"id": 2, "endpoints": ["CONTROLLER://10.0.0.92:9093"]}, {"id": 3, "endpoints": ["CONTROLLER://10.0.0.93:9093"]}]
CurrentObservers:       []

• LeaderId正常（1/2/3）

• CurrentVoters: [1,2,3]

• MaxFollowerLag接近 0

2. 创建测试Topic

[root@elk-91 ~]# kafka-topics.sh --create --bootstrap-server 10.0.0.91:9092 --topic sre-test --partitions 6 --replication-factor 3 --config min.insync.replicas=2
Created topic sre-test.

3. 查看分区与副本分布

[root@elk-91 ~]# kafka-topics.sh --describe --bootstrap-server 10.0.0.91:9092 --topic sre-test
Topic: sre-test	TopicId: kKYqZawdRhqw1tuL0neDwg	PartitionCount: 6	ReplicationFactor: 3	Configs: min.insync.replicas=2,cleanup.policy=delete,segment.bytes=1073741824,unclean.leader.election.enable=false,retention.bytes=107374182400,segment.ms=3600000
	Topic: sre-test	Partition: 0	Leader: 2	Replicas: 2,3,1	Isr: 2,3,1	Elr: 	LastKnownElr: 
	Topic: sre-test	Partition: 1	Leader: 3	Replicas: 3,1,2	Isr: 3,1,2	Elr: 	LastKnownElr: 
	Topic: sre-test	Partition: 2	Leader: 1	Replicas: 1,2,3	Isr: 1,2,3	Elr: 	LastKnownElr: 
	Topic: sre-test	Partition: 3	Leader: 3	Replicas: 3,1,2	Isr: 3,1,2	Elr: 	LastKnownElr: 
	Topic: sre-test	Partition: 4	Leader: 1	Replicas: 1,2,3	Isr: 1,2,3	Elr: 	LastKnownElr: 
	Topic: sre-test	Partition: 5	Leader: 2	Replicas: 2,3,1	Isr: 2,3,1	Elr: 	LastKnownElr: 

每个 Partition 的 Leader/Follower 均匀分布在 3 节点，且 ISR 全齐。

4. 生产消费验证

# Producer
[root@elk-91 ~]# kafka-console-producer.sh --bootstrap-server 10.0.0.91:9092 --topic sre-test
>1111111111111111
>3333333333333333
>5555555555555555

# Consumer
[root@elk-92 ~]# kafka-console-consumer.sh --bootstrap-server 10.0.0.92:9092 --topic sre-test --from-beginning
1111111111111111
3333333333333333
5555555555555555

5. 容灾与回滚

容灾：单节点宕机 → Leader 自动切换至其他节点，RTO < 5s（Kafka 4.x 优化）。

回滚：保留 Kafka 4.x 二进制包与配置备份，如出现兼容性问题可降级至 3.7.x（需停写+数据快照恢复）。

备份：定期 kafka-log-dirs.sh导出分区状态，结合 Tiered Storage 沉降策略做冷备。

6. 根因分析与故障排查逻辑树