Kubernetes--在k8s中安装和使用kafka

1. 前言

Apache Kafka 作为当今最流行的分布式流处理平台之一，已被广泛应用于实时数据管道、事件驱动架构和流分析等场景。随着云原生技术的普及，在 Kubernetes 上运行 Kafka 已成为主流选择。本文将详细介绍在Kubernetes 环境中部署和管理 Kafka 集群的方案。

2. 架构设计

2.1 核心组件

在 Kubernetes 中部署 Kafka 主要依赖以下核心组件：

• StatefulSet：确保每个 Kafka Broker 拥有稳定的网络标识和持久化存储
• Headless Service：为集群内部提供稳定的 DNS 发现机制
• 持久化存储：使用 StorageClass 动态提供持久卷
• KRaft 模式（推荐）：Kafka 3.3+ 版本支持的内置共识机制，替代 ZooKeeper

2.2 部署架构

┌─────────────────────────────────────────────────────────────┐
│ Kubernetes Cluster │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Kafka Pod 0 │ │ Kafka Pod 1 │ │ Kafka Pod N │ │
│ │ (broker-0) │ │ (broker-1) │ │ (broker-N) │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
│ │ │ │ │
│ ┌──────┴──────────────────┴──────────────────┴──────┐ │
│ │ Kafka Headless Service │ │
│ └───────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Persistent Volume Claims │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ StorageClass (动态供给) │ │
│ └─────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘

3. 环境准备

3.1 Kubernetes 集群要求

• Kubernetes 版本：1.25 或更高
• 存储类支持：确保至少有一个 StorageClass 可用
• 网络策略：支持 Pod 间通信
• 资源：建议至少 3 个节点，每个节点 4GB+ RAM

3.2 工具准备

# 1. 安装 kubectl
curl -LO "https://dl.k8s.io/release/$(curl -L -s https://dl.k8s.io/release/stable.txt)/bin/linux/amd64/kubectl"
chmod +x kubectl
sudo mv kubectl /usr/local/bin/

# 2. 安装 helm（可选，用于使用 Kafka Operator）
curl -fsSL -o get_helm.sh https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3
chmod 700 get_helm.sh
./get_helm.sh

4. 存储配置

4.1 创建 StorageClass

创建 storage-class.yaml：

apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
name: kafka–ssd
provisioner: kubernetes.io/gce–pd # 根据云提供商调整
parameters:
type: pd–ssd
reclaimPolicy: Retain
allowVolumeExpansion: true
volumeBindingMode: WaitForFirstConsumer

应用配置：

kubectl apply -f storage-class.yaml

4.2 本地存储示例（开发环境）

apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
name: local–storage
provisioner: kubernetes.io/no–provisioner
volumeBindingMode: WaitForFirstConsumer

5. Kafka 集群部署

5.1 使用 KRaft 模式（推荐）

5.1.1 创建配置 ConfigMap

创建 kafka-config.yaml：

apiVersion: v1
kind: ConfigMap
metadata:
name: kafka–config
labels:
app: kafka
data:
server.properties: |
# Kafka 服务器配置
process.roles=broker,controller
node.id=0
controller.quorum.voters=0@kafka-0.kafka-headless.default.svc.cluster.local:9093,1@kafka-1.kafka-headless.default.svc.cluster.local:9093,2@kafka-2.kafka-headless.default.svc.cluster.local:9093

# 监听器配置
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
advertised.listeners=PLAINTEXT://$(POD_NAME).kafka–headless.default.svc.cluster.local:9092
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
inter.broker.listener.name=PLAINTEXT

# 日志配置
log.dirs=/var/lib/kafka/data
num.partitions=3
default.replication.factor=3
min.insync.replicas=2

# 自动创建主题
auto.create.topics.enable=false

# 其他配置
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2

5.2.2 创建 Headless Service

创建 kafka-service.yaml：

apiVersion: v1
kind: Service
metadata:
name: kafka–headless
labels:
app: kafka
spec:
clusterIP: None
ports:
– name: client
port: 9092
targetPort: 9092
– name: controller
port: 9093
targetPort: 9093
selector:
app: kafka
publishNotReadyAddresses: true
—
apiVersion: v1
kind: Service
metadata:
name: kafka–external
labels:
app: kafka
spec:
type: LoadBalancer
ports:
– name: client
port: 9094
targetPort: 9092
selector:
app: kafka

5.2.3 创建 StatefulSet

创建 kafka-statefulset.yaml：

apiVersion: apps/v1
kind: StatefulSet
metadata:
name: kafka
labels:
app: kafka
spec:
serviceName: kafka–headless
replicas: 3
podManagementPolicy: Parallel
updateStrategy:
type: RollingUpdate
selector:
matchLabels:
app: kafka
template:
metadata:
labels:
app: kafka
spec:
terminationGracePeriodSeconds: 60
containers:
– name: kafka
image: apache/kafka:3.7.0
imagePullPolicy: IfNotPresent
ports:
– containerPort: 9092
name: client
– containerPort: 9093
name: controller
env:
– name: POD_NAME
valueFrom:
fieldRef:
fieldPath: metadata.name
– name: KAFKA_NODE_ID
valueFrom:
fieldRef:
fieldPath: metadata.annotations['kafka.node.id']
– name: KAFKA_CLUSTER_ID
value: "Lbr5W0rKTJCp5T6A6-QZOA"
command:
– /bin/bash
– –c
– |
# 从 Pod 名称提取节点 ID
NODE_ID=${HOSTNAME##*-}
echo "Node ID: ${NODE_ID}"

# 生成配置文件
cat /etc/kafka/server.properties.template | \\
sed "s/#node.id=/node.id=${NODE_ID}/" | \\
sed "s/#process.roles=/process.roles=broker,controller/" > \\
/etc/kafka/server.properties

# 启动 Kafka
exec /opt/kafka/bin/kafka–server–start.sh /etc/kafka/server.properties
volumeMounts:
– name: config
mountPath: /etc/kafka
– name: data
mountPath: /var/lib/kafka/data
– name: logs
mountPath: /opt/kafka/logs
resources:
requests:
memory: "2Gi"
cpu: "1000m"
limits:
memory: "4Gi"
cpu: "2000m"
readinessProbe:
tcpSocket:
port: 9092
initialDelaySeconds: 30
periodSeconds: 10
livenessProbe:
tcpSocket:
port: 9092
initialDelaySeconds: 60
periodSeconds: 20
volumes:
– name: config
configMap:
name: kafka–config
items:
– key: server.properties
path: server.properties.template
volumeClaimTemplates:
– metadata:
name: data
spec:
accessModes: [ "ReadWriteOnce" ]
storageClassName: kafka–ssd
resources:
requests:
storage: 100Gi

5.3 部署集群

# 1. 创建命名空间
kubectl create namespace kafka

# 2. 应用所有配置
kubectl apply -f kafka-config.yaml -n kafka
kubectl apply -f kafka-service.yaml -n kafka
kubectl apply -f kafka-statefulset.yaml -n kafka

# 3. 查看部署状态
kubectl get pods -n kafka -w
kubectl get statefulsets -n kafka
kubectl get pvc -n kafka

6. 验证与测试

6.1 验证集群状态

# 1. 检查 Pod 状态
kubectl get pods -n kafka -l app=kafka

# 2. 查看日志
kubectl logs -n kafka kafka-0 –tail=50

# 3. 进入 Pod 执行命令
kubectl exec -it -n kafka kafka-0 — /opt/kafka/bin/kafka-topics.sh \\
–bootstrap-server localhost:9092 \\
–list

6.2 创建测试主题

# 创建测试主题
kubectl exec -it -n kafka kafka-0 — /opt/kafka/bin/kafka-topics.sh \\
–bootstrap-server kafka-headless.kafka.svc.cluster.local:9092 \\
–create \\
–topic test-topic \\
–partitions 3 \\
–replication-factor 3

# 查看主题详情
kubectl exec -it -n kafka kafka-0 — /opt/kafka/bin/kafka-topics.sh \\
–bootstrap-server localhost:9092 \\
–describe \\
–topic test-topic

6.3 生产与消费消息

# 终端 1: 启动生产者
kubectl exec -it -n kafka kafka-0 — /opt/kafka/bin/kafka-console-producer.sh \\
–bootstrap-server localhost:9092 \\
–topic test-topic

# 终端 2: 启动消费者
kubectl exec -it -n kafka kafka-1 — /opt/kafka/bin/kafka-console-consumer.sh \\
–bootstrap-server localhost:9092 \\
–topic test-topic \\
–from-beginning

7. 监控与运维

7.1 集成 Prometheus 监控

创建 kafka-monitoring.yaml：

apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
name: kafka–monitor
namespace: monitoring
labels:
app: kafka
spec:
selector:
matchLabels:
app: kafka
endpoints:
– port: client
interval: 30s
path: /metrics
namespaceSelector:
matchNames:
– kafka

7.2 关键指标监控

apiVersion: v1
kind: ConfigMap
metadata:
name: kafka–alerts
namespace: monitoring
data:
kafka-alerts.yaml: |
groups:
– name: kafka
rules:
– alert: KafkaBrokerDown
expr: up{job="kafka"} == 0
for: 5m
labels:
severity: critical
annotations:
summary: "Kafka broker {{ $labels.pod }} is down"

– alert: KafkaUnderReplicatedPartitions
expr: kafka_cluster_partition_underreplicated > 0
for: 10m
labels:
severity: warning
annotations:
summary: "Kafka has under-replicated partitions"

– alert: KafkaOfflinePartitions
expr: kafka_controller_kafkacontroller_offlinepartitionscount > 0
for: 5m
labels:
severity: critical
annotations:
summary: "Kafka has offline partitions"

7.3 备份与恢复

# 备份主题数据
kubectl exec -it -n kafka kafka-0 — /opt/kafka/bin/kafka-topics.sh \\
–bootstrap-server localhost:9092 \\
–export \\
–topic test-topic \\
–output-file /tmp/test-topic-backup.json

# 从备份恢复
kubectl exec -it -n kafka kafka-0 — /opt/kafka/bin/kafka-topics.sh \\
–bootstrap-server localhost:9092 \\
–import \\
–topic test-topic \\
–input-file /tmp/test-topic-backup.json

8. 性能优化建议

8.1 资源配置优化

# 在 StatefulSet 中添加资源优化配置
resources:
requests:
memory: "4Gi"
cpu: "2000m"
limits:
memory: "8Gi"
cpu: "4000m"
env:
– name: KAFKA_HEAP_OPTS
value: "-Xmx6g -Xms6g"
– name: KAFKA_JVM_PERFORMANCE_OPTS
value: "-XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true"

8.2 存储优化

# 使用更高性能的存储类
storageClassName: premium–ssd
resources:
requests:
storage: 200Gi

8.3 网络优化

# 在 StatefulSet spec 中添加
spec:
template:
spec:
hostNetwork: false
dnsPolicy: ClusterFirst
dnsConfig:
options:
– name: ndots
value: "2"

9. 安全配置

9.1 启用 TLS 加密

# 在 server.properties 中添加
listeners=SSL://:9092,CONTROLLER://:9093
advertised.listeners=SSL://$(POD_NAME).kafka–headless.default.svc.cluster.local:9092
ssl.keystore.location=/etc/kafka/secrets/kafka.keystore.jks
ssl.keystore.password=changeit
ssl.key.password=changeit
ssl.truststore.location=/etc/kafka/secrets/kafka.truststore.jks
ssl.truststore.password=changeit
ssl.client.auth=required

9.2 启用 SASL 认证

# 在 server.properties 中添加
sasl.enabled.mechanisms=SCRAM–SHA–512
sasl.mechanism.inter.broker.protocol=SCRAM–SHA–512
security.inter.broker.protocol=SASL_SSL

10. 故障排除

10.1 常见问题及解决方案

10.2 调试命令

# 获取集群状态
kubectl exec -it -n kafka kafka-0 — /opt/kafka/bin/kafka-cluster.sh \\
–bootstrap-server localhost:9092 \\
–describe

# 检查控制器状态
kubectl exec -it -n kafka kafka-0 — /opt/kafka/bin/kafka-metadata-quorum.sh \\
–bootstrap-server localhost:9092 \\
–describe

11. 使用 Operator 的替代方案

对于生产环境，推荐使用 Kafka Operator 简化管理：

11.1 使用 Strimzi Operator

# 安装 Strimzi Operator
kubectl create namespace kafka
kubectl create -f 'https://strimzi.io/install/latest?namespace=kafka' -n kafka

# 部署 Kafka 集群
kubectl apply -f – <<EOF
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
name: my-cluster
namespace: kafka
spec:
kafka:
version: 3.7.0
replicas: 3
listeners:
– name: plain
port: 9092
type: internal
tls: false
– name: tls
port: 9093
type: internal
tls: true
config:
offsets.topic.replication.factor: 3
transaction.state.log.replication.factor: 3
transaction.state.log.min.isr: 2
default.replication.factor: 3
min.insync.replicas: 2
inter.broker.protocol.version: "3.7"
storage:
type: persistent-claim
size: 100Gi
class: kafka-ssd
entityOperator:
topicOperator: {}
userOperator: {}
EOF

12. 总结

本文详细介绍了在 Kubernetes 中部署 Apache Kafka 方案，重点强调了：

对于生产环境，建议：

• 使用 Kafka Operator（如 Strimzi）进行专业级管理
• 实施完整的监控告警体系
• 配置 TLS 和认证授权机制
• 定期备份重要数据

通过以上方案，可以在 Kubernetes 上构建稳定、高效且易于维护的 Kafka 集群，为实时数据处理提供可靠的基础设施支持。