跳到主要内容

Kafka 集群模式部署

节点服务器 IP主机角色节点 ID
Node01192.168.10.7Kafka Broker、Controller1
Node02192.168.10.8Kafka Broker、Controller2
Node03192.168.10.9Kafka Broker、Controller3

安装 Kafka​

执行服务器：Node01、Node02、Node03（三台服务器均需执行）。

  1. 下载 JDK 安装包

    wget https://pdpublic.mingdao.com/private-deployment/offline/common/OpenJDK21U-jdk_x64_linux_hotspot_21.0.8_9.tar.gz

  2. 解压 JDK 并配置 Java 软链接

    tar -zxvf OpenJDK21U-jdk_x64_linux_hotspot_21.0.8_9.tar.gz -C /usr/local
    mv /usr/local/jdk-21.0.8+9 /usr/local/openjdk-21
    ln -s /usr/local/openjdk-21/bin/java /bin/java
    java -version

  3. 下载 Kafka 安装包

    wget https://pdpublic.mingdao.com/private-deployment/offline/common/kafka_2.13-4.3.1.tgz

  4. 安装 Kafka 并创建数据目录

    tar -zxvf kafka_2.13-4.3.1.tgz -C /usr/local
    mv /usr/local/kafka_2.13-4.3.1 /usr/local/kafka
    mkdir -p /data/kafka/kafka-logs

  5. 修改 Kafka JVM 内存为 4G

    sed -i ':a;N;$!ba;s/Xm[xs]1G/Xmx4G/1' /usr/local/kafka/bin/kafka-server-start.sh
    sed -i ':a;N;$!ba;s/Xm[xs]1G/Xms4G/1' /usr/local/kafka/bin/kafka-server-start.sh

  6. 创建 Kafka 运行账户

    useradd -M -s /sbin/nologin kafka

写入节点配置​

以下分别为 Node01、Node02、Node03 的 server.properties 配置。写入前，根据实际环境确认节点 IP 和端口，并注意修改以下参数：

  • node.id：Node01、Node02、Node03 分别使用 1、2、3，节点 ID 不可重复。

  • controller.quorum.bootstrap.servers：三台节点的配置必须完全一致，填写全部 3 个 Controller 节点的 IP 和 9093 端口。

  • advertised.listeners：填写当前节点的实际 IP 和 Broker 端口（9092），业务服务通过该地址访问 Kafka。

Node01​

执行服务器：Node01。

cat > /usr/local/kafka/config/server.properties <<'EOF'
process.roles=broker,controller
node.id=1
controller.quorum.bootstrap.servers=192.168.10.7:9093,192.168.10.8:9093,192.168.10.9:9093
listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
advertised.listeners=PLAINTEXT://192.168.10.7:9092
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
inter.broker.listener.name=PLAINTEXT
controller.listener.names=CONTROLLER
log.dirs=/data/kafka/kafka-logs
num.partitions=10
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
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
group.initial.rebalance.delay.ms=3000
message.max.bytes=10485760
replica.fetch.max.bytes=10485760
EOF

Node02​

执行服务器：Node02。

cat > /usr/local/kafka/config/server.properties <<'EOF'
process.roles=broker,controller
node.id=2
controller.quorum.bootstrap.servers=192.168.10.7:9093,192.168.10.8:9093,192.168.10.9:9093
listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
advertised.listeners=PLAINTEXT://192.168.10.8:9092
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
inter.broker.listener.name=PLAINTEXT
controller.listener.names=CONTROLLER
log.dirs=/data/kafka/kafka-logs
num.partitions=10
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
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
group.initial.rebalance.delay.ms=3000
message.max.bytes=10485760
replica.fetch.max.bytes=10485760
EOF

Node03​

执行服务器：Node03。

cat > /usr/local/kafka/config/server.properties <<'EOF'
process.roles=broker,controller
node.id=3
controller.quorum.bootstrap.servers=192.168.10.7:9093,192.168.10.8:9093,192.168.10.9:9093
listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
advertised.listeners=PLAINTEXT://192.168.10.9:9092
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
inter.broker.listener.name=PLAINTEXT
controller.listener.names=CONTROLLER
log.dirs=/data/kafka/kafka-logs
num.partitions=10
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
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
group.initial.rebalance.delay.ms=3000
message.max.bytes=10485760
replica.fetch.max.bytes=10485760
EOF

初始化并启动集群​

  1. 生成并保存集群 ID 和 Controller 目录 ID

    执行服务器：Node01。

    在 Node01 执行以下命令，生成 1 个集群 ID 和 3 个 Controller 目录 ID。记录命令输出的 4 个值，后续格式化三台节点时使用相同的值。

    KAFKA_CLUSTER_ID=$(/usr/local/kafka/bin/kafka-storage.sh random-uuid)
    CONTROLLER_1_UUID=$(/usr/local/kafka/bin/kafka-storage.sh random-uuid)
    CONTROLLER_2_UUID=$(/usr/local/kafka/bin/kafka-storage.sh random-uuid)
    CONTROLLER_3_UUID=$(/usr/local/kafka/bin/kafka-storage.sh random-uuid)
    echo "KAFKA_CLUSTER_ID=$KAFKA_CLUSTER_ID"
    echo "CONTROLLER_1_UUID=$CONTROLLER_1_UUID"
    echo "CONTROLLER_2_UUID=$CONTROLLER_2_UUID"
    echo "CONTROLLER_3_UUID=$CONTROLLER_3_UUID"

  2. 格式化三台节点的数据目录

    执行服务器：Node01、Node02、Node03（三台服务器均需执行）。

    在每台 Kafka 服务器执行以下命令，将变量替换为上一步记录的实际值。同时注意修改 --initial-controllers 参数值中的 IP。

    KAFKA_CLUSTER_ID=替换为实际集群ID
    CONTROLLER_1_UUID=替换为实际Controller目录ID
    CONTROLLER_2_UUID=替换为实际Controller目录ID
    CONTROLLER_3_UUID=替换为实际Controller目录ID
    /usr/local/kafka/bin/kafka-storage.sh format \
      --cluster-id "$KAFKA_CLUSTER_ID" \
      --initial-controllers "1@192.168.10.7:9093:${CONTROLLER_1_UUID},2@192.168.10.8:9093:${CONTROLLER_2_UUID},3@192.168.10.9:9093:${CONTROLLER_3_UUID}" \
      --config /usr/local/kafka/config/server.properties
    • 此操作仅用于首次初始化空数据目录。已有数据或已加入集群的节点严禁重复格式化，否则会清除本地元数据并导致节点无法正常加入集群。

  3. 授权 Kafka 目录

    执行服务器：Node01、Node02、Node03（三台服务器均需执行）。

    存储目录格式化完成后，在三台服务器上重新设置目录权限，确保 Kafka 服务可以读写程序目录和数据目录。

    chown -R kafka:kafka /usr/local/kafka /data/kafka

  4. 配置 systemd 服务

    执行服务器：Node01、Node02、Node03（三台服务器均需执行）。

    cat > /etc/systemd/system/kafka.service <<'EOF'
    [Unit]
    Description=Kafka KRaft
    After=network.target
    [Service]
    User=kafka
    Group=kafka
    LimitNOFILE=102400
    LimitNPROC=102400
    ExecStart=/usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties
    ExecStop=/usr/bin/kill $MAINPID
    Restart=on-failure
    [Install]
    WantedBy=multi-user.target
    EOF
    systemctl daemon-reload

  5. 启动 Kafka 集群

    执行服务器：Node01、Node02、Node03（三台服务器均需执行）。

    systemctl enable kafka
    systemctl start kafka
    systemctl status kafka

集群验证​

  1. 确认 Controller 仲裁集群状态

    在任意一节点执行以下命令，确认已选举出 Controller Leader，且三个节点均在仲裁集群中。

    /usr/local/kafka/bin/kafka-metadata-quorum.sh --bootstrap-server 127.0.0.1:9092 describe --status

  2. 创建并检查三副本 Topic

    在任意一个节点执行以下命令。

    /usr/local/kafka/bin/kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --create --if-not-exists --topic test01 --partitions 3 --replication-factor 3
    /usr/local/kafka/bin/kafka-topics.sh --bootstrap-server 127.0.0.1:9092 --describe --topic test01

    确认输出结果中每个分区的副本数为 3，且 Isr 列表包含 3 个节点 ID。

  3. 在 Node01 启动生产者

    /usr/local/kafka/bin/kafka-console-producer.sh --bootstrap-server 127.0.0.1:9092 --topic test01

  4. 在 Node02 或 Node03 启动消费者

    /usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic test01 --from-beginning

    如果消费者能够正常消费生产者发送的消息，则表示 Kafka 集群的消息收发验证通过。