properties

每个 Broker 的唯一 ID，集群内不可重复

broker.id=0

监听客户端请求的地址和端口

格式为：协议://主机名:端口

listeners=PLAINTEXT://your.host.name:9092

对外通告的地址，如果存在代理或 NAT 必须正确设置

advertised.listeners=PLAINTEXT://公网或外部可达IP:9092

连接 Zookeeper 集群地址，用于元数据管理（Kafka 2.8+ 可免 ZooKeeper，使用 KRaft）

zookeeper.connect=zk1:2181,zk2:2181,zk3:2181/kafka

### 日志存储配置

Kafka 的数据以日志文件形式存放，以下参数直接影响磁盘占用与清理策略。

```properties
# 日志文件的存放路径，建议使用独立磁盘，避免与系统盘争抢 IO
log.dirs=/data/kafka-logs

# 每个 Topic 默认的分区数量 (可被创建 Topic 时覆盖)
num.partitions=3

# 默认副本因子，生产环境建议至少为 2
default.replication.factor=2

# 日志段大小，达到该大小即创建一个新 Segment，默认 1GB
log.segment.bytes=1073741824

# 日志保留策略：按时间保留（小时），超过即删除
log.retention.hours=168

# 按容量保留：每个分区日志总大小超过该值则清理旧段
log.retention.bytes=10737418240

# 日志清理策略：delete（直接删除）或 compact（压缩合并，保留每个 key 的最新值）
log.cleanup.policy=delete

集群内部协议配置

为保证高可用和一致性，需调整副本和 Leader 选举相关参数。

# 允许 ISR（In-Sync Replicas）的最小数量，生产者设置 acks=-1 时，数据必须写入该数量的副本才算成功
# 防止数据丢失，但会增加延迟
min.insync.replicas=2

# 是否允许非 ISR 副本被选举为 Leader
# 设为 false 可保证数据一致性，但可能降低可用性
unclean.leader.election.enable=false

# 副本从 Leader 拉取数据的线程数，CPU 密集时可适当增大
num.replica.fetchers=4

# Follower 向 Leader 发送 Fetch 请求的间隔，影响延迟，默认 1 秒
replica.fetch.wait.max.ms=1000
replica.fetch.min.bytes=1

网络与性能优化

# Socket 发送/接收缓冲区大小，大流量场景建议调大
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576

# 请求处理线程数
num.network.threads=8
num.io.threads=16

# Segment 写操作刷盘配置（通常交给 OS 管理，但可增加保护）
# 强制刷盘的间隔时间（毫秒）和消息条数，谨慎使用，影响性能
# log.flush.interval.ms=10000
# log.flush.interval.messages=100000

KRaft 模式配置（KIP-500）

若使用 KRaft 替代 ZooKeeper，需额外配置：

# 启用 KRaft 控制器模式
process.roles=broker,controller
# 控制器仲裁组，所有控制器节点列表
controller.quorum.voters=1@broker1:9093,2@broker2:9093,3@broker3:9093
# 此节点的控制器ID（需 voter 一致）
node.id=1
# 控制器监听地址
controller.listener.names=CONTROLLER
listeners=PLAINTEXT://:9092,CONTROLLER://:9093

客户端配置（生产者与消费者）

客户端配置通过 Properties 对象或配置文件注入。下面列出生产级别的关键参数。

生产者配置

# Kafka 集群地址，多个逗号分隔
bootstrap.servers=kafka1:9092,kafka2:9092

# 消息确认模式：0-不等确认；1-只等 Leader 确认；all 或 -1-等所有 ISR 确认
acks=all

# 重试次数，与 acks=all 配合使用，防止瞬时故障导致消息丢失
retries=3

# 消息发送的批处理设置，提高吞吐
batch.size=16384
linger.ms=10

# 缓冲区内存总量
buffer.memory=33554432

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

# 压缩算法，可选 none/gzip/snappy/lz4/zstd
compression.type=lz4

# 幂等性，保证单分区不重复消息，默认 true（Kafka 3.0+）
enable.idempotence=true

消费者配置

bootstrap.servers=kafka1:9092,kafka2:9092

# 消费者组 ID，相同组共享 offset 消费
group.id=my-consumer-group

# 是否自动提交偏移量（简易场景开启，生产环境推荐手动控制）
enable.auto.commit=false

# 自动偏移量重置策略：earliest（从头开始）/ latest（仅新消息）/ none（无偏移量时报错）
auto.offset.reset=earliest

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

# 单次拉取的最大记录数
max.poll.records=500

# 会话超时，消费者未在限定时间内心跳则被踢出并触发再均衡
session.timeout.ms=30000
heartbeat.interval.ms=10000

# 两次 poll 之间的最大允许间隔，超时后消费者被视作死亡
max.poll.interval.ms=300000

Topic 配置管理

通过 kafka-topics.sh 脚本或 AdminClient 修改 Topic 级别配置。

# 创建 Topic 指定分区和副本
kafka-topics.sh --create --topic my-topic --partitions 6 --replication-factor 3 \
  --config retention.ms=86400000 \
  --config max.message.bytes=2097152 \
  --bootstrap-server localhost:9092

# 修改已有 Topic 级别配置
kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics \
  --entity-name my-topic --alter --add-config cleanup.policy=compact

# 查看 Topic 当前配置
kafka-topics.sh --describe --topic my-topic --bootstrap-server localhost:9092