Kafka 3.9 教程（三）：配置详解

📌 本文是「Kafka 3.9 教程」系列第 三 篇 · 系列目录

本文档基于 Apache Kafka 3.9 官方文档整理

目录

• 3.1 Broker 配置
• 3.2 主题级配置
• 3.3 生产者配置
• 3.4 消费者配置
• 3.5 Kafka Connect 配置
• 3.6 Kafka Streams 配置

3.1 Broker 配置

Broker 是 Kafka 集群中的服务节点，负责接收、存储和转发消息。

3.1.1 基础配置

# ========== 基础配置 ==========
# Broker ID，每个 Broker 唯一
broker.id=0

# 监听器配置
listeners=PLAINTEXT://0.0.0.0:9092

# 对外公告的监听器地址（客户端连接使用）
advertised.listeners=PLAINTEXT://localhost:9092

# 消息协议与监听器映射
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_SSL:SASL_SSL

# ========== 日志存储配置 ==========
# 日志目录（可配置多个，用逗号分隔）
log.dirs=/tmp/kafka-logs

# 默认分区数
num.partitions=1

# 默认副本因子
default.replication.factor=1

# 每个日志段的大小（字节）
log.segment.bytes=1073741824  # 1GB

# 日志段滚动时间（毫秒）
log.roll.ms=604800000  # 7天

3.1.2 网络配置

# ========== 网络配置 ==========
# 最大连接数
max.connections=2147483647

# 每个 IP 的最大连接数
max.connections.per.ip=2147483647

# 套接字发送缓冲区
socket.send.buffer.bytes=102400  # 100KB

# 套接字接收缓冲区
socket.receive.buffer.bytes=102400  # 100KB

# 请求最大大小
socket.request.max.bytes=104857600  # 100MB

# 连接空闲超时时间
connections.max.idle.ms=600000  # 10分钟

# 最小拉取数据量（字节）
fetch.min.bytes=1

# 等待拉取的最大时间（毫秒）
fetch.max.wait.ms=500

3.1.3 数据保留配置

# ========== 数据保留配置 ==========
# 消息保留时间（毫秒）
log.retention.hours=168  # 7天

# 消息保留大小（字节）
log.retention.bytes=-1  # 无限制

# 最小可删除日志段大小（字节）
log.retention.check.interval.ms=300000  # 5分钟

# 清理策略
log.cleanup.policy=delete

# 压缩比率阈值
log.cleaner.min.compaction.ratio=0.5

3.1.4 ZooKeeper 配置（传统模式）

# ========== ZooKeeper 配置 ==========
zookeeper.connect=localhost:2181/kafka

# ZooKeeper 连接超时
zookeeper.connection.timeout.ms=18000

# ZooKeeper 会话超时
zookeeper.session.timeout.ms=6000

# 是否启用 ZooKeeper ACL
zookeeper.set.acl=false

3.1.5 KRaft 配置（新模式）

# ========== KRaft 配置 ==========
# 进程角色（broker, controller）
process.roles=broker,controller

# 节点 ID
node.id=1

# Controller 投票者列表
controller.quorum.voters=1@localhost:9093

# Controller 监听器
controller.listener.names=CONTROLLER

# 元数据日志副本数
metadata.log.replication.factor=3

3.1.6 完整 Broker 配置示例

# server.properties 完整示例

# 基础配置
broker.id=0
listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
advertised.listeners=PLAINTEXT://localhost:9092
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
process.roles=broker,controller
node.id=1
controller.quorum.voters=1@localhost:9093
controller.listener.names=CONTROLLER

# 日志配置
log.dirs=/data/kafka-logs
num.partitions=3
default.replication.factor=3
log.segment.bytes=1073741824
log.retention.hours=168
log.retention.check.interval.ms=300000

# 性能配置
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600

# 副本配置
min.insync.replicas=2
unclean.leader.election.enable=false

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

# 删除主题配置
delete.topic.enable=true

3.2 主题级配置

主题配置可以覆盖 Broker 默认值，精细控制每个主题的行为。

3.2.1 创建主题时配置

# 创建主题并指定配置
kafka-topics.sh --create \
  --topic my-topic \
  --bootstrap-server localhost:9092 \
  --partitions 6 \
  --replication-factor 3 \
  --config cleanup.policy=delete \
  --config retention.ms=604800000 \
  --config compression.type=lz4

3.2.2 修改主题配置

# 修改主题配置
kafka-configs.sh --alter \
  --topic my-topic \
  --add-config retention.ms=259200000,cleanup.policy=compact \
  --bootstrap-server localhost:9092

# 查看主题配置
kafka-configs.sh --describe \
  --topic my-topic \
  --bootstrap-server localhost:9092

# 删除主题配置覆盖
kafka-configs.sh --alter \
  --topic my-topic \
  --delete-config retention.ms \
  --bootstrap-server localhost:9092

3.2.3 完整配置项详解

数据保留配置

日志段配置

压缩配置

消息配置

刷盘配置

副本配置

分层存储配置

3.2.4 分区配置

# 增加分区数（注意：分区数只能增加，不能减少）
kafka-topics.sh --alter \
  --topic my-topic \
  --partitions 12 \
  --bootstrap-server localhost:9092

# 查看分区分配
kafka-topics.sh --describe \
  --topic my-topic \
  --bootstrap-server localhost:9092

3.2.5 分区分配策略

# Broker 默认分区分配策略
partition.assignment.strategy=org.apache.kafka.clients.consumer.RangeAssignor

# 可用策略：
# - org.apache.kafka.clients.consumer.RangeAssignor
# - org.apache.kafka.clients.consumer.RoundRobinAssignor
# - org.apache.kafka.clients.consumer.StickyAssignor

3.3 生产者配置

3.3.1 必填配置

# ========== 必填配置 ==========
# Broker 列表
bootstrap.servers=localhost:9092

# 键序列化器
key.serializer=org.apache.kafka.common.serialization.StringSerializer

# 值序列化器
value.serializer=org.apache.kafka.common.serialization.StringSerializer

3.3.2 可靠性配置

# ========== 可靠性配置 ==========
# 确认机制
acks=all

# 重试次数
retries=2147483647

# 重试间隔
retry.backoff.ms=100

# 幂等性（Kafka 0.11.0+）
enable.idempotence=true

# 事务 ID（幂等/事务需要）
transactional.id=my-transactional-id

# 事务超时
transaction.timeout.ms=60000

3.3.3 性能优化配置

# ========== 性能配置 ==========
# 批量大小（字节）
batch.size=16384  # 16KB

# 发送延迟（毫秒），0 表示立即发送
linger.ms=0

# 缓冲区大小（字节）
buffer.memory=33554432  # 32MB

# 缓冲区满时的阻塞时间
max.block.ms=5000

# 单次请求最大大小
max.request.size=1048576  # 1MB

# 压缩类型
compression.type=lz4  # none, gzip, snappy, lz4, zstd

3.3.4 连接配置

# ========== 连接配置 ==========
# 连接超时
connections.max.idle.ms=540000  # 9分钟

# 请求超时
request.timeout.ms=30000

# 重试次数
retries=3

# 重试间隔
retry.backoff.ms=100

# 元数据刷新间隔
metadata.max.age.ms=300000  # 5分钟

3.3.5 完整生产者配置示例

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class ProducerConfigExample {
    public static Properties getConfig() {
        Properties props = new Properties();

        // 基础配置
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 可靠性配置
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

        // 性能配置
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);  // 16KB
        props.put(ProducerConfig.LINGER_MS_CONFIG, 5);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432L);  // 32MB
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");

        // 连接配置
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);

        return props;
    }
}

3.3.6 生产者配置项详解

核心配置（高优先级）

批次与性能配置（中优先级）

网络与安全配置（中优先级）

可靠性与事务配置（低优先级）

3.3.7 生产者配置最佳实践

3.4 消费者配置

3.4.1 必填配置

# ========== 必填配置 ==========
# Broker 列表
bootstrap.servers=localhost:9092

# 消费者组 ID
group.id=my-consumer-group

# 键反序列化器
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer

# 值反序列化器
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

3.4.2 偏移量管理配置

# ========== 偏移量管理 ==========
# 无初始偏移量时的策略
auto.offset.reset=earliest  # earliest, latest, none

# 是否自动提交偏移量
enable.auto.commit=true

# 自动提交间隔
auto.commit.interval.ms=5000

# 是否允许消费者手动分配分区（禁用消费者组）
partition.assignment.strategy=org.apache.kafka.clients.consumer.RangeAssignor

3.4.3 消费控制配置

# ========== 消费控制 ==========
# 每次拉取最大记录数
max.poll.records=500

# 拉取超时时间
fetch.min.bytes=1
fetch.max.wait.ms=500

# 最大拉取大小（字节）
max.partition.fetch.bytes=1048576  # 1MB

# 会话超时
session.timeout.ms=10000

# 心跳间隔
heartbeat.interval.ms=3000

# 消费者最长空闲时间
max.poll.interval.ms=300000  # 5分钟

3.4.4 完整消费者配置示例

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Properties;

public class ConsumerConfigExample {
    public static Properties getConfig() {
        Properties props = new Properties();

        // 基础配置
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // 偏移量管理
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);

        // 消费控制
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1);
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);

        // 会话管理
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10000);
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 3000);

        return props;
    }
}

3.4.5 消费者配置项详解

核心配置（高优先级）

偏移量与消费控制配置（中优先级）

网络与安全配置（中优先级）

低优先级配置

分区分配策略详解

3.4.6 消费者配置最佳实践

3.5 Kafka Connect 配置

3.5.1 通用配置

# ========== 通用配置 ==========
# Broker 列表
bootstrap.servers=localhost:9092

# 集群名称
group.id=connect-cluster

# 任务数
tasks.max=1

# 键转换器
key.converter=org.apache.kafka.connect.json.JsonConverter

# 值转换器
value.converter=org.apache.kafka.connect.json.JsonConverter

# 转换器配置
key.converter.schemas.enable=true
value.converter.schemas.enable=true

# 偏移量存储
offset.storage.topic=connect-offsets
offset.storage.replication.factor=3

# 配置存储
config.storage.topic=connect-configs
config.storage.replication.factor=3

# 状态存储
status.storage.topic=connect-status
status.storage.replication.factor=3

3.5.2 独立模式配置

# connect-standalone.properties

# Worker 配置
bootstrap.servers=localhost:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
offset.storage.file.filename=/tmp/connect.offsets

# 插件路径
plugin.path=/opt/kafka/plugins

3.5.3 分布式模式配置

# connect-distributed.properties

# Worker 配置
bootstrap.servers=localhost:9092
group.id=connect-cluster
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter

# 内部主题配置
offset.storage.topic=connect-offsets
offset.storage.replication.factor=3
config.storage.topic=connect-configs
config.storage.replication.factor=3
status.storage.topic=connect-status
status.storage.replication.factor=3

# REST API 配置
rest.port=8083
rest.advertised.port=8083
rest.advertised.host.name=connect-host

3.6 Kafka Streams 配置

3.6.1 应用配置

# ========== 应用配置 ==========
# 应用 ID（必须唯一）
application.id=my-streams-app

# Broker 列表
bootstrap.servers=localhost:9092

# 状态目录
state.dir=/tmp/kafka-streams

3.6.2 处理配置

# ========== 处理配置 ==========
# 处理保证
processing.guarantee=exactly_once_v2  # at_least_once, exactly_once, exactly_once_v2

# 提交间隔
commit.interval.ms=1000

# 拓扑超时
topology.max.executors=100
topology.optimization=all

3.6.3 并行度配置

# ========== 并行度配置 ==========
# Streams 线程数
num.stream.threads=1

# 任务超时
task.timeout.ms=300000  # 5分钟

# 分区缓冲区大小
buffered.records.per.partition=1000

3.6.4 内存配置

# ========== 内存配置 ==========
# 最大缓存大小
cache.max.bytes.buffering=10485760  # 10MB

# 缓存刷新阈值
cache.flush.percent=0.2

# RocksDB 配置
rocksdb.config.setter=com.example.MyRocksDBConfig

3.6.5 完整 Streams 配置示例

import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.common.serialization.Serdes;

import java.util.Properties;

public class StreamsConfigExample {
    public static Properties getConfig() {
        Properties props = new Properties();

        // 基础配置
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        // 序列化
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        // 处理保证
        props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
                  StreamsConfig.EXACTLY_ONCE_V2);

        // 提交间隔
        props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);

        // 状态目录
        props.put(StreamsConfig.STATE_DIR_CONFIG, "/tmp/kafka-streams");

        return props;
    }
}

附录：配置优先级

配置生效优先级（从高到低）：

1. 动态主题配置（kafka-configs.sh）
2. 主题创建时的 --config 参数
3. Broker 默认配置（server.properties）
4. 生产者/消费者代码配置

下一步学习

• 第四章：实现原理 - 深入理解 Kafka 内部机制
• 第五章：运维指南 - 集群运维最佳实践
• 第六章：安全认证 - 安全配置详解