Kafka Windows 安装启动与 YAML 配置全解析 - 实践

Kafka Windows 安装启动与 YAML 配置全解析

一、Kafka Windows 版安装与启动

1. 下载地址

Kafka 官方提供了适用于 Windows 的二进制包，下载地址：
Apache Kafka 官方下载
建议选择最新的稳定版本（如 kafka_2.13-3.6.1.tgz，其中 2.13 是 Scala 版本，3.6.1 是 Kafka 版本）

2. 安装步骤

1. 解压文件：使用解压工具（如 WinRAR、7-Zip）将下载的 tgz 文件解压到本地目录（建议路径无空格和中文），例如：D:\kafka_2.13-3.6.1

2. 配置环境变量（可选）：

   • 新增环境变量 KAFKA_HOME，值为 Kafka 解压路径（如 D:\kafka_2.13-3.6.1）
   • 在 Path 变量中添加 %KAFKA_HOME%\bin\windows

3. 启动服务

Kafka 依赖 Zookeeper，需要先启动 Zookeeper，再启动 Kafka 服务器。

启动 Zookeeper

打开命令提示符（CMD）或 PowerShell，执行以下命令：

# 切换到 Kafka 目录
cd D:\kafka_2.13-3.6.1
# 启动 Zookeeper（使用默认配置）
bin\windows\zookeeper-server-start.bat config\zookeeper.properties

启动 Kafka 服务器

打开另一个命令提示符窗口，执行：

# 切换到 Kafka 目录
cd D:\kafka_2.13-3.6.1
# 启动 Kafka 服务器（使用默认配置）
bin\windows\kafka-server-start.bat config\server.properties

创建测试主题

再打开一个命令提示符窗口，执行：

# 创建名为 test-topic 的主题
bin\windows\kafka-topics.bat --create --topic test-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
# 查看所有主题
bin\windows\kafka-topics.bat --list --bootstrap-server localhost:9092

停止服务

• 停止 Kafka 服务器：在启动窗口按 Ctrl+C
• 停止 Zookeeper：在启动窗口按 Ctrl+C

二、Spring Boot Kafka 完整 YAML 配置详解

以下是 application.yml 中所有可能用到的 Kafka 配置项，按功能分类说明：

spring:
kafka:
# 1. 基础配置
bootstrap-servers: localhost:9092  # Kafka 服务器地址，集群用逗号分隔（如：host1:9092,host2:9092）
client-id: my-kafka-client         # 客户端ID，用于标识请求来源，默认自动生成
properties:                        # 通用属性（适用于生产者、消费者、admin）
security.protocol: PLAINTEXT     # 安全协议（PLAINTEXT/SASL_PLAINTEXT/SASL_SSL/SSL）
sasl.mechanism: GSSAPI           # SASL 机制（如使用安全认证时配置）
ssl.truststore.location: classpath:kafka.truststore.jks  # SSL 信任库路径
# 2. 生产者配置（spring.kafka.producer）
producer:
# 序列化配置
key-serializer: org.apache.kafka.common.serialization.StringSerializer  # 键序列化器
value-serializer: org.springframework.kafka.support.serializer.JsonSerializer  # 值序列化器
# 消息确认机制
acks: 1  # 0:不等待确认；1:等待首领副本确认；all:等待所有同步副本确认（最可靠）
# 重试配置
retries: 3  # 发送失败重试次数
retry-backoff-ms: 1000  # 重试间隔（毫秒）
# 批量发送配置
batch-size: 16384  # 批量发送的大小阈值（字节），默认16KB
linger-ms: 0       # 批量发送的延迟时间（毫秒），默认0（立即发送）
buffer-memory: 33554432  # 发送缓冲区大小（字节），默认32MB
# 压缩配置
compression-type: none  # 消息压缩类型（none/gzip/snappy/lz4/zstd）
# 事务配置（启用事务时需要）
transaction-id-prefix: tx-  # 事务ID前缀，用于标识事务生产者
# 其他属性
max-block-ms: 60000  # 获取元数据或缓冲区满时的阻塞超时时间
max-request-size: 1048576  # 单个请求的最大大小（字节），默认1MB
# 3. 消费者配置（spring.kafka.consumer）
consumer:
# 反序列化配置
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer  # 键反序列化器
value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer  # 值反序列化器
# 基础配置
group-id: my-consumer-group  # 消费者组ID，必须配置
client-id: ${spring.kafka.client-id}-consumer  # 消费者客户端ID
# 偏移量配置
auto-offset-reset: earliest  # 无偏移量记录时的策略：earliest(从头读)/latest(从最新)/none(报错)
enable-auto-commit: false    # 是否自动提交偏移量，默认true
auto-commit-interval: 5000   # 自动提交间隔（毫秒），默认5000
# 拉取配置
fetch-min-size: 1  # 每次拉取的最小数据量（字节），默认1
fetch-max-wait-ms: 500  # 拉取等待超时时间（毫秒），默认500
max-poll-records: 500   # 每次poll()拉取的最大记录数，默认500
max-poll-interval-ms: 300000  # 两次poll之间的最大间隔（毫秒），默认5分钟
# 其他属性
heartbeat-interval-ms: 3000  # 消费者向协调器发送心跳的间隔，默认3000
session-timeout-ms: 10000    # 会话超时时间，默认10秒
properties:
spring.json.trusted.packages: "*"  # JSON反序列化信任的包，*表示所有
spring.json.use.type.headers: true  # 是否使用类型头信息进行反序列化
# 4. 监听器配置（spring.kafka.listener）
listener:
# 偏移量提交模式
ack-mode: manual_immediate  # 提交模式：
# - none:不提交
# - auto:自动提交（依赖enable-auto-commit）
# - manual:手动调用Acknowledge.acknowledge()
# - manual_immediate:手动调用后立即提交
# 并发配置
concurrency: 3  # 消费者并发数，建议不超过主题分区数
max-concurrency: 10  # 最大并发数
# 批量消费配置
batch-listener: false  # 是否启用批量消费，默认false
batch-error-handler: batchLoggingErrorHandler  # 批量消费错误处理器
# 重试配置
retry:
enabled: true  # 是否启用重试，默认false
max-attempts: 3  # 最大重试次数
backoff:
initial-interval: 1000  # 初始重试间隔（毫秒）
multiplier: 2.0         # 间隔乘数
max-interval: 10000     # 最大重试间隔（毫秒）
# 其他配置
poll-timeout: 3000  # poll()方法超时时间（毫秒）
idle-event-interval: 60000  # 空闲事件发布间隔（毫秒）
log-container-config: true  # 是否记录容器配置信息
# 5. Admin客户端配置（用于管理主题等）
admin:
client-id: ${spring.kafka.client-id}-admin  # Admin客户端ID
fail-fast: true  # 连接失败时是否快速失败，默认false
properties:
retry.backoff.ms: 1000  # Admin操作重试间隔
timeout: 30000  # 操作超时时间（毫秒）

三、配置注意事项

1. 版本兼容性：确保 spring-kafka 版本与 Kafka 服务器版本兼容（参考官方文档）

2. 序列化/反序列化：

   • 生产者和消费者的序列化器/反序列化器必须匹配
   • 使用 JSON 序列化时，需配置 spring.json.trusted.packages 避免安全限制

3. 偏移量提交：

   • 生产环境建议使用 manual_immediate 模式，确保消息处理完成后再提交偏移量
   • 自动提交（enable-auto-commit: true）可能导致消息未处理但偏移量已提交，造成数据丢失

4. 并发配置：

   • 消费者并发数（concurrency）不应超过主题的分区数，否则多余的消费者会空闲
   • 分区数决定了最大并行处理能力，合理规划分区数量

5. Windows 特有问题：

   • 路径分隔符使用 \ 而非 /
   • 启动脚本需使用 bin\windows 目录下的 .bat 文件
   • 若出现端口占用，可修改 config/server.properties 中的 port 配置（默认9092）

通过以上配置，你可以根据实际需求灵活调整 Kafka 在 Spring Boot 中的行为，平衡性能与可靠性。