安装与集群搭建

环境要求

Kafka 运行在 JVM 之上，需要 Java 11 或更高版本（推荐 Java 17）。Kafka 3.x 在 KRaft 模式下不再需要 ZooKeeper。

# 检查 Java 版本（需要 11+）
java -version
# openjdk version "17.0.9" 2023-10-17

# macOS 安装 Java 17
brew install openjdk@17
export JAVA_HOME=$(/usr/libexec/java_home -v 17)

方式一：tar 包单机安装（macOS / Linux）

# 下载 Kafka 3.7.x（最新稳定版）
wget https://downloads.apache.org/kafka/3.7.0/kafka_2.13-3.7.0.tgz
tar -xzf kafka_2.13-3.7.0.tgz
cd kafka_2.13-3.7.0

# ── KRaft 模式启动（无需 ZooKeeper）──

# 1. 生成集群唯一 ID
KAFKA_CLUSTER_ID=$(bin/kafka-storage.sh random-uuid)
# 输出类似：MkU3OEVBNTcwNTJENDM2Qg==

# 2. 格式化存储目录
bin/kafka-storage.sh format \
  -t "$KAFKA_CLUSTER_ID" \
  -c config/kraft/server.properties

# 3. 启动 Kafka
bin/kafka-server-start.sh config/kraft/server.properties

# 验证启动成功（另开终端）
bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

方式二：Docker Compose（推荐开发环境）

以下配置使用 KRaft 模式（单节点 Kafka + Kafka UI 可视化管理），无需 ZooKeeper。

# docker-compose.yml
version: '3.8'

services:
  kafka:
    image: confluentinc/cp-kafka:7.6.0
    hostname: kafka
    container_name: kafka
    ports:
      - "9092:9092"
      - "9101:9101"  # JMX 监控端口
    environment:
      # KRaft 模式：此节点同时作为 Controller 和 Broker
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: 'broker,controller'
      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka:29093'
      # 监听器配置
      KAFKA_LISTENERS: 'PLAINTEXT://kafka:29092,CONTROLLER://kafka:29093,EXTERNAL://0.0.0.0:9092'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka:29092,EXTERNAL://localhost:9092'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT'
      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
      # 性能与存储
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
      KAFKA_LOG_RETENTION_HOURS: 168
      KAFKA_LOG_SEGMENT_BYTES: 1073741824
      KAFKA_JMX_PORT: 9101
      KAFKA_JMX_HOSTNAME: localhost
      # 集群 ID（使用 kafka-storage.sh random-uuid 生成）
      CLUSTER_ID: 'MkU3OEVBNTcwNTJENDM2Qg=='
    volumes:
      - kafka-data:/var/lib/kafka/data

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    container_name: kafka-ui
    ports:
      - "8080:8080"
    depends_on:
      - kafka
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092

volumes:
  kafka-data:

# 启动
docker compose up -d

# 查看日志
docker compose logs -f kafka

# 访问 Kafka UI
# 浏览器打开 http://localhost:8080

KRaft 模式 server.properties 关键参数

# config/kraft/server.properties 关键配置说明

# 节点角色：broker（处理客户端请求）或 controller（元数据管理）或两者
process.roles=broker,controller

# 本节点 ID，集群内唯一
node.id=1

# Controller Quorum 投票成员（KRaft 选举）
# 格式：node_id1@host1:port1,node_id2@host2:port2
controller.quorum.voters=1@localhost:9093

# 监听器：客户端连接地址
listeners=PLAINTEXT://:9092,CONTROLLER://:9093

# 对外广播地址（其他节点和客户端用来连接）
advertised.listeners=PLAINTEXT://your-hostname:9092

# 日志存储目录
log.dirs=/tmp/kraft-combined-logs

# 默认分区数
num.partitions=3

# 消息保留时间（毫秒）：7天
log.retention.ms=604800000

# 单个日志段文件最大大小
log.segment.bytes=1073741824  # 1GB

三节点集群配置

生产环境需要至少 3 个 Broker 以保证高可用（允许 1 个 Broker 故障）。以下是三节点 KRaft 集群的核心配置差异：

# Broker 1（server-1.properties）
node.id=1
controller.quorum.voters=1@broker1:9093,2@broker2:9093,3@broker3:9093
advertised.listeners=PLAINTEXT://broker1:9092
log.dirs=/data/kafka/node1

# Broker 2（server-2.properties）
node.id=2
controller.quorum.voters=1@broker1:9093,2@broker2:9093,3@broker3:9093
advertised.listeners=PLAINTEXT://broker2:9092
log.dirs=/data/kafka/node2

# Broker 3（server-3.properties）
node.id=3
controller.quorum.voters=1@broker1:9093,2@broker2:9093,3@broker3:9093
advertised.listeners=PLAINTEXT://broker3:9092
log.dirs=/data/kafka/node3

# 三节点共同格式化（相同 CLUSTER_ID）
bin/kafka-storage.sh format -t "$CLUSTER_ID" -c config/kraft/server-1.properties
bin/kafka-storage.sh format -t "$CLUSTER_ID" -c config/kraft/server-2.properties
bin/kafka-storage.sh format -t "$CLUSTER_ID" -c config/kraft/server-3.properties

Kafka UI（Provectus）使用指南

启动后访问 http://localhost:8080，Kafka UI 提供以下功能：

• Topics：查看所有 Topic 的分区数、副本状态、消息数量、保留大小
• Consumer Groups：查看各消费组的 Lag（消费延迟）和成员状态
• Brokers：集群节点信息、Controller 状态、配置参数
• Messages：浏览 Topic 内的消息（支持按 Offset/时间范围过滤）
• Schema Registry：（需配置）查看和管理 Avro/Protobuf Schema

常用管理脚本速查

# ── Topic 管理 ──

# 创建 Topic（3个分区，副本因子1）
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --create --topic orders \
  --partitions 3 --replication-factor 1

# 列出所有 Topic
bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

# 查看 Topic 详情（分区、副本、ISR）
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --describe --topic orders

# 修改分区数（只能增加）
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --alter --topic orders --partitions 6

# 删除 Topic
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --delete --topic orders

# ── 生产者测试 ──

# 命令行生产消息（每行一条消息）
bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic orders

# 带 Key 的生产者（key:value 格式，按 Tab 分隔）
bin/kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic orders \
  --property key.separator=: \
  --property parse.key=true

# ── 消费者测试 ──

# 从头开始消费
bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic orders \
  --from-beginning

# 指定消费组消费（记录 Offset）
bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic orders \
  --group my-consumer-group

# ── Consumer Group 管理 ──

# 列出所有消费组
bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 --list

# 查看消费组 Lag
bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --describe --group my-consumer-group

# 重置 Offset（从最早开始）
bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --group my-consumer-group \
  --topic orders \
  --reset-offsets --to-earliest --execute