Henry's xyz

外观

Kafka 集群部署 ​

本文记录一个生产可用的 Kafka KRaft 集群部署流程：3 节点 Controller + Broker 合一（Combined 模式），以及独立 Controller 和 Broker 的 Separated 模式对比。

架构规划 ​

Combined 模式（小集群推荐） ​

3 节点兼具 Controller 和 Broker 角色，适合中小规模集群：

     ┌───────────────────────┐  ┌───────────────────────┐  ┌───────────────────────┐
     │      Kafka Node 1     │  │      Kafka Node 2     │  │      Kafka Node 3     │
     │  controller + broker  │  │  controller + broker  │  │  controller + broker  │
     │         :9092         │  │         :9093         │  │         :9094         │
     │  CONTROLLER::9092     │  │  CONTROLLER::9093     │  │  CONTROLLER::9094     │
     └───────────────────────┘  └───────────────────────┘  └───────────────────────┘
              ▲  ▲  ▲                   ▲  ▲  ▲                   ▲  ▲  ▲
              │  │  │                   │  │  │                   │  │  │
         Partition-0 Leader      Partition-1 Leader       Partition-2 Leader
         (rf=3, ISR: 1,2,3)     (rf=3, ISR: 1,2,3)      (rf=3, ISR: 1,2,3)

     KRaft Quorum: Node 1 (Leader) + Node 2 + Node 3 → 元数据管理
     Broker Cluster: Node 1 + Node 2 + Node 3 → 消息存储

Separated 模式（大集群推荐） ​

Controller 和 Broker 分离部署，职责清晰，适合大规模集群：

   Controller Layer (KRaft Quorum)            Broker Layer (Data Plane)
   ┌──────────────┐ ┌──────────────┐ ┌──────────────┐
   │ Controller 1 │ │ Controller 2 │ │ Controller 3 │     元数据 + Leader 选举
   │   :9093      │ │   :9093      │ │   :9093      │
   └──────────────┘ └──────────────┘ └──────────────┘
          │               │               │
          └───────────────┼───────────────┘
                          │ 管理元数据
          ┌───────────────┼───────────────┐
          │               │               │
   ┌──────────────┐ ┌──────────────┐ ┌──────────────┐
   │   Broker 1   │ │   Broker 2   │ │   Broker 3   │     数据存储 + 客户端读写
   │   :9092      │ │   :9092      │ │   :9092      │
   └──────────────┘ └──────────────┘ └──────────────┘
对比Combined 模式Separated 模式
节点数3≥ 6（3 Controller + 3 Broker）
资源隔离无，Controller/Broker 共享 CPU/内存有，互不影响
故障隔离Controller 故障影响所在 Broker完全隔离
适用规模< 10 Broker，< 100K Partition> 10 Broker，> 100K Partition
滚动重启Broker 重启不影响 Controller 稳定性独立滚动，风险更低

生产环境绝大多数场景 Combined 模式足够。Partition 超过 10 万或物理机部署时再考虑 Separated。

节点角色 ​

角色说明
ControllerKRaft Quorum 成员，负责元数据管理（Topic/Partition 创建、Leader 选举、ISR 管理）
Broker数据面，负责消息存储和客户端请求处理
ObserverKRaft 中的只读副本，不参与选举（v3.7+）

Controller｜元数据管理 ​

Controller 负责管理集群所有元数据，通过 KRaft Raft 协议在 Quorum 之间同步：

元数据说明
Cluster metadataBroker 注册、Topic/Partition 信息、配置
Leader 选举检测 Broker 故障，从 ISR 中选举新 Leader
ISR 管理决定 Follower 是否在 ISR 中
Partition 分配Topic 创建和扩容时分配 Partition

KRaft Controller 将元数据存储在 __cluster_metadata Topic 中（而非外部的 ZooKeeper），数据与 Broker 使用相同的日志存储机制。

Broker｜数据存储 ​

Broker 负责实际的消息存储和客户端请求处理：

职责说明
消息存储接收 Producer 消息，写入本地日志文件
请求处理处理 Produce、Fetch、Commit 等请求
副本同步Leader 负责读写，Follower 从 Leader 同步数据
数据删除按 retention.ms / retention.bytes 清理过期数据

Partition Leader 分布 ​

Topic "orders" (3 Partitions, replication-factor=3)

      Node 1                    Node 2                    Node 3
┌──────────────────┐    ┌──────────────────┐    ┌──────────────────┐
│ P0 (Leader)      │    │ P1 (Leader)      │    │ P2 (Leader)      │
│ P1 (Follower)    │    │ P2 (Follower)    │    │ P0 (Follower)    │
│ P2 (Follower)    │    │ P0 (Follower)    │    │ P1 (Follower)    │
└──────────────────┘    └──────────────────┘    └──────────────────┘
    写入 P0 在这里            写入 P1 在这里            写入 P2 在这里

Controller 负责将不同 Partition 的 Leader 均匀分布到各 Broker，实现负载均衡。

组件端口规划 ​

组件端口用途数量
Broker Listener9092客户端（Producer/Consumer）连接≥ 3
Controller Listener9093KRaft Quorum 间通信3（Combined 模式与 Broker 共存）

部署前提 ​

网络连通性 ​

每个节点都需要能够访问所有其他节点的 Controller 端口和 Broker 端口。确保防火墙/安全组放行相关端口。

主机名 ​

生产环境使用 DNS 主机名而非 IP。所有节点的 advertised.listeners 应配置为客户端可达的地址。

环境准备 ​

shell
# 创建数据目录
sudo mkdir -p /data/kafka/{node1,node2,node3}
sudo chown -R $USER:$USER /data/kafka

# 创建日志目录
sudo mkdir -p /var/log/kafka
sudo chown -R $USER:$USER /var/log/kafka

# 确保 JDK 17+
java -version

Combined 模式部署（3 节点） ​

生成 Cluster UUID ​

shell
KAFKA_CLUSTER_ID=$(kafka-storage.sh random-uuid)
echo $KAFKA_CLUSTER_ID
# 示例: 7XOkR-LDRj6i0QVkYXJzLg

Node 1 配置 ​

properties
# /opt/kafka/config/kraft/node1.properties
process.roles=broker,controller
node.id=1
controller.quorum.voters=1@node1:9093,2@node2:9093,3@node3:9093

# Broker 监听
listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
advertised.listeners=PLAINTEXT://node1:9092
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
controller.listener.names=CONTROLLER

# 日志目录
log.dirs=/data/kafka/node1

# 集群默认配置
num.partitions=3
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

Node 2、Node 3 只需修改 node.id、advertised.listeners 和 log.dirs。

格式化并启动 ​

shell
# 每个节点执行
kafka-storage.sh format -t $KAFKA_CLUSTER_ID \
  -c /opt/kafka/config/kraft/node1.properties

# 启动
kafka-server-start.sh -daemon /opt/kafka/config/kraft/node1.properties
shell
# 检查 KRaft 元数据状态
kafka-metadata-quorum.sh --describe --status \
  --bootstrap-server node1:9092,node2:9092,node3:9092

关键配置解读 ​

配置推荐值说明
process.rolesbroker,controllerCombined 模式
controller.quorum.voters全部 Controller 节点列表格式: nodeId@host:port
default.replication.factor3默认 3 副本
min.insync.replicas2至少 2 个 ISR 才写入（配合 acks=all）
offsets.topic.replication.factor3Consumer Offset Topic 副本数
transaction.state.log.replication.factor3事务状态日志副本数
auto.create.topics.enablefalse禁止自动创建 Topic
num.network.threadsCPU 核数 × 0.5网络线程数
num.io.threadsCPU 核数 × 2I/O 线程数
log.retention.hours168（7 天）消息保留时间
log.segment.bytes1073741824（1GB）日志段大小
compression.typeproducer保留 Producer 指定的压缩格式

Docker Compose 部署 ​

yaml
version: '3.8'
services:
  kafka1:
    image: apache/kafka:4.0.0
    container_name: kafka1
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: "broker,controller"
      KAFKA_CONTROLLER_QUORUM_VOTERS: "1@kafka1:9093,2@kafka2:9093,3@kafka3:9093"
      KAFKA_LISTENERS: "PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://localhost:9092"
      KAFKA_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT"
      KAFKA_NUM_PARTITIONS: 3
      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
      KAFKA_MIN_INSYNC_REPLICAS: 2
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_CLUSTER_ID: "7XOkR-LDRj6i0QVkYXJzLg"
    volumes:
      - kafka1-data:/var/lib/kafka/data

  kafka2:
    image: apache/kafka:4.0.0
    container_name: kafka2
    ports:
      - "9093:9092"
    environment:
      KAFKA_NODE_ID: 2
      KAFKA_PROCESS_ROLES: "broker,controller"
      KAFKA_CONTROLLER_QUORUM_VOTERS: "1@kafka1:9093,2@kafka2:9093,3@kafka3:9093"
      KAFKA_LISTENERS: "PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://localhost:9093"
      KAFKA_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT"
      KAFKA_NUM_PARTITIONS: 3
      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
      KAFKA_MIN_INSYNC_REPLICAS: 2
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_CLUSTER_ID: "7XOkR-LDRj6i0QVkYXJzLg"
    volumes:
      - kafka2-data:/var/lib/kafka/data

  kafka3:
    image: apache/kafka:4.0.0
    container_name: kafka3
    ports:
      - "9094:9092"
    environment:
      KAFKA_NODE_ID: 3
      KAFKA_PROCESS_ROLES: "broker,controller"
      KAFKA_CONTROLLER_QUORUM_VOTERS: "1@kafka1:9093,2@kafka2:9093,3@kafka3:9093"
      KAFKA_LISTENERS: "PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://localhost:9094"
      KAFKA_CONTROLLER_LISTENER_NAMES: "CONTROLLER"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT"
      KAFKA_NUM_PARTITIONS: 3
      KAFKA_DEFAULT_REPLICATION_FACTOR: 3
      KAFKA_MIN_INSYNC_REPLICAS: 2
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_CLUSTER_ID: "7XOkR-LDRj6i0QVkYXJzLg"
    volumes:
      - kafka3-data:/var/lib/kafka/data

volumes:
  kafka1-data:
  kafka2-data:
  kafka3-data:

Docker Compose 适合本地开发和功能验证。生产环境推荐使用物理机/虚拟机或 Kubernetes StatefulSet。

Kubernetes 部署 ​

Kafka 在 K8s 上有两种方式：Operator（推荐）和手动 StatefulSet。

StatefulSet（手动） ​

yaml
apiVersion: v1
kind: Service
metadata:
  name: kafka-broker
  namespace: kafka
spec:
  clusterIP: None          # Headless Service — 必须，提供 Pod 直连 DNS
  selector:
    app: kafka
  ports:
    - name: plaintext
      port: 9092
    - name: controller
      port: 9093
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafka
  namespace: kafka
spec:
  serviceName: kafka-broker
  replicas: 3
  podManagementPolicy: Parallel
  selector:
    matchLabels:
      app: kafka
  template:
    metadata:
      labels:
        app: kafka
    spec:
      terminationGracePeriodSeconds: 30
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            - topologyKey: kubernetes.io/hostname
              labelSelector:
                matchLabels:
                  app: kafka
      containers:
        - name: kafka
          image: apache/kafka:4.0.0
          env:
            - name: KAFKA_NODE_ID
              valueFrom:
                fieldRef:
                  fieldPath: metadata.name  # 不可用于 ID（需为整数），用环境变量计算
            - name: KAFKA_PROCESS_ROLES
              value: "broker,controller"
            - name: KAFKA_CONTROLLER_QUORUM_VOTERS
              value: "1@kafka-0.kafka-broker.kafka:9093,2@kafka-1.kafka-broker.kafka:9093,3@kafka-2.kafka-broker.kafka:9093"
            - name: KAFKA_LISTENERS
              value: "PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093"
            - name: KAFKA_CONTROLLER_LISTENER_NAMES
              value: "CONTROLLER"
            - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
              value: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT"
          resources:
            requests:
              cpu: "2"
              memory: 4Gi
            limits:
              memory: 8Gi
          volumeMounts:
            - name: data
              mountPath: /var/lib/kafka/data
          livenessProbe:
            tcpSocket:
              port: 9092
            initialDelaySeconds: 30
            periodSeconds: 10
  volumeClaimTemplates:
    - metadata:
        name: data
      spec:
        accessModes: ["ReadWriteOnce"]
        storageClassName: ssd-high-iops
        resources:
          requests:
            storage: 100Gi

StatefulSet 的 Pod 名称为 kafka-0, kafka-1, kafka-2。需在启动前通过 InitContainer 将 Pod 序号写入 node.id 文件。

使用 InitContainer 写入 node.id ​

yaml
initContainers:
  - name: init-node-id
    image: busybox:1.36
    command:
      - sh
      - -c
      - |
        # 从 Pod 名称提取序号: kafka-0 → 1, kafka-1 → 2, kafka-2 → 3
        NODE_ID=$((${HOSTNAME##*-} + 1))
        mkdir -p /var/lib/kafka/data
        # 写入 KRaft meta.properties
        cat > /var/lib/kafka/data/meta.properties <<EOF
        node.id=$NODE_ID
        version=1
        cluster.id=7XOkR-LDRj6i0QVkYXJzLg
        EOF
    volumeMounts:
      - name: data
        mountPath: /var/lib/kafka/data

Pod 启动后验证：

shell
# Pod DNS：kafka-0.kafka-broker.kafka.svc.cluster.local
kubectl exec -n kafka kafka-0 -- kafka-metadata-quorum.sh \
  --describe --status -c /opt/kafka/config/kraft/server.properties

Strimzi Operator（推荐） ​

Strimzi 是 CNCF 项目，提供完整的 Kafka on K8s 声明式管理：

shell
# 安装 Operator
kubectl create namespace kafka
kubectl apply -f https://strimzi.io/install/latest?namespace=kafka -n kafka
yaml
# kafka-cluster.yaml
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-cluster
  namespace: kafka
spec:
  kafka:
    version: 4.0.0
    replicas: 3
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
    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
    storage:
      type: persistent-claim
      size: 100Gi
      class: ssd-high-iops
  entityOperator:
    topicOperator: {}
    userOperator: {}
shell
kubectl apply -f kafka-cluster.yaml -n kafka
# Strimzi 自动创建 StatefulSet、Service、ConfigMap，管理证书和滚动更新
Strimzi 特性说明
自动滚动更新配置变更后自动滚动重启 Broker
TLS/证书管理自动生成和管理内部 TLS 证书
Topic/User CRD通过 KafkaTopic、KafkaUser CRD 声明式管理
Kafka Connect内置 Connect 集群管理
监控集成内置 JMX Exporter + Prometheus 支持

K8s 部署要点 ​

要点说明
Headless ServiceclusterIP: None，提供 Pod 直连 DNS
podAntiAffinity确保 Broker 分散到不同物理节点
volumeClaimTemplates每个 Pod 独立 PVC
Parallel podManagementPolicy并行启动 Pod，加速初始化
InitContainer自动计算 node.id（基于 Pod 序号）
IOPS 要求Kafka 依赖顺序 I/O，使用 SSD + 高 IOPS 存储类
预热StatefulSet 首次启动较慢，建议先预热 PVC

验证集群 ​

shell
# 1. 查看集群元数据
kafka-metadata-quorum.sh --describe --status \
  --bootstrap-server node1:9092,node2:9092,node3:9092

# 2. 创建测试 Topic
kafka-topics.sh --create --topic test --partitions 6 --replication-factor 3 \
  --bootstrap-server node1:9092,node2:9092,node3:9092

# 3. 查看 Topic ISR 分布
kafka-topics.sh --describe --topic test \
  --bootstrap-server node1:9092,node2:9092,node3:9092

# 4. 生产/消费验证
echo "hello cluster" | kafka-console-producer.sh --topic test \
  --bootstrap-server node1:9092,node2:9092,node3:9092

kafka-console-consumer.sh --topic test --from-beginning --max-messages 1 \
  --bootstrap-server node1:9092,node2:9092,node3:9092

# 5. 查看 Broker 列表
kafka-broker-api-versions.sh --bootstrap-server node1:9092,node2:9092,node3:9092

监控 ​

JMX 指标导出 ​

shell
# 启动时附加 JMX Exporter
export KAFKA_OPTS="-javaagent:/opt/jmx_exporter/jmx_prometheus_javaagent.jar=9090:/opt/jmx_exporter/kafka.yml"
kafka-server-start.sh /opt/kafka/config/kraft/node1.properties

关键指标 ​

类别指标说明
Brokerkafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec消息流入速率
Brokerkafka.server:type=BrokerTopicMetrics,name=BytesInPerSec字节流入速率
Brokerkafka.network:type=RequestMetrics,name=TotalTimeMs,request=ProduceProduce 请求延迟
Replicakafka.server:type=ReplicaManager,name=UnderReplicatedPartitions未完全同步的分区数
Partitionkafka.server:type=ReplicaManager,name=PartitionCount总分区数
Controllerkafka.controller:type=ControllerStats,name=ActiveControllerCount活跃 Controller 数（应为 1）
OSkafka.server:type=OperatingSystem,name=FreePhysicalMemorySize空闲内存
JVMjava.lang:type=GarbageCollector,name=G1 Young GenerationGC 耗时

推荐监控栈 ​

方案适用
Prometheus + JMX Exporter + Grafana开源自建
Datadog Kafka Integration商业平台
Confluent Control CenterConfluent Platform
Strimzi + Prometheus OperatorK8s 部署

常见问题 ​

问题原因解决
Under-Replicated PartitionsFollower 不同步检查 replica.lag.time.max.ms，增加 num.replica.fetchers
Consumer Lag 持续增长Consumer 处理慢增加 Consumer 实例或 Partition 数
Controller 频繁切换网络不稳定或 GC 停顿检查网络延迟、调整 JVM 堆大小
磁盘写满retention.ms 过长或 retention.bytes 无限制调整保留策略，增加 Broker
Quorum 不可用Controller Quorum 节点数不足确保 ≥ 2 个节点存活（3 节点集群）
Topic 创建失败replication.factor > Broker 数减少副本因子或增加 Broker
OOM (Out of Memory)堆大小过低或分区太多调大 KAFKA_HEAP_OPTS=-Xmx4G，减少分区

生产检查清单 ​

  • [ ] 至少 3 个 Broker 节点（Combined 模式兼具 Controller 和 Broker）
  • [ ] default.replication.factor=3，min.insync.replicas=2
  • [ ] offsets.topic.replication.factor=3
  • [ ] transaction.state.log.replication.factor=3
  • [ ] auto.create.topics.enable=false
  • [ ] unclean.leader.election.enable=false
  • [ ] 所有节点使用 DNS 主机名，advertised.listeners 配置正确
  • [ ] 数据目录使用 SSD 磁盘，独立于系统盘
  • [ ] 操作系统 vm.swappiness=1，关闭 swap
  • [ ] JVM 堆内存 ≤ 32GB，预留 50% 物理内存给 Page Cache
  • [ ] Prometheus + Grafana 监控已接入
  • [ ] 网络延迟 < 10ms（Controller 间）
  • [ ] 制定 Partition 扩容和 Broker 扩容预案

安全加固 ​

启用认证与加密 ​

properties
# 每个 Broker 的 server.properties
# SASL/PLAIN 认证
listeners=SASL_PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
advertised.listeners=SASL_PLAINTEXT://node1:9092
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT
sasl.enabled.mechanisms=PLAIN

# Super User
super.users=User:admin

创建 JAAS 配置文件并设置 KAFKA_OPTS=-Djava.security.auth.login.config=/opt/kafka/config/kafka.jaas：

// kafka.jaas
KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-secret"
    user_admin="admin-secret"
    user_producer="producer-secret"
    user_consumer="consumer-secret";
};

ACL 授权 ​

shell
# 创建 ACL：允许 producer-user 对 orders Topic 写入
kafka-acls.sh --bootstrap-server node1:9092 \
  --add --allow-principal User:producer-user \
  --operation Write --topic orders \
  --command-config admin-client.conf

# 允许 consumer-group "order-service" 读取
kafka-acls.sh --bootstrap-server node1:9092 \
  --add --allow-principal User:consumer-user \
  --operation Read --topic orders --group order-service \
  --command-config admin-client.conf

参考 ​